From 92f74707570ed1cafab7148e86f6829eae27424d Mon Sep 17 00:00:00 2001
From: Yulia Kovshova <y.kovshova@yadro.com>
Date: Mon, 26 Sep 2022 11:25:49 +0300
Subject: [PATCH] [#312] add version test
Signed-off-by: Yulia Kovshova <y.kovshova@yadro.com>
---
.../services/s3_gate/test_s3_versioning.py | 91 +++++++++++++++++++
1 file changed, 91 insertions(+)
create mode 100644 pytest_tests/testsuites/services/s3_gate/test_s3_versioning.py
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 00000000..cf43920c
--- /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}"