Andrei Strongin
69cdda3e75
<!-- Raise an issue to propose your change (https://github.com/cvat-ai/cvat/issues). It helps to avoid duplication of efforts from multiple independent contributors. Discuss your ideas with maintainers to be sure that changes will be approved and merged. Read the [Contribution guide](https://docs.cvat.ai/docs/contributing/). --> <!-- Provide a general summary of your changes in the Title above --> ### Motivation and context This branch completes the CLI workflow for persistent authentication profiles. Users can save a CVAT server URL and Personal Access Token under a named profile, then reuse it with `--profile` instead of repeatedly passing credentials on the command line. It adds `cvat-cli profile` commands to: - List saved profiles, including the active default. - Create a profile from a PAT, optionally deriving its name from the server. - Set, print, or clear the default profile. - Delete profiles safely. The implementation builds on the auth/config resolution already present in `develop`. It uses the SDK’s `AuthStore.put_profile` API, preserves copy-on-write store updates, and applies the same server-port validation used by normal CLI client creation. URL normalization was also moved into shared SDK utilities so both `Client` and profile creation use one implementation for default HTTPS schemes, trailing-slash removal, and invalid-scheme validation. ### How has this been tested? <!-- Please describe in detail how you tested your changes. Include details of your testing environment, and the tests you ran to see how your change affects other areas of the code, etc. --> ### Checklist <!-- Go over all the following points, and put an `x` in all the boxes that apply. If an item isn't applicable for some reason, then ~~explicitly strikethrough~~ the whole line. If you don't do that, GitHub will show incorrect progress for the pull request. If you're unsure about any of these, don't hesitate to ask. We're here to help! --> - [x] I submit my changes into the `develop` branch - [x] I have created a changelog fragment <!-- see top comment in CHANGELOG.md --> - [ ] I have updated the documentation accordingly - [x] I have added tests to cover my changes - [x] I have linked related issues (see [GitHub docs]( https://help.github.com/en/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword)) ### License - [x] I submit _my code changes_ under the same [MIT License]( https://github.com/cvat-ai/cvat/blob/develop/LICENSE) that covers the project. Feel free to contact the maintainers if that's a concern. --------- Co-authored-by: Maxim Zhiltsov <maxim@cvat.ai>
328 lines
12 KiB
Python
328 lines
12 KiB
Python
# Copyright (C) CVAT.ai Corporation
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
import io
|
|
from contextlib import ExitStack
|
|
from logging import Logger
|
|
|
|
import packaging.version as pv
|
|
import pytest
|
|
from cvat_sdk import Client, models
|
|
from cvat_sdk.core.client import AccessTokenCredentials, Config, PasswordCredentials, make_client
|
|
from cvat_sdk.core.exceptions import IncompatibleVersionException, InvalidHostException
|
|
from cvat_sdk.core.utils import normalize_server_url
|
|
from cvat_sdk.exceptions import ApiException
|
|
|
|
from shared.utils.config import BASE_URL, USER_PASS
|
|
|
|
from .util import https_reverse_proxy
|
|
|
|
|
|
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(PasswordCredentials(self.user, USER_PASS))
|
|
|
|
assert self.client.users.retrieve_current_user().username == self.user
|
|
|
|
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_after_basic_auth_login(self):
|
|
self.client.login((self.user, USER_PASS))
|
|
|
|
self.client.logout()
|
|
|
|
assert not self.client.has_credentials()
|
|
|
|
def test_can_login_with_pat_auth(self, access_tokens_by_username):
|
|
user, token = next((u, t) for u, ts in access_tokens_by_username.items() for t in ts)
|
|
|
|
self.client.login(AccessTokenCredentials(token["private_key"]))
|
|
|
|
assert self.client.users.retrieve_current_user().username == user
|
|
|
|
assert self.client.has_credentials()
|
|
|
|
def test_can_logout_after_pat_login(self, access_tokens):
|
|
token = next(t for t in access_tokens)
|
|
|
|
self.client.login(AccessTokenCredentials(token["private_key"]))
|
|
|
|
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)
|
|
|
|
|
|
class TestClientFactory:
|
|
def test_can_make_client_with_pat_auth(self, access_tokens_by_username):
|
|
user, token = next((u, t) for u, ts in access_tokens_by_username.items() for t in ts)
|
|
host, port = BASE_URL.rsplit(":", maxsplit=1)
|
|
|
|
with make_client(host=host, port=port, access_token=token["private_key"]) as client:
|
|
assert client.users.retrieve_current_user().username == user
|
|
|
|
def test_can_strip_trailing_slash_in_hostname(self, admin_user: str):
|
|
host, port = BASE_URL.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(self, 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_add_default_server_schema(self):
|
|
assert normalize_server_url("localhost:8080/") == "https://localhost:8080"
|
|
with https_reverse_proxy() as proxy_url:
|
|
with Client(
|
|
url=proxy_url.removeprefix("https://"), config=Config(verify_ssl=False)
|
|
) as client:
|
|
assert client.api_map.host == proxy_url
|
|
|
|
def test_can_reject_invalid_server_schema(self):
|
|
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'")
|
|
|
|
def test_can_use_server_url(self, admin_user):
|
|
with make_client(BASE_URL, credentials=(admin_user, USER_PASS)):
|
|
pass
|
|
|
|
def test_cannot_use_server_url_with_port_and_port_parameter(self):
|
|
with pytest.raises(ValueError, match="Please specify only one port"):
|
|
make_client(BASE_URL, port=1)
|
|
|
|
def test_cannot_use_both_credentials_and_access_token(self):
|
|
with pytest.raises(ValueError, match="'credentials' and 'access_token' cannot"):
|
|
make_client(BASE_URL, credentials=("name", "pass"), access_token="token")
|
|
|
|
|
|
@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"}, 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"}, 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"
|