import allure import pytest from frostfs_testlib import reporter from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus from frostfs_testlib.steps.s3 import s3_helper from frostfs_testlib.storage.dataclasses.object_size import ObjectSize from frostfs_testlib.utils.file_utils import generate_file, generate_file_with_content @pytest.mark.s3_gate @pytest.mark.s3_gate_versioning class TestS3GateVersioning: @allure.title("Impossible to disable versioning with object_lock (s3_client={s3_client})") def test_s3_version_off(self, s3_client: S3ClientWrapper): bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True) with pytest.raises(Exception): s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus.SUSPENDED) @allure.title("Enable and disable versioning without object_lock (s3_client={s3_client})") def test_s3_version(self, s3_client: S3ClientWrapper, simple_object_size: ObjectSize): file_path = generate_file(simple_object_size.value) file_name = s3_helper.object_key_from_file_path(file_path) bucket_objects = [file_name] bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=False) s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus.SUSPENDED) with reporter.step("Put object into bucket"): s3_client.put_object(bucket, file_path) objects_list = s3_client.list_objects(bucket) assert objects_list == bucket_objects, f"Expected list with single objects in bucket, got {objects_list}" object_version = s3_client.list_objects_versions(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 in list-object-versions, got {object_version}" object_0 = s3_client.head_object(bucket, file_name) assert ( object_0.get("VersionId") == "null" ), f"Expected version is null in head-object, got {object_0.get('VersionId')}" s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus.ENABLED) with reporter.step("Put several versions of object into bucket"): version_id_1 = s3_client.put_object(bucket, file_path) file_name_1 = generate_file_with_content(simple_object_size.value, file_path=file_path) version_id_2 = s3_client.put_object(bucket, file_name_1) with reporter.step("Check bucket shows all versions"): versions = s3_client.list_objects_versions(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 reporter.step("Get object"): object_1 = s3_client.get_object(bucket, file_name, full_output=True) assert object_1.get("VersionId") == version_id_2, f"Get object with version {version_id_2}" with reporter.step("Get first version of object"): object_2 = s3_client.get_object(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 reporter.step("Get second version of object"): object_3 = s3_client.get_object(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}"