Revision bbdc4c3fd60c1febcb663f3b6cbfd90febeb23a2 authored by Andrey Zhavoronkov on 29 June 2023, 11:23:12 UTC, committed by Andrey Zhavoronkov on 29 June 2023, 11:23:12 UTC
1 parent 75c1e6f
test_client.py
# Copyright (C) 2022 CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT
import io
from contextlib import ExitStack
from logging import Logger
from typing import List, Tuple
import packaging.version as pv
import pytest
from cvat_sdk import Client, models
from cvat_sdk.core.client import Config, make_client
from cvat_sdk.core.exceptions import IncompatibleVersionException, InvalidHostException
from cvat_sdk.exceptions import ApiException
from shared.utils.config import BASE_URL, USER_PASS
class TestClientUsecases:
@pytest.fixture(autouse=True)
def setup(
self,
restore_db_per_function, # force fixture call order to allow DB setup
fxt_logger: Tuple[Logger, io.StringIO],
fxt_client: Client,
fxt_stdout: io.StringIO,
admin_user: str,
):
_, self.logger_stream = fxt_logger
self.client = fxt_client
self.stdout = fxt_stdout
self.user = admin_user
yield
def test_can_login_with_basic_auth(self):
self.client.login((self.user, USER_PASS))
assert self.client.has_credentials()
def test_can_fail_to_login_with_basic_auth(self):
with pytest.raises(ApiException):
self.client.login((self.user, USER_PASS + "123"))
def test_can_logout(self):
self.client.login((self.user, USER_PASS))
self.client.logout()
assert not self.client.has_credentials()
def test_can_get_server_version(self):
self.client.login((self.user, USER_PASS))
version = self.client.get_server_version()
assert (version.major, version.minor) >= (2, 0)
def test_can_strip_trailing_slash_in_hostname_in_make_client(admin_user: str):
host, port = BASE_URL.split("://", maxsplit=1)[1].rsplit(":", maxsplit=1)
with make_client(host=host + "/", port=port, credentials=(admin_user, USER_PASS)) as client:
assert client.api_map.host == BASE_URL
def test_can_strip_trailing_slash_in_hostname_in_client_ctor(admin_user: str):
with Client(url=BASE_URL + "/") as client:
client.login((admin_user, USER_PASS))
assert client.api_map.host == BASE_URL
def test_can_detect_server_schema_if_not_provided():
host, port = BASE_URL.split("://", maxsplit=1)[1].rsplit(":", maxsplit=1)
client = make_client(host=host, port=int(port))
assert client.api_map.host == "http://" + host + ":" + port
def test_can_fail_to_detect_server_schema_if_not_provided():
host, port = BASE_URL.split("://", maxsplit=1)[1].rsplit(":", maxsplit=1)
with pytest.raises(InvalidHostException) as capture:
make_client(host=host, port=int(port) + 1)
assert capture.match(r"Failed to detect host schema automatically")
def test_can_reject_invalid_server_schema():
host, port = BASE_URL.split("://", maxsplit=1)[1].rsplit(":", maxsplit=1)
with pytest.raises(InvalidHostException) as capture:
make_client(host="ftp://" + host, port=int(port) + 1)
assert capture.match(r"Invalid url schema 'ftp'")
@pytest.mark.parametrize("raise_exception", (True, False))
def test_can_warn_on_mismatching_server_version(
fxt_logger: Tuple[Logger, io.StringIO], monkeypatch, raise_exception: bool
):
logger, logger_stream = fxt_logger
def mocked_version(_):
return pv.Version("0")
monkeypatch.setattr(Client, "get_server_version", mocked_version)
config = Config()
with ExitStack() as es:
if raise_exception:
config.allow_unsupported_server = False
es.enter_context(pytest.raises(IncompatibleVersionException))
Client(url=BASE_URL, logger=logger, config=config)
assert "Server version '0' is not compatible with SDK version" in logger_stream.getvalue()
@pytest.mark.parametrize("do_check", (True, False))
def test_can_check_server_version_in_ctor(
fxt_logger: Tuple[Logger, io.StringIO], monkeypatch, do_check: bool
):
logger, logger_stream = fxt_logger
def mocked_version(_):
return pv.Version("0")
monkeypatch.setattr(Client, "get_server_version", mocked_version)
config = Config()
config.allow_unsupported_server = False
with ExitStack() as es:
if do_check:
es.enter_context(pytest.raises(IncompatibleVersionException))
Client(url=BASE_URL, logger=logger, config=config, check_server_version=do_check)
assert (
"Server version '0' is not compatible with SDK version" in logger_stream.getvalue()
) == do_check
def test_can_check_server_version_in_method(fxt_logger: Tuple[Logger, io.StringIO], monkeypatch):
logger, logger_stream = fxt_logger
def mocked_version(_):
return pv.Version("0")
monkeypatch.setattr(Client, "get_server_version", mocked_version)
config = Config()
config.allow_unsupported_server = False
client = Client(url=BASE_URL, logger=logger, config=config, check_server_version=False)
with client, pytest.raises(IncompatibleVersionException):
client.check_server_version()
assert "Server version '0' is not compatible with SDK version" in logger_stream.getvalue()
@pytest.mark.parametrize(
"server_version, supported_versions, expect_supported",
[
# Currently, it is ~=, as defined in https://peps.python.org/pep-0440/
("3.2", ["2.0"], False),
("2", ["2.1"], False),
("2.1", ["2.1"], True),
("2.1a", ["2.1"], False),
("2.1.post1", ["2.1"], True),
("2.1", ["2.1.pre1"], True),
("2.1.1", ["2.1"], True),
("2.2", ["2.1"], False),
("2.2", ["2.1.0", "2.3"], False),
("2.2", ["2.1", "2.2", "2.3"], True),
("2.2.post1", ["2.1", "2.2", "2.3"], True),
("2.2.pre1", ["2.1", "2.2", "2.3"], False),
("2.2", ["2.3"], False),
("2.1.0.dev123", ["2.1.post2"], False),
("1!1.3", ["2.1"], False),
("1!1.3.1", ["2.1", "1!1.3"], True),
("1!1.1.dev12", ["1!1.1"], False),
],
)
def test_can_check_server_version_compatibility(
fxt_logger: Tuple[Logger, io.StringIO],
monkeypatch: pytest.MonkeyPatch,
server_version: str,
supported_versions: List[str],
expect_supported: bool,
):
logger, _ = fxt_logger
monkeypatch.setattr(Client, "get_server_version", lambda _: pv.Version(server_version))
monkeypatch.setattr(
Client, "SUPPORTED_SERVER_VERSIONS", [pv.Version(v) for v in supported_versions]
)
config = Config(allow_unsupported_server=False)
with ExitStack() as es:
if not expect_supported:
es.enter_context(pytest.raises(IncompatibleVersionException))
Client(url=BASE_URL, logger=logger, config=config, check_server_version=True)
@pytest.mark.parametrize("verify", [True, False])
def test_can_control_ssl_verification_with_config(verify: bool):
config = Config(verify_ssl=verify)
client = Client(BASE_URL, config=config)
assert client.api_client.configuration.verify_ssl == verify
def test_organization_contexts(admin_user: str):
with make_client(BASE_URL, credentials=(admin_user, USER_PASS)) as client:
assert client.organization_slug is None
org = client.organizations.create(models.OrganizationWriteRequest(slug="testorg"))
# create a project in the personal workspace
client.organization_slug = ""
personal_project = client.projects.create(models.ProjectWriteRequest(name="Personal"))
assert personal_project.organization is None
# create a project in the organization
client.organization_slug = org.slug
org_project = client.projects.create(models.ProjectWriteRequest(name="Org"))
assert org_project.organization == org.id
# both projects should be visible with no context
client.organization_slug = None
client.projects.retrieve(personal_project.id)
client.projects.retrieve(org_project.id)
# retrieve personal and org projects by id
client.organization_slug = ""
client.projects.retrieve(personal_project.id)
client.projects.retrieve(org_project.id)
# org context doesn't make sense for detailed request
client.organization_slug = org.slug
client.projects.retrieve(org_project.id)
client.projects.retrieve(personal_project.id)
@pytest.mark.usefixtures("restore_db_per_function")
def test_organization_filtering(regular_lonely_user: str, fxt_image_file):
with make_client(BASE_URL, credentials=(regular_lonely_user, USER_PASS)) as client:
org = client.organizations.create(models.OrganizationWriteRequest(slug="testorg"))
# create a project and task in sandbox
client.organization_slug = None
client.projects.create(models.ProjectWriteRequest(name="personal_project"))
client.tasks.create_from_data(
spec={"name": "personal_task", "labels": [{"name": "a"}]}, resources=[fxt_image_file]
)
# create a project and task in the organization
client.organization_slug = org.slug
client.projects.create(models.ProjectWriteRequest(name="org_project"))
client.tasks.create_from_data(
spec={"name": "org_task", "labels": [{"name": "a"}]}, resources=[fxt_image_file]
)
# return only non-org objects if org parameter is empty
client.organization_slug = ""
projects, tasks, jobs = client.projects.list(), client.tasks.list(), client.jobs.list()
assert len(projects) == len(tasks) == len(jobs) == 1
assert projects[0].organization == tasks[0].organization == jobs[0].organization == None
# return all objects if org parameter wasn't presented
client.organization_slug = None
projects, tasks, jobs = client.projects.list(), client.tasks.list(), client.jobs.list()
assert len(projects) == len(tasks) == len(jobs) == 2
assert {None, org.id} == set([a.organization for a in (*projects, *tasks, *jobs)])
# return only org objects if org parameter is presented and not empty
client.organization_slug = org.slug
projects, tasks, jobs = client.projects.list(), client.tasks.list(), client.jobs.list()
assert len(projects) == len(tasks) == len(jobs) == 1
assert projects[0].organization == tasks[0].organization == jobs[0].organization == org.id
def test_organization_context_manager():
client = Client(BASE_URL)
client.organization_slug = "abc"
with client.organization_context("def"):
assert client.organization_slug == "def"
assert client.organization_slug == "abc"
Computing file changes ...