diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_versioning.py b/pytest_tests/testsuites/services/s3_gate/test_s3_versioning.py new file mode 100644 index 0000000..cf43920 --- /dev/null +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_versioning.py @@ -0,0 +1,91 @@ +import os + +import allure +import pytest +from python_keywords.utility_keywords import generate_file +from s3_helper import set_bucket_versioning +from steps import s3_gate_bucket, s3_gate_object +from steps.aws_cli_client import AwsCliClient +from steps.s3_gate_base import TestS3GateBase +from utility import create_file_with_content + + +def pytest_generate_tests(metafunc): + if "s3_client" in metafunc.fixturenames: + metafunc.parametrize("s3_client", ["aws cli", "boto3"], indirect=True) + + +@pytest.mark.s3_gate +class TestS3GateVersioning(TestS3GateBase): + @staticmethod + def object_key_from_file_path(full_path: str) -> str: + return os.path.basename(full_path) + + @allure.title("Test S3: try to disable versioning") + def test_s3_version_off(self): + + bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True) + with pytest.raises(Exception): + set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.SUSPENDED) + + @allure.title("Test S3: Enable and disable versioning") + def test_s3_version(self): + file_path = generate_file() + file_name = self.object_key_from_file_path(file_path) + bucket_objects = [file_name] + bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, False) + set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.SUSPENDED) + + with allure.step("Put object into bucket"): + s3_gate_object.put_object_s3(self.s3_client, bucket, file_path) + objects_list = s3_gate_object.list_objects_s3(self.s3_client, bucket) + assert ( + objects_list == bucket_objects + ), f"Expected list with single objects in bucket, got {objects_list}" + object_version = s3_gate_object.list_objects_versions_s3(self.s3_client, bucket) + actual_version = [ + version.get("VersionId") + for version in object_version + if version.get("Key") == file_name + ] + assert actual_version == ["null"], f"Expected version is null, got {object_version}" + + set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.ENABLED) + + with allure.step("Put several versions of object into bucket"): + version_id_1 = s3_gate_object.put_object_s3(self.s3_client, bucket, file_path) + file_name_1 = create_file_with_content(file_path=file_path) + version_id_2 = s3_gate_object.put_object_s3(self.s3_client, bucket, file_name_1) + + with allure.step("Check bucket shows all versions"): + versions = s3_gate_object.list_objects_versions_s3(self.s3_client, bucket) + obj_versions = [ + version.get("VersionId") for version in versions if version.get("Key") == file_name + ] + assert ( + obj_versions.sort() == [version_id_1, version_id_2, "null"].sort() + ), f"Expected object has versions: {version_id_1, version_id_2, 'null'}" + + with allure.step("Get object"): + object_1 = s3_gate_object.get_object_s3( + self.s3_client, bucket, file_name, full_output=True + ) + assert ( + object_1.get("VersionId") == version_id_2 + ), f"Get object with version {version_id_2}" + + with allure.step("Get first version of object"): + object_2 = s3_gate_object.get_object_s3( + self.s3_client, bucket, file_name, version_id_1, full_output=True + ) + assert ( + object_2.get("VersionId") == version_id_1 + ), f"Get object with version {version_id_1}" + + with allure.step("Get second version of object"): + object_3 = s3_gate_object.get_object_s3( + self.s3_client, bucket, file_name, version_id_2, full_output=True + ) + assert ( + object_3.get("VersionId") == version_id_2 + ), f"Get object with version {version_id_2}"