Refactor resource import export tests (#5429)

Extracted some enhancements from
https://github.com/opencv/cvat/pull/4819

- Extracted common s3 manipulations in tests
- Refactored import/export tests to be more clear

tests/python/rest_api/test_resource_import_export.py

@@ -5,109 +5,131 @@
 
 import functools
 import json
+from contextlib import ExitStack
 from http import HTTPStatus
+from typing import Any, Dict, TypeVar
 
-import boto3
 import pytest
-from botocore.exceptions import ClientError
 
-from shared.utils.config import (
+from shared.utils.config import get_method, post_method
-    MINIO_ENDPOINT_URL,
+from shared.utils.s3 import make_client
-    MINIO_KEY,
+
-    MINIO_SECRET_KEY,
+T = TypeVar("T")
-    get_method,
-    post_method,
-)
 
 FILENAME_TEMPLATE = "cvat/{}/{}.zip"
 FORMAT = "COCO 1.0"
 
 
-def _use_custom_settings(obj, resource, cloud_storage_id):
+def _make_custom_resource_params(obj: str, resource: str, cloud_storage_id: int) -> Dict[str, Any]:
-    return {
+    params = {
         "filename": FILENAME_TEMPLATE.format(obj, resource),
-        "use_default_location": False,
         "location": "cloud_storage",
         "cloud_storage_id": cloud_storage_id,
-        "format": FORMAT,
+        "use_default_location": False,
     }
+    if resource != "backup":
+        params["format"] = FORMAT
+    return params
 
 
-def _use_default_settings(obj, resource):
+def _make_default_resource_params(obj: str, resource: str) -> Dict[str, Any]:
-    return {
+    params = {
         "filename": FILENAME_TEMPLATE.format(obj, resource),
         "use_default_location": True,
-        "format": FORMAT,
     }
+    if resource != "backup":
+        params["format"] = FORMAT
+    return params
 
 
-def define_client():
+class _S3ResourceTest:
-    s3 = boto3.resource(
+    @pytest.fixture(autouse=True)
-        "s3",
+    def setup(self, admin_user: str):
-        aws_access_key_id=MINIO_KEY,
+        self.user = admin_user
-        aws_secret_access_key=MINIO_SECRET_KEY,
+        self.s3_client = make_client()
-        endpoint_url=MINIO_ENDPOINT_URL,
+        self.exit_stack = ExitStack()
-    )
+        with self.exit_stack:
-    return s3.meta.client
+            yield
 
+    def _ensure_file_created(self, func: T, storage: Dict[str, Any]) -> T:
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            filename = kwargs["filename"]
+            bucket = storage["resource"]
 
-def assert_file_does_not_exist(client, bucket, filename):
+            # check that file doesn't exist on the bucket
-    try:
+            assert not self.s3_client.file_exists(bucket, filename)
-        client.head_object(Bucket=bucket, Key=filename)
-        raise AssertionError(f"File {filename} on bucket {bucket} already exists")
-    except ClientError:
-        pass
 
+            func(*args, **kwargs)
 
-def assert_file_exists(client, bucket, filename):
+            # check that file exists on the bucket
-    try:
+            assert self.s3_client.file_exists(bucket, filename)
-        client.head_object(Bucket=bucket, Key=filename)
-    except ClientError:
-        raise AssertionError(f"File {filename} on bucket {bucket} doesn't exist")
 
+        return wrapper
 
-def assert_file_status(func):
+    def _export_resource_to_cloud_storage(
-    @functools.wraps(func)
+        self, obj_id: int, obj: str, resource: str, *, user: str, **kwargs
-    def wrapper(user, storage_conf, *args, **kwargs):
+    ):
-        filename = kwargs["filename"]
+        response = get_method(user, f"{obj}/{obj_id}/{resource}", **kwargs)
-        bucket = storage_conf["resource"]
+        status = response.status_code
-        # get storage client
-        client = define_client()
-        # check that file doesn't exist on the bucket
-        assert_file_does_not_exist(client, bucket, filename)
-        func(user, storage_conf, *args, **kwargs)
-        # check that file exists on the bucket
-        assert_file_exists(client, bucket, filename)
 
-    return wrapper
+        while status != HTTPStatus.OK:
+            assert status in (HTTPStatus.CREATED, HTTPStatus.ACCEPTED)
+            response = get_method(user, f"{obj}/{obj_id}/{resource}", action="download", **kwargs)
+            status = response.status_code
 
+    def _import_annotations_from_cloud_storage(self, obj_id, obj, *, user, **kwargs):
+        url = f"{obj}/{obj_id}/annotations"
+        response = post_method(user, url, data=None, **kwargs)
+        status = response.status_code
 
-def remove_asset(bucket, filename):
+        while status != HTTPStatus.CREATED:
-    client = define_client()
+            assert status == HTTPStatus.ACCEPTED
-    client.delete_object(Bucket=bucket, Key=filename)
+            response = post_method(user, url, data=None, **kwargs)
+            status = response.status_code
 
+    def _import_backup_from_cloud_storage(self, obj_id, obj, *, user, **kwargs):
+        url = f"{obj}/backup"
+        response = post_method(user, url, data=None, **kwargs)
+        status = response.status_code
 
-@assert_file_status
+        while status != HTTPStatus.CREATED:
-def _save_resource_to_cloud_storage(user, storage_conf, obj_id, obj, resource, **kwargs):
+            assert status == HTTPStatus.ACCEPTED
-    response = get_method(user, f"{obj}/{obj_id}/{resource}", **kwargs)
+            data = json.loads(response.content.decode("utf8"))
-    status = response.status_code
+            response = post_method(user, url, data=data, **kwargs)
+            status = response.status_code
 
-    while status != HTTPStatus.OK:
+    def _import_dataset_from_cloud_storage(self, obj_id, obj, *, user, **kwargs):
-        assert status in (HTTPStatus.CREATED, HTTPStatus.ACCEPTED)
+        url = f"{obj}/{obj_id}/dataset"
-        response = get_method(user, f"{obj}/{obj_id}/{resource}", action="download", **kwargs)
+        response = post_method(user, url, data=None, **kwargs)
         status = response.status_code
 
+        while status != HTTPStatus.CREATED:
+            assert status == HTTPStatus.ACCEPTED
+            response = get_method(user, url, action="import_status")
+            status = response.status_code
 
-def _idempotent_saving_resource_to_cloud_storage(*args, **kwargs):
+    def _export_resource(self, cloud_storage: Dict[str, Any], *args, **kwargs):
-    _save_resource_to_cloud_storage(*args, **kwargs)
+        org_id = cloud_storage["organization"]
-    remove_asset(args[1]["resource"], kwargs["filename"])
+        if org_id:
+            kwargs.setdefault("org_id", org_id)
 
+        kwargs.setdefault("user", self.user)
+
+        export_callback = self._ensure_file_created(
+            self._export_resource_to_cloud_storage, storage=cloud_storage
+        )
+        export_callback(*args, **kwargs)
+
+        self.exit_stack.callback(
+            self.s3_client.remove_file,
+            bucket=cloud_storage["resource"],
+            filename=kwargs["filename"],
+        )
 
-@pytest.mark.usefixtures("restore_db_per_class")
-class TestSaveResource:
-    _USERNAME = "admin1"
-    _ORG = 2
 
+@pytest.mark.usefixtures("restore_db_per_class")
+class TestExportResource(_S3ResourceTest):
     @pytest.mark.parametrize("cloud_storage_id", [3])
     @pytest.mark.parametrize(
         "obj_id, obj, resource",
@@ -126,13 +148,9 @@ class TestSaveResource:
         self, cloud_storage_id, obj_id, obj, resource, cloud_storages
     ):
         cloud_storage = cloud_storages[cloud_storage_id]
-        kwargs = _use_custom_settings(obj, resource, cloud_storage_id)
+        kwargs = _make_custom_resource_params(obj, resource, cloud_storage_id)
-        if resource == "backup":
-            kwargs.pop("format")
 
-        _idempotent_saving_resource_to_cloud_storage(
+        self._export_resource(cloud_storage, obj_id, obj, resource, **kwargs)
-            self._USERNAME, cloud_storage, obj_id, obj, resource, org_id=self._ORG, **kwargs
-        )
 
     @pytest.mark.parametrize(
         "obj_id, obj, resource",
@@ -168,56 +186,28 @@ class TestSaveResource:
             task_id = jobs[obj_id]["task_id"]
             cloud_storage_id = tasks[task_id]["target_storage"]["cloud_storage_id"]
         cloud_storage = cloud_storages[cloud_storage_id]
+        kwargs = _make_default_resource_params(obj, resource)
 
-        kwargs = _use_default_settings(obj, resource)
+        self._export_resource(cloud_storage, obj_id, obj, resource, **kwargs)
-
-        if resource == "backup":
-            kwargs.pop("format")
-
-        _idempotent_saving_resource_to_cloud_storage(
-            self._USERNAME, cloud_storage, obj_id, obj, resource, org_id=self._ORG, **kwargs
-        )
-
-
-def _import_annotations_from_cloud_storage(user, obj_id, obj, **kwargs):
-    url = f"{obj}/{obj_id}/annotations"
-    response = post_method(user, url, data=None, **kwargs)
-    status = response.status_code
-
-    while status != HTTPStatus.CREATED:
-        assert status == HTTPStatus.ACCEPTED
-        response = post_method(user, url, data=None, **kwargs)
-        status = response.status_code
-
 
-def _import_backup_from_cloud_storage(user, obj_id, obj, **kwargs):
-    url = f"{obj}/backup"
-    response = post_method(user, url, data=None, **kwargs)
-    status = response.status_code
 
-    while status != HTTPStatus.CREATED:
+@pytest.mark.usefixtures("restore_db_per_function")
-        assert status == HTTPStatus.ACCEPTED
+@pytest.mark.usefixtures("restore_cvat_data")
-        data = json.loads(response.content.decode("utf8"))
+class TestImportResource(_S3ResourceTest):
-        response = post_method(user, url, data=data, **kwargs)
+    def _import_resource(self, cloud_storage: Dict[str, Any], resource_type: str, *args, **kwargs):
-        status = response.status_code
+        methods = {
-
+            "annotations": self._import_annotations_from_cloud_storage,
-
+            "dataset": self._import_dataset_from_cloud_storage,
-def _import_dataset_from_cloud_storage(user, obj_id, obj, **kwargs):
+            "backup": self._import_backup_from_cloud_storage,
-    url = f"{obj}/{obj_id}/dataset"
+        }
-    response = post_method(user, url, data=None, **kwargs)
-    status = response.status_code
 
-    while status != HTTPStatus.CREATED:
+        org_id = cloud_storage["organization"]
-        assert status == HTTPStatus.ACCEPTED
+        if org_id:
-        response = get_method(user, url, action="import_status")
+            kwargs.setdefault("org_id", org_id)
-        status = response.status_code
 
+        kwargs.setdefault("user", self.user)
 
-@pytest.mark.usefixtures("restore_db_per_function")
+        return methods[resource_type](*args, **kwargs)
-@pytest.mark.usefixtures("restore_cvat_data")
-class TestImportResource:
-    _USERNAME = "admin1"
-    _ORG = 2
 
     @pytest.mark.parametrize("cloud_storage_id", [3])
     @pytest.mark.parametrize(
@@ -234,26 +224,11 @@ class TestImportResource:
         self, cloud_storage_id, obj_id, obj, resource, cloud_storages
     ):
         cloud_storage = cloud_storages[cloud_storage_id]
-        kwargs = _use_custom_settings(obj, resource, cloud_storage_id)
+        kwargs = _make_custom_resource_params(obj, resource, cloud_storage_id)
-        export_kwargs = _use_custom_settings(obj, resource, cloud_storage_id)
+        export_kwargs = _make_custom_resource_params(obj, resource, cloud_storage_id)
-
+        self._export_resource(cloud_storage, obj_id, obj, resource, **export_kwargs)
-        if resource == "backup":
-            kwargs.pop("format")
-            kwargs.pop("use_default_location")
-            export_kwargs.pop("format")
 
-        # export current resource to cloud storage
+        self._import_resource(cloud_storage, resource, obj_id, obj, **kwargs)
-        _save_resource_to_cloud_storage(
-            self._USERNAME, cloud_storage, obj_id, obj, resource, org_id=self._ORG, **export_kwargs
-        )
-
-        import_resource = {
-            "annotations": _import_annotations_from_cloud_storage,
-            "dataset": _import_dataset_from_cloud_storage,
-            "backup": _import_backup_from_cloud_storage,
-        }
-        import_resource[resource](self._USERNAME, obj_id, obj, org_id=self._ORG, **kwargs)
-        remove_asset(cloud_storage["resource"], kwargs["filename"])
 
     @pytest.mark.parametrize(
         "obj_id, obj, resource",
@@ -284,17 +259,8 @@ class TestImportResource:
             task_id = jobs[obj_id]["task_id"]
             cloud_storage_id = tasks[task_id]["source_storage"]["cloud_storage_id"]
         cloud_storage = cloud_storages[cloud_storage_id]
-        kwargs = _use_default_settings(obj, resource)
 
-        # export current resource to cloud storage
+        kwargs = _make_default_resource_params(obj, resource)
-        _save_resource_to_cloud_storage(
+        self._export_resource(cloud_storage, obj_id, obj, resource, **kwargs)
-            self._USERNAME, cloud_storage, obj_id, obj, resource, org_id=self._ORG, **kwargs
-        )
 
-        import_resource = {
+        self._import_resource(cloud_storage, resource, obj_id, obj, **kwargs)
-            "annotations": _import_annotations_from_cloud_storage,
-            "dataset": _import_dataset_from_cloud_storage,
-            "backup": _import_backup_from_cloud_storage,
-        }
-        import_resource[resource](self._USERNAME, obj_id, obj, org_id=self._ORG, **kwargs)
-        remove_asset(cloud_storage["resource"], kwargs["filename"])

tests/python/shared/utils/s3.py

@@ -0,0 +1,47 @@
+# Copyright (C) 2022 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+import boto3
+from botocore.exceptions import ClientError
+
+from shared.utils.config import MINIO_ENDPOINT_URL, MINIO_KEY, MINIO_SECRET_KEY
+
+
+class S3Client:
+    def __init__(self, endpoint_url: str, *, access_key: str, secret_key: str) -> None:
+        self.client = self._make_boto_client(
+            endpoint_url=endpoint_url, access_key=access_key, secret_key=secret_key
+        )
+
+    @staticmethod
+    def _make_boto_client(endpoint_url: str, *, access_key: str, secret_key: str):
+        s3 = boto3.resource(
+            "s3",
+            aws_access_key_id=access_key,
+            aws_secret_access_key=secret_key,
+            endpoint_url=endpoint_url,
+        )
+        return s3.meta.client
+
+    def create_file(self, bucket: str, filename: str, data: bytes = b""):
+        self.client.put_object(Body=data, Bucket=bucket, Key=filename)
+
+    def remove_file(self, bucket: str, filename: str):
+        self.client.delete_object(Bucket=bucket, Key=filename)
+
+    def file_exists(self, bucket: str, filename: str) -> bool:
+        try:
+            self.client.head_object(Bucket=bucket, Key=filename)
+            return True
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "404":
+                return False
+            else:
+                raise
+
+
+def make_client() -> S3Client:
+    return S3Client(
+        endpoint_url=MINIO_ENDPOINT_URL, access_key=MINIO_KEY, secret_key=MINIO_SECRET_KEY
+    )