diff --git a/pytest_tests/testsuites/services/test_s3_gate.py b/pytest_tests/testsuites/services/test_s3_gate.py index 05b4130..805d499 100644 --- a/pytest_tests/testsuites/services/test_s3_gate.py +++ b/pytest_tests/testsuites/services/test_s3_gate.py @@ -61,9 +61,21 @@ class TestS3Gate: def bucket(self): bucket = s3_gate_bucket.create_bucket_s3(self.s3_client) yield bucket - objects = s3_gate_object.list_objects_s3(self.s3_client, bucket) - if objects: - s3_gate_object.delete_objects_s3(self.s3_client, bucket, objects) + + # Delete all objects from bucket + versioning_status = s3_gate_bucket.get_bucket_versioning_status(self.s3_client, bucket) + if versioning_status == s3_gate_bucket.VersioningStatus.ENABLED.value: + # From versioned bucket we should delete all versions of all objects + objects_versions = s3_gate_object.list_objects_versions_s3(self.s3_client, bucket) + if objects_versions: + s3_gate_object.delete_object_versions_s3(self.s3_client, bucket, objects_versions) + else: + # From non-versioned bucket it's sufficient to delete objects by key + objects = s3_gate_object.list_objects_s3(self.s3_client, bucket) + if objects: + s3_gate_object.delete_objects_s3(self.s3_client, bucket, objects) + + # Delete the bucket itself s3_gate_bucket.delete_bucket_s3(self.s3_client, bucket) @allure.title('Test S3 Bucket API') diff --git a/robot/resources/lib/python_keywords/s3_gate_object.py b/robot/resources/lib/python_keywords/s3_gate_object.py index 1388947..93b499b 100644 --- a/robot/resources/lib/python_keywords/s3_gate_object.py +++ b/robot/resources/lib/python_keywords/s3_gate_object.py @@ -137,6 +137,27 @@ def delete_objects_s3(s3_client, bucket: str, object_keys: list): raise Exception(f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}') from err +@keyword('Delete object versions S3') +def delete_object_versions_s3(s3_client, bucket: str, object_versions: list): + try: + # Build deletion list in S3 format + delete_list = { + "Objects": [ + { + "Key": object_version["Key"], + "VersionId": object_version["VersionId"], + } + for object_version in object_versions + ] + } + response = s3_client.delete_objects(Bucket=bucket, Delete=delete_list) + log_command_execution('S3 Delete objects result', response) + return response + + except ClientError as err: + raise Exception(f'Error Message: {err.response["Error"]["Message"]}\n' + f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}') from err + @keyword('Copy object S3') def copy_object_s3(s3_client, bucket, object_key, bucket_dst=None):