forked from TrueCloudLab/frostfs-testcases
125 lines
7 KiB
Python
125 lines
7 KiB
Python
import os
|
|
|
|
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, get_file_content
|
|
|
|
|
|
@pytest.mark.nightly
|
|
@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("Object versioning (s3_client={s3_client})")
|
|
def test_s3_api_versioning(self, s3_client: S3ClientWrapper, bucket: str, simple_object_size: ObjectSize):
|
|
"""
|
|
Test checks basic versioning functionality for S3 bucket.
|
|
"""
|
|
version_1_content = "Version 1"
|
|
version_2_content = "Version 2"
|
|
file_name_simple = generate_file_with_content(simple_object_size.value, content=version_1_content)
|
|
obj_key = os.path.basename(file_name_simple)
|
|
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_name_simple)
|
|
generate_file_with_content(simple_object_size.value, file_path=file_name_simple, content=version_2_content)
|
|
version_id_2 = s3_client.put_object(bucket, file_name_simple)
|
|
|
|
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") == obj_key}
|
|
assert obj_versions == {
|
|
version_id_1,
|
|
version_id_2,
|
|
}, f"Expected object has versions: {version_id_1, version_id_2}"
|
|
|
|
with reporter.step("Show information about particular version"):
|
|
for version_id in (version_id_1, version_id_2):
|
|
response = s3_client.head_object(bucket, obj_key, version_id=version_id)
|
|
assert "LastModified" in response, "Expected LastModified field"
|
|
assert "ETag" in response, "Expected ETag field"
|
|
assert response.get("VersionId") == version_id, f"Expected VersionId is {version_id}"
|
|
assert response.get("ContentLength") != 0, "Expected ContentLength is not zero"
|
|
|
|
with reporter.step("Check object's attributes"):
|
|
for version_id in (version_id_1, version_id_2):
|
|
got_attrs = s3_client.get_object_attributes(bucket, obj_key, ["ETag"], version_id=version_id)
|
|
if got_attrs:
|
|
assert got_attrs.get("VersionId") == version_id, f"Expected VersionId is {version_id}"
|
|
|
|
with reporter.step("Delete object and check it was deleted"):
|
|
response = s3_client.delete_object(bucket, obj_key)
|
|
version_id_delete = response.get("VersionId")
|
|
|
|
with pytest.raises(Exception, match=r".*Not Found.*"):
|
|
s3_client.head_object(bucket, obj_key)
|
|
|
|
with reporter.step("Get content for all versions and check it is correct"):
|
|
for version, content in (
|
|
(version_id_2, version_2_content),
|
|
(version_id_1, version_1_content),
|
|
):
|
|
file_name = s3_client.get_object(bucket, obj_key, version_id=version)
|
|
got_content = get_file_content(file_name)
|
|
assert got_content == content, f"Expected object content is\n{content}\nGot\n{got_content}"
|
|
|
|
with reporter.step("Restore previous object version"):
|
|
s3_client.delete_object(bucket, obj_key, version_id=version_id_delete)
|
|
|
|
file_name = s3_client.get_object(bucket, obj_key)
|
|
got_content = get_file_content(file_name)
|
|
assert got_content == version_2_content, f"Expected object content is\n{version_2_content}\nGot\n{got_content}"
|
|
|
|
@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}"
|