You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
399 lines
13 KiB
Python
399 lines
13 KiB
Python
# Copyright (C) 2022 Intel Corporation
|
|
# Copyright (C) 2022-2023 CVAT.ai Corporation
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
import io
|
|
from http import HTTPStatus
|
|
from typing import List
|
|
|
|
import pytest
|
|
from cvat_sdk.api_client import ApiClient, models
|
|
from cvat_sdk.api_client.api_client import Endpoint
|
|
from deepdiff import DeepDiff
|
|
from PIL import Image
|
|
|
|
from shared.utils.config import get_method, patch_method, post_method
|
|
|
|
from .utils import CollectionSimpleFilterTestBase
|
|
|
|
# https://docs.pytest.org/en/7.1.x/example/markers.html#marking-whole-classes-or-modules
|
|
pytestmark = [pytest.mark.with_external_services]
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_class")
|
|
class TestGetCloudStorage:
|
|
def _test_can_see(self, user, storage_id, data, **kwargs):
|
|
response = get_method(user, f"cloudstorages/{storage_id}", **kwargs)
|
|
response_data = response.json()
|
|
response_data = response_data.get("results", response_data)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert (
|
|
DeepDiff(data, response_data, ignore_order=True, exclude_paths="root['updated_date']")
|
|
== {}
|
|
)
|
|
|
|
def _test_cannot_see(self, user, storage_id, **kwargs):
|
|
response = get_method(user, f"cloudstorages/{storage_id}", **kwargs)
|
|
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
@pytest.mark.parametrize("storage_id", [1])
|
|
@pytest.mark.parametrize(
|
|
"group, is_owner, is_allow",
|
|
[
|
|
("admin", False, True),
|
|
("business", False, False),
|
|
("user", True, True),
|
|
],
|
|
)
|
|
def test_sandbox_user_get_coud_storage(
|
|
self, storage_id, group, is_owner, is_allow, users, cloud_storages
|
|
):
|
|
org = ""
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in users
|
|
if group in u["groups"] and u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_see(username, storage_id, cloud_storage, org=org)
|
|
else:
|
|
self._test_cannot_see(username, storage_id, org=org)
|
|
|
|
@pytest.mark.parametrize("org_id", [2])
|
|
@pytest.mark.parametrize("storage_id", [2])
|
|
@pytest.mark.parametrize(
|
|
"role, is_owner, is_allow",
|
|
[
|
|
("worker", True, True),
|
|
("supervisor", False, True),
|
|
("worker", False, False),
|
|
],
|
|
)
|
|
def test_org_user_get_coud_storage(
|
|
self, org_id, storage_id, role, is_owner, is_allow, find_users, cloud_storages
|
|
):
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in find_users(role=role, org=org_id)
|
|
if u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_see(username, storage_id, cloud_storage, org_id=org_id)
|
|
else:
|
|
self._test_cannot_see(username, storage_id, org_id=org_id)
|
|
|
|
|
|
class TestCloudStoragesListFilters(CollectionSimpleFilterTestBase):
|
|
field_lookups = {
|
|
"owner": ["owner", "username"],
|
|
"name": ["display_name"],
|
|
}
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup(self, restore_db_per_class, admin_user, cloud_storages):
|
|
self.user = admin_user
|
|
self.samples = cloud_storages
|
|
|
|
def _get_endpoint(self, api_client: ApiClient) -> Endpoint:
|
|
return api_client.cloudstorages_api.list_endpoint
|
|
|
|
def _retrieve_collection(self, **kwargs) -> List:
|
|
# TODO: fix invalid serializer schema for manifests
|
|
results = super()._retrieve_collection(_parse_response=False, return_json=True, **kwargs)
|
|
for r in results:
|
|
r["manifests"] = [{"filename": m} for m in r["manifests"]]
|
|
return [models.CloudStorageRead._from_openapi_data(**r) for r in results]
|
|
|
|
@pytest.mark.parametrize(
|
|
"field",
|
|
("provider_type", "name", "resource", "credentials_type", "owner"),
|
|
)
|
|
def test_can_use_simple_filter_for_object_list(self, field):
|
|
return super().test_can_use_simple_filter_for_object_list(field)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestPostCloudStorage:
|
|
_SPEC = {
|
|
"provider_type": "AWS_S3_BUCKET",
|
|
"resource": "test",
|
|
"display_name": "Bucket",
|
|
"credentials_type": "KEY_SECRET_KEY_PAIR",
|
|
"key": "minio_access_key",
|
|
"secret_key": "minio_secret_key",
|
|
"specific_attributes": "endpoint_url=http://minio:9000",
|
|
"description": "Some description",
|
|
"manifests": ["manifest.jsonl"],
|
|
}
|
|
_EXCLUDE_PATHS = [
|
|
f"root['{extra_field}']"
|
|
for extra_field in {
|
|
# unchanged fields
|
|
"created_date",
|
|
"id",
|
|
"organization",
|
|
"owner",
|
|
"updated_date",
|
|
# credentials that server doesn't return
|
|
"key",
|
|
"secret_key",
|
|
}
|
|
]
|
|
|
|
def _test_can_create(self, user, spec, **kwargs):
|
|
response = post_method(user, "cloudstorages", spec, **kwargs)
|
|
response_data = response.json()
|
|
response_data = response_data.get("results", response_data)
|
|
|
|
assert response.status_code == HTTPStatus.CREATED
|
|
assert (
|
|
DeepDiff(
|
|
self._SPEC, response_data, ignore_order=True, exclude_paths=self._EXCLUDE_PATHS
|
|
)
|
|
== {}
|
|
)
|
|
|
|
def _test_cannot_create(self, user, spec, **kwargs):
|
|
response = post_method(user, "cloudstorages", spec, **kwargs)
|
|
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
@pytest.mark.parametrize("group, is_allow", [("user", True), ("worker", False)])
|
|
def test_sandbox_user_create_cloud_storage(self, group, is_allow, users):
|
|
org = ""
|
|
username = [u for u in users if group in u["groups"]][0]["username"]
|
|
|
|
if is_allow:
|
|
self._test_can_create(username, self._SPEC, org=org)
|
|
else:
|
|
self._test_cannot_create(username, self._SPEC, org=org)
|
|
|
|
@pytest.mark.parametrize("org_id", [2])
|
|
@pytest.mark.parametrize(
|
|
"role, is_allow",
|
|
[
|
|
("owner", True),
|
|
("maintainer", True),
|
|
("worker", False),
|
|
("supervisor", False),
|
|
],
|
|
)
|
|
def test_org_user_create_coud_storage(self, org_id, role, is_allow, find_users):
|
|
username = find_users(role=role, org=org_id)[0]["username"]
|
|
|
|
if is_allow:
|
|
self._test_can_create(username, self._SPEC, org_id=org_id)
|
|
else:
|
|
self._test_cannot_create(username, self._SPEC, org_id=org_id)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestPatchCloudStorage:
|
|
_SPEC = {
|
|
"display_name": "New display name",
|
|
"description": "New description",
|
|
"manifests": [
|
|
"manifest_1.jsonl",
|
|
"manifest_2.jsonl",
|
|
],
|
|
}
|
|
_PRIVATE_BUCKET_SPEC = {
|
|
"display_name": "New display name",
|
|
"description": "New description",
|
|
"manifests": [
|
|
"sub/manifest_1.jsonl",
|
|
"sub/manifest_2.jsonl",
|
|
],
|
|
}
|
|
_EXCLUDE_PATHS = [
|
|
f"root['{extra_field}']"
|
|
for extra_field in {
|
|
# unchanged fields
|
|
"created_date",
|
|
"credentials_type",
|
|
"id",
|
|
"organization",
|
|
"owner",
|
|
"provider_type",
|
|
"resource",
|
|
"specific_attributes",
|
|
"updated_date",
|
|
}
|
|
]
|
|
|
|
def _test_can_update(self, user, storage_id, spec, **kwargs):
|
|
response = patch_method(user, f"cloudstorages/{storage_id}", spec, **kwargs)
|
|
response_data = response.json()
|
|
response_data = response_data.get("results", response_data)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert (
|
|
DeepDiff(spec, response_data, ignore_order=True, exclude_paths=self._EXCLUDE_PATHS)
|
|
== {}
|
|
)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
def _test_cannot_update(self, user, storage_id, spec, **kwargs):
|
|
response = patch_method(user, f"cloudstorages/{storage_id}", spec, **kwargs)
|
|
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
@pytest.mark.parametrize("storage_id", [1])
|
|
@pytest.mark.parametrize(
|
|
"group, is_owner, is_allow",
|
|
[
|
|
("admin", False, True),
|
|
("business", False, False),
|
|
("worker", True, True),
|
|
],
|
|
)
|
|
def test_sandbox_user_update_cloud_storage(
|
|
self, storage_id, group, is_owner, is_allow, users, cloud_storages
|
|
):
|
|
org = ""
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in users
|
|
if group in u["groups"] and u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_update(username, storage_id, self._SPEC, org=org)
|
|
else:
|
|
self._test_cannot_update(username, storage_id, self._SPEC, org=org)
|
|
|
|
@pytest.mark.parametrize("org_id", [2])
|
|
@pytest.mark.parametrize("storage_id", [2])
|
|
@pytest.mark.parametrize(
|
|
"role, is_owner, is_allow",
|
|
[
|
|
("worker", True, True),
|
|
("maintainer", False, True),
|
|
("supervisor", False, False),
|
|
],
|
|
)
|
|
def test_org_user_update_cloud_storage(
|
|
self, org_id, storage_id, role, is_owner, is_allow, find_users, cloud_storages
|
|
):
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in find_users(role=role, org=org_id)
|
|
if u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_update(username, storage_id, self._PRIVATE_BUCKET_SPEC, org_id=org_id)
|
|
else:
|
|
self._test_cannot_update(username, storage_id, self._PRIVATE_BUCKET_SPEC, org_id=org_id)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_class")
|
|
class TestGetCloudStoragePreview:
|
|
def _test_can_see(self, user, storage_id, **kwargs):
|
|
response = get_method(user, f"cloudstorages/{storage_id}/preview", **kwargs)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
(width, height) = Image.open(io.BytesIO(response.content)).size
|
|
assert width > 0 and height > 0
|
|
|
|
def _test_cannot_see(self, user, storage_id, **kwargs):
|
|
response = get_method(user, f"cloudstorages/{storage_id}/preview", **kwargs)
|
|
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
@pytest.mark.parametrize("storage_id", [1])
|
|
@pytest.mark.parametrize(
|
|
"group, is_owner, is_allow",
|
|
[
|
|
("admin", False, True),
|
|
("business", False, False),
|
|
("user", True, True),
|
|
],
|
|
)
|
|
def test_sandbox_user_get_cloud_storage_preview(
|
|
self, storage_id, group, is_owner, is_allow, users, cloud_storages
|
|
):
|
|
org = ""
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in users
|
|
if group in u["groups"] and u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_see(username, storage_id, org=org)
|
|
else:
|
|
self._test_cannot_see(username, storage_id, org=org)
|
|
|
|
@pytest.mark.parametrize("org_id", [2])
|
|
@pytest.mark.parametrize("storage_id", [2])
|
|
@pytest.mark.parametrize(
|
|
"role, is_owner, is_allow",
|
|
[
|
|
("worker", True, True),
|
|
("supervisor", False, True),
|
|
("worker", False, False),
|
|
],
|
|
)
|
|
def test_org_user_get_cloud_storage_preview(
|
|
self, org_id, storage_id, role, is_owner, is_allow, find_users, cloud_storages
|
|
):
|
|
cloud_storage = cloud_storages[storage_id]
|
|
username = (
|
|
cloud_storage["owner"]["username"]
|
|
if is_owner
|
|
else next(
|
|
(
|
|
u
|
|
for u in find_users(role=role, org=org_id)
|
|
if u["id"] != cloud_storage["owner"]["id"]
|
|
)
|
|
)["username"]
|
|
)
|
|
|
|
if is_allow:
|
|
self._test_can_see(username, storage_id, org_id=org_id)
|
|
else:
|
|
self._test_cannot_see(username, storage_id, org_id=org_id)
|