frostfs-testcases/pytest_tests/testsuites/services/s3_gate/test_s3_multipart.py

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
from frostfs_testlib.steps.cli.container import list_objects, search_container_by_name
from frostfs_testlib.steps.s3 import s3_helper
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.utils.file_utils import generate_file, get_file_hash, split_file

PART_SIZE = 5 * 1024 * 1024


@pytest.mark.s3_gate
@pytest.mark.s3_gate_multipart
class TestS3GateMultipart(ClusterTestBase):
    NO_SUCH_UPLOAD = "The upload ID may be invalid, or the upload may have been aborted or completed."

    @allure.title("Object Multipart API (s3_client={s3_client}, bucket versioning = {versioning_status})")
    @pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED, VersioningStatus.UNDEFINED], indirect=True)
    def test_s3_object_multipart(
        self, s3_client: S3ClientWrapper, bucket: str, default_wallet: WalletInfo, versioning_status: str
    ):
        parts_count = 5
        file_name_large = generate_file(PART_SIZE * parts_count)  # 5Mb - min part
        object_key = s3_helper.object_key_from_file_path(file_name_large)
        part_files = split_file(file_name_large, parts_count)
        parts = []

        with reporter.step(f"Get related container_id for bucket"):
            for cluster_node in self.cluster.cluster_nodes:
                container_id = search_container_by_name(bucket, cluster_node)
                if container_id:
                    break

        with reporter.step("Upload first part"):
            upload_id = s3_client.create_multipart_upload(bucket, object_key)
            uploads = s3_client.list_multipart_uploads(bucket)
            etag = s3_client.upload_part(bucket, object_key, upload_id, 1, part_files[0])
            parts.append((1, etag))
            got_parts = s3_client.list_parts(bucket, object_key, upload_id)
            assert len(got_parts) == 1, f"Expected {1} parts, got\n{got_parts}"

        with reporter.step("Upload last parts"):
            for part_id, file_path in enumerate(part_files[1:], start=2):
                etag = s3_client.upload_part(bucket, object_key, upload_id, part_id, file_path)
                parts.append((part_id, etag))
            got_parts = s3_client.list_parts(bucket, object_key, upload_id)
            response = s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)

            version_id = None
            if versioning_status == VersioningStatus.ENABLED:
                version_id = response["VersionId"]
            assert len(got_parts) == len(part_files), f"Expected {parts_count} parts, got\n{got_parts}"

        with reporter.step("Check upload list is empty"):
            uploads = s3_client.list_multipart_uploads(bucket)
            assert not uploads, f"Expected there is no uploads in bucket {bucket}"

        with reporter.step("Check we can get whole object from bucket"):
            got_object = s3_client.get_object(bucket, object_key)
            assert get_file_hash(got_object) == get_file_hash(file_name_large)

        if version_id:
            with reporter.step("Delete the object version"):
                s3_client.delete_object(bucket, object_key, version_id)
        else:
            with reporter.step("Delete the object"):
                s3_client.delete_object(bucket, object_key)

        with reporter.step("List objects in the bucket, expect to be empty"):
            objects_list = s3_client.list_objects(bucket)
            assert not objects_list, f"Expected empty bucket, got {objects_list}"

        with reporter.step("List objects in the container via rpc, expect to be empty"):
            objects = list_objects(default_wallet, self.shell, container_id, self.cluster.default_rpc_endpoint)
            assert len(objects) == 0, f"Expected no objects in container, got\n{objects}"

    @allure.title("Abort Multipart Upload (s3_client={s3_client})")
    @pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED], indirect=True)
    def test_s3_abort_multipart(
        self,
        s3_client: S3ClientWrapper,
        default_wallet: WalletInfo,
        bucket: str,
        simple_object_size: ObjectSize,
        complex_object_size: ObjectSize,
    ):
        complex_file = generate_file(complex_object_size.value)
        simple_file = generate_file(simple_object_size.value)
        to_upload = [complex_file, complex_file, simple_file]
        files_count = len(to_upload)
        upload_key = "multipart_abort"

        with reporter.step(f"Get related container_id for bucket '{bucket}'"):
            for cluster_node in self.cluster.cluster_nodes:
                container_id = search_container_by_name(bucket, cluster_node)
                if container_id:
                    break

        with reporter.step("Create multipart upload"):
            upload_id = s3_client.create_multipart_upload(bucket, upload_key)

        with reporter.step(f"Upload {files_count} files to multipart upload"):
            for i, file in enumerate(to_upload, 1):
                s3_client.upload_part(bucket, upload_key, upload_id, i, file)

        with reporter.step(f"Check that we have {files_count} files in bucket"):
            parts = s3_client.list_parts(bucket, upload_key, upload_id)
            assert len(parts) == files_count, f"Expected {files_count} parts, got\n{parts}"

        with reporter.step(f"Check that we have {files_count} files in container '{container_id}'"):
            objects = list_objects(default_wallet, self.shell, container_id, self.cluster.default_rpc_endpoint)
            assert len(objects) == files_count, f"Expected {files_count} objects in container, got\n{objects}"

        with reporter.step("Abort multipart upload"):
            s3_client.abort_multipart_upload(bucket, upload_key, upload_id)
            uploads = s3_client.list_multipart_uploads(bucket)
            assert not uploads, f"Expected no uploads in bucket {bucket}"

        with reporter.step("Check that we have no files in bucket since upload was aborted"):
            with pytest.raises(Exception, match=self.NO_SUCH_UPLOAD):
                s3_client.list_parts(bucket, upload_key, upload_id)

        with reporter.step("Check that we have no files in container since upload was aborted"):
            objects = list_objects(default_wallet, self.shell, container_id, self.cluster.default_rpc_endpoint)
            assert len(objects) == 0, f"Expected no objects in container, got\n{objects}"

    @allure.title("Upload Part Copy (s3_client={s3_client})")
    @pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED], indirect=True)
    def test_s3_multipart_copy(self, s3_client: S3ClientWrapper, bucket: str):
        parts_count = 3
        file_name_large = generate_file(PART_SIZE * parts_count)  # 5Mb - min part
        object_key = s3_helper.object_key_from_file_path(file_name_large)
        part_files = split_file(file_name_large, parts_count)
        parts = []
        objs = []

        with reporter.step(f"Put {parts_count} objects in bucket"):
            for part in part_files:
                s3_client.put_object(bucket, part)
                objs.append(s3_helper.object_key_from_file_path(part))
            s3_helper.check_objects_in_bucket(s3_client, bucket, objs)

        with reporter.step("Create multipart upload object"):
            upload_id = s3_client.create_multipart_upload(bucket, object_key)
            uploads = s3_client.list_multipart_uploads(bucket)
            assert uploads, f"Expected there are uploads in bucket {bucket}"

        with reporter.step("Upload parts to multipart upload"):
            for part_id, obj_key in enumerate(objs, start=1):
                etag = s3_client.upload_part_copy(bucket, object_key, upload_id, part_id, f"{bucket}/{obj_key}")
                parts.append((part_id, etag))
            got_parts = s3_client.list_parts(bucket, object_key, upload_id)

        with reporter.step("Complete multipart upload"):
            s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)
            assert len(got_parts) == len(part_files), f"Expected {parts_count} parts, got\n{got_parts}"

        with reporter.step("Check we can get whole object from bucket"):
            got_object = s3_client.get_object(bucket, object_key)
            assert get_file_hash(got_object) == get_file_hash(file_name_large)