diff --git a/pytest_tests/helpers/s3_helper.py b/pytest_tests/helpers/s3_helper.py index cb18e87..5f55a66 100644 --- a/pytest_tests/helpers/s3_helper.py +++ b/pytest_tests/helpers/s3_helper.py @@ -2,8 +2,8 @@ import os from typing import Optional import allure - -from steps import s3_gate_bucket, s3_gate_object +import s3_gate_bucket +import s3_gate_object @allure.step("Expected all objects are presented in the bucket") @@ -48,3 +48,45 @@ def set_bucket_versioning(s3_client, bucket: str, status: s3_gate_bucket.Version def object_key_from_file_path(full_path: str) -> str: return os.path.basename(full_path) + + +def assert_tags( + actual_tags: list, expected_tags: Optional[list] = None, unexpected_tags: Optional[list] = None +) -> None: + expected_tags = ( + [{"Key": key, "Value": value} for key, value in expected_tags] if expected_tags else [] + ) + unexpected_tags = ( + [{"Key": key, "Value": value} for key, value in unexpected_tags] if unexpected_tags else [] + ) + if expected_tags == []: + assert not actual_tags, f"Expected there is no tags, got {actual_tags}" + assert len(expected_tags) == len(actual_tags) + for tag in expected_tags: + assert tag in actual_tags, f"Tag {tag} must be in {actual_tags}" + for tag in unexpected_tags: + assert tag not in actual_tags, f"Tag {tag} should not be in {actual_tags}" + + +@allure.step("Expected all tags are presented in object") +def check_tags_by_object( + s3_client, + bucket: str, + key_name: str, + expected_tags: list, + unexpected_tags: Optional[list] = None, +) -> None: + actual_tags = s3_gate_object.get_object_tagging(s3_client, bucket, key_name) + assert_tags( + expected_tags=expected_tags, unexpected_tags=unexpected_tags, actual_tags=actual_tags + ) + + +@allure.step("Expected all tags are presented in bucket") +def check_tags_by_bucket( + s3_client, bucket: str, expected_tags: list, unexpected_tags: Optional[list] = None +) -> None: + actual_tags = s3_gate_bucket.get_bucket_tagging(s3_client, bucket) + assert_tags( + expected_tags=expected_tags, unexpected_tags=unexpected_tags, actual_tags=actual_tags + ) diff --git a/pytest_tests/steps/aws_cli_client.py b/pytest_tests/steps/aws_cli_client.py index 9dbff50..ff77975 100644 --- a/pytest_tests/steps/aws_cli_client.py +++ b/pytest_tests/steps/aws_cli_client.py @@ -149,7 +149,7 @@ class AwsCliClient: for key, value in Metadata.items(): cmd += f" {key}={value}" if Tagging: - cmd += f" --tagging {Tagging}" + cmd += f" --tagging '{Tagging}'" if ACL: cmd += f" --acl {ACL}" if ObjectLockMode: diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py b/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py index 2448bbe..fa79190 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_gate.py @@ -15,6 +15,8 @@ from python_keywords.utility_keywords import ( ) from s3_helper import ( check_objects_in_bucket, + check_tags_by_bucket, + check_tags_by_object, set_bucket_versioning, try_to_get_objects_and_expect_error, ) @@ -344,17 +346,10 @@ class TestS3Gate(TestS3GateBase): key_value_pair = [("some-key", "some-value"), ("some-key-2", "some-value-2")] s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, key_value_pair) - got_tags = s3_gate_bucket.get_bucket_tagging(self.s3_client, bucket) - - with allure.step("Check all tags are presented"): - assert got_tags, f"Expected tags, got {got_tags}" - expected_tags = [{"Key": key, "Value": value} for key, value in key_value_pair] - for tag in expected_tags: - assert tag in got_tags + check_tags_by_bucket(self.s3_client, bucket, key_value_pair) s3_gate_bucket.delete_bucket_tagging(self.s3_client, bucket) - tags = s3_gate_bucket.get_bucket_tagging(self.s3_client, bucket) - assert not tags, f"Expected there is no tags for bucket {bucket}, got {tags}" + check_tags_by_bucket(self.s3_client, bucket, []) @allure.title("Test S3 Object tagging API") def test_s3_api_object_tagging(self, bucket): @@ -376,16 +371,15 @@ class TestS3Gate(TestS3GateBase): for tags in (key_value_pair_obj, key_value_pair_obj_new): s3_gate_object.put_object_tagging(self.s3_client, bucket, obj_key, tags) - - got_tags = s3_gate_object.get_object_tagging(self.s3_client, bucket, obj_key) - assert got_tags, f"Expected tags, got {got_tags}" - expected_tags = [{"Key": key, "Value": value} for key, value in tags] - for tag in expected_tags: - assert tag in got_tags + check_tags_by_object( + self.s3_client, + bucket, + obj_key, + tags, + ) s3_gate_object.delete_object_tagging(self.s3_client, bucket, obj_key) - got_tags = s3_gate_object.get_object_tagging(self.s3_client, bucket, obj_key) - assert not got_tags, f"Expected there is no tags for bucket {bucket}, got {got_tags}" + check_tags_by_object(self.s3_client, bucket, obj_key, []) @allure.title("Test S3: Delete object & delete objects S3 API") def test_s3_api_delete(self, create_buckets): diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py index cf6ea45..a58be58 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py @@ -772,16 +772,6 @@ class TestS3GateObject(TestS3GateBase): file_name = self.object_key_from_file_path(file_path_1) bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True) set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.ENABLED) - with allure.step("Put object with lock-mode"): - with pytest.raises( - Exception, - match=r".*fetch time to epoch: time '0001-01-01T00:00:00Z' must be in the future.*", - ): - # An error occurred (InternalError) when calling the PutObject operation (reached max retries: 2): - # fetch time to epoch: time '0001-01-01T00:00:00Z' must be in the future (after 2022-09-15T08:59:30Z) - s3_gate_object.put_object_s3( - self.s3_client, bucket, file_path_1, ObjectLockMode="COMPLIANCE" - ) with allure.step( "Put object with lock-mode GOVERNANCE lock-retain-until-date +1day, lock-legal-hold-status" @@ -824,8 +814,8 @@ class TestS3GateObject(TestS3GateBase): self.s3_client, bucket, file_name, full_output=True ) assert ( - object_4.get("ObjectLockMode") == "GOVERNANCE" - ), "Expected Object Lock Mode is GOVERNANCE" + object_4.get("ObjectLockMode") == "COMPLIANCE" + ), "Expected Object Lock Mode is COMPLIANCE" assert str(date_obj.strftime("%Y-%m-%dT%H:%M:%S")) in object_4.get( "ObjectLockRetainUntilDate" ), f'Expected Object Lock Retain Until Date is {str(date_obj.strftime("%Y-%m-%dT%H:%M:%S"))}' @@ -850,8 +840,8 @@ class TestS3GateObject(TestS3GateBase): self.s3_client, bucket, file_name, full_output=True ) assert ( - object_4.get("ObjectLockMode") == "GOVERNANCE" - ), "Expected Object Lock Mode is GOVERNANCE" + object_4.get("ObjectLockMode") == "COMPLIANCE" + ), "Expected Object Lock Mode is COMPLIANCE" assert str(date_obj.strftime("%Y-%m-%dT%H:%M:%S")) in object_4.get( "ObjectLockRetainUntilDate" ), f'Expected Object Lock Retain Until Date is {str(date_obj.strftime("%Y-%m-%dT%H:%M:%S"))}' @@ -859,6 +849,17 @@ class TestS3GateObject(TestS3GateBase): object_4.get("ObjectLockLegalHoldStatus") == "ON" ), "Expected Object Lock Legal Hold Status is ON" + with allure.step("Put object with lock-mode"): + with pytest.raises( + Exception, + match=r".*fetch time to epoch: time '0001-01-01T00:00:00Z' must be in the future.*", + ): + # An error occurred (InternalError) when calling the PutObject operation (reached max retries: 2): + # fetch time to epoch: time '0001-01-01T00:00:00Z' must be in the future (after 2022-09-15T08:59:30Z) + s3_gate_object.put_object_s3( + self.s3_client, bucket, file_path_1, ObjectLockMode="COMPLIANCE" + ) + @allure.title("Test S3 Sync directory") @pytest.mark.parametrize("sync_type", ["sync", "cp"]) def test_s3_sync_dir(self, sync_type): diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_tagging.py b/pytest_tests/testsuites/services/s3_gate/test_s3_tagging.py new file mode 100644 index 0000000..75fa53d --- /dev/null +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_tagging.py @@ -0,0 +1,114 @@ +import os +import uuid +from random import choice +from string import ascii_letters +from typing import Tuple + +import allure +import pytest +from python_keywords.utility_keywords import generate_file +from s3_helper import check_tags_by_bucket, check_tags_by_object, object_key_from_file_path + +from steps import s3_gate_bucket, s3_gate_object +from steps.s3_gate_base import TestS3GateBase + + +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 TestS3GateTagging(TestS3GateBase): + @staticmethod + def create_tags(count: int) -> Tuple[list, list]: + tags = [] + for _ in range(count): + tag_key = "".join(choice(ascii_letters) for _ in range(8)) + tag_value = "".join(choice(ascii_letters) for _ in range(12)) + tags.append((tag_key, tag_value)) + return tags + + @allure.title("Test S3: Object tagging") + def test_s3_object_tagging(self): + file_path = generate_file() + file_name = object_key_from_file_path(file_path) + + bucket = s3_gate_bucket.create_bucket_s3(self.s3_client) + + with allure.step("Put with 3 tags object into bucket"): + tag_1 = "Tag1=Value1" + s3_gate_object.put_object_s3(self.s3_client, bucket, file_path, Tagging=tag_1) + got_tags = s3_gate_object.get_object_tagging(self.s3_client, bucket, file_name) + assert got_tags, f"Expected tags, got {got_tags}" + assert got_tags == [{"Key": "Tag1", "Value": "Value1"}], "Tags must be the same" + + with allure.step("Put 10 new tags for object"): + tags_2 = self.create_tags(10) + s3_gate_object.put_object_tagging(self.s3_client, bucket, file_name, tags=tags_2) + check_tags_by_object(self.s3_client, bucket, file_name, tags_2, [("Tag1", "Value1")]) + + with allure.step("Put 10 extra new tags for object"): + tags_3 = self.create_tags(10) + s3_gate_object.put_object_tagging(self.s3_client, bucket, file_name, tags=tags_3) + check_tags_by_object(self.s3_client, bucket, file_name, tags_3, tags_2) + + with allure.step("Copy one object with tag"): + copy_obj_path_1 = s3_gate_object.copy_object_s3( + self.s3_client, bucket, file_name, tagging_directive="COPY" + ) + check_tags_by_object(self.s3_client, bucket, copy_obj_path_1, tags_3, tags_2) + + with allure.step("Put 11 new tags to object and expect an error"): + tags_4 = self.create_tags(11) + with pytest.raises(Exception, match=r".*Object tags cannot be greater than 10*"): + # An error occurred (BadRequest) when calling the PutObjectTagging operation: Object tags cannot be greater than 10 + s3_gate_object.put_object_tagging(self.s3_client, bucket, file_name, tags=tags_4) + + with allure.step("Put empty tag"): + tags_5 = [] + s3_gate_object.put_object_tagging(self.s3_client, bucket, file_name, tags=tags_5) + check_tags_by_object(self.s3_client, bucket, file_name, []) + + with allure.step("Put 10 object tags"): + tags_6 = self.create_tags(10) + s3_gate_object.put_object_tagging(self.s3_client, bucket, file_name, tags=tags_6) + check_tags_by_object(self.s3_client, bucket, file_name, tags_6) + + with allure.step("Delete tags by delete-object-tagging"): + s3_gate_object.delete_object_tagging(self.s3_client, bucket, file_name) + check_tags_by_object(self.s3_client, bucket, file_name, []) + + @allure.title("Test S3: bucket tagging") + def test_s3_bucket_tagging(self): + bucket = s3_gate_bucket.create_bucket_s3(self.s3_client) + + with allure.step("Put 10 bucket tags"): + tags_1 = self.create_tags(10) + s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, tags_1) + check_tags_by_bucket(self.s3_client, bucket, tags_1) + + with allure.step("Put new 10 bucket tags"): + tags_2 = self.create_tags(10) + s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, tags_2) + check_tags_by_bucket(self.s3_client, bucket, tags_2, tags_1) + + with allure.step("Put 11 new tags to bucket and expect an error"): + tags_3 = self.create_tags(11) + with pytest.raises(Exception, match=r".*Object tags cannot be greater than 10.*"): + # An error occurred (BadRequest) when calling the PutBucketTagging operation (reached max retries: 0): Object tags cannot be greater than 10 + s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, tags_3) + + with allure.step("Put empty tag"): + tags_4 = [] + s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, tags_4) + check_tags_by_bucket(self.s3_client, bucket, tags_4) + + with allure.step("Put new 10 bucket tags"): + tags_5 = self.create_tags(10) + s3_gate_bucket.put_bucket_tagging(self.s3_client, bucket, tags_5) + check_tags_by_bucket(self.s3_client, bucket, tags_5, tags_2) + + with allure.step("Delete tags by delete-bucket-tagging"): + s3_gate_bucket.delete_bucket_tagging(self.s3_client, bucket) + check_tags_by_bucket(self.s3_client, bucket, [])