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

import logging

import allure
import pytest

import pytest_tests.helpers.container as container
from pytest_tests.helpers.file_helper import generate_file, get_file_hash, split_file
from pytest_tests.helpers.s3_helper import check_objects_in_bucket, object_key_from_file_path
from pytest_tests.steps import s3_gate_bucket, s3_gate_object
from pytest_tests.steps.s3_gate_base import TestS3GateBase

PART_SIZE = 5 * 1024 * 1024


def pytest_generate_tests(metafunc):
    if "s3_client" in metafunc.fixturenames:
        metafunc.parametrize("s3_client", ["aws cli", "boto3"], indirect=True)


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

    @allure.title("Test S3 Object Multipart API")
    @pytest.mark.parametrize("bucket", [s3_gate_bucket.VersioningStatus.ENABLED], indirect=True)
    def test_s3_object_multipart(self, bucket):
        parts_count = 5
        file_name_large = generate_file(PART_SIZE * parts_count)  # 5Mb - min part
        object_key = object_key_from_file_path(file_name_large)
        part_files = split_file(file_name_large, parts_count)
        parts = []

        with allure.step("Upload first part"):
            upload_id = s3_gate_object.create_multipart_upload_s3(
                self.s3_client, bucket, object_key
            )
            uploads = s3_gate_object.list_multipart_uploads_s3(self.s3_client, bucket)
            etag = s3_gate_object.upload_part_s3(
                self.s3_client, bucket, object_key, upload_id, 1, part_files[0]
            )
            parts.append((1, etag))
            got_parts = s3_gate_object.list_parts_s3(self.s3_client, bucket, object_key, upload_id)
            assert len(got_parts) == 1, f"Expected {1} parts, got\n{got_parts}"

        with allure.step("Upload last parts"):
            for part_id, file_path in enumerate(part_files[1:], start=2):
                etag = s3_gate_object.upload_part_s3(
                    self.s3_client, bucket, object_key, upload_id, part_id, file_path
                )
                parts.append((part_id, etag))
            got_parts = s3_gate_object.list_parts_s3(self.s3_client, bucket, object_key, upload_id)
            s3_gate_object.complete_multipart_upload_s3(
                self.s3_client, bucket, object_key, upload_id, parts
            )
            assert len(got_parts) == len(
                part_files
            ), f"Expected {parts_count} parts, got\n{got_parts}"

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

        with allure.step("Check we can get whole object from bucket"):
            got_object = s3_gate_object.get_object_s3(self.s3_client, bucket, object_key)
            assert get_file_hash(got_object) == get_file_hash(file_name_large)

    @allure.title("Test S3 Multipart abort")
    @pytest.mark.parametrize("bucket", [s3_gate_bucket.VersioningStatus.ENABLED], indirect=True)
    def test_s3_abort_multipart(
        self, bucket: str, simple_object_size: int, complex_object_size: int
    ):
        complex_file = generate_file(complex_object_size)
        simple_file = generate_file(simple_object_size)
        to_upload = [complex_file, complex_file, simple_file]
        files_count = len(to_upload)
        upload_key = "multipart_abort"

        with allure.step(f"Get related container_id for bucket '{bucket}'"):
            container_id = container.search_container_by_name(
                self.wallet, bucket, self.shell, self.cluster.default_rpc_endpoint
            )

        with allure.step("Create multipart upload"):
            upload_id = s3_gate_object.create_multipart_upload_s3(
                self.s3_client, bucket, upload_key
            )

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

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

        with allure.step(f"Check that we have {files_count} files in container '{container_id}'"):
            objects = container.list_objects(
                self.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 allure.step("Abort multipart upload"):
            s3_gate_object.abort_multipart_upload_s3(self.s3_client, bucket, upload_key, upload_id)
            uploads = s3_gate_object.list_multipart_uploads_s3(self.s3_client, bucket)
            assert not uploads, f"Expected no uploads in bucket {bucket}"

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

        with allure.step("Check that we have no files in container since upload was aborted"):
            objects = container.list_objects(
                self.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("Test S3 Upload Part Copy")
    @pytest.mark.parametrize("bucket", [s3_gate_bucket.VersioningStatus.ENABLED], indirect=True)
    def test_s3_multipart_copy(self, bucket):
        parts_count = 3
        file_name_large = generate_file(PART_SIZE * parts_count)  # 5Mb - min part
        object_key = object_key_from_file_path(file_name_large)
        part_files = split_file(file_name_large, parts_count)
        parts = []
        objs = []

        with allure.step(f"Put {parts_count} objects in bucket"):
            for part in part_files:
                s3_gate_object.put_object_s3(self.s3_client, bucket, part)
                objs.append(object_key_from_file_path(part))
            check_objects_in_bucket(self.s3_client, bucket, objs)

        with allure.step("Create multipart upload object"):
            upload_id = s3_gate_object.create_multipart_upload_s3(
                self.s3_client, bucket, object_key
            )
            uploads = s3_gate_object.list_multipart_uploads_s3(self.s3_client, bucket)
            assert uploads, f"Expected there are uploads in bucket {bucket}"

        with allure.step("Start multipart upload"):
            for part_id, obj_key in enumerate(objs, start=1):
                etag = s3_gate_object.upload_part_copy_s3(
                    self.s3_client, bucket, object_key, upload_id, part_id, f"{bucket}/{obj_key}"
                )
                parts.append((part_id, etag))
            got_parts = s3_gate_object.list_parts_s3(self.s3_client, bucket, object_key, upload_id)
            s3_gate_object.complete_multipart_upload_s3(
                self.s3_client, bucket, object_key, upload_id, parts
            )
            assert len(got_parts) == len(
                part_files
            ), f"Expected {parts_count} parts, got\n{got_parts}"

        with allure.step("Check we can get whole object from bucket"):
            got_object = s3_gate_object.get_object_s3(self.s3_client, bucket, object_key)
            assert get_file_hash(got_object) == get_file_hash(file_name_large)