frostfs-testcases/pytest_tests/testsuites/services/s3_gate/test_s3_locking.py

192 lines
9.8 KiB
Python
Raw Normal View History

import time
from datetime import datetime, timedelta
import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.clients import S3ClientWrapper
from frostfs_testlib.steps 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
@allure.title("[Module] Create bucket with object_lock_enabled_for_bucket")
@pytest.fixture(scope="module")
def bucket_w_lock(s3_client: S3ClientWrapper):
return s3_client.create_bucket(object_lock_enabled_for_bucket=True)
@allure.title("[Module] Create bucket without object_lock_enabled_for_bucket")
@pytest.fixture(scope="module")
def bucket_no_lock(s3_client: S3ClientWrapper):
return s3_client.create_bucket(object_lock_enabled_for_bucket=False)
@pytest.mark.nightly
@pytest.mark.s3_gate
@pytest.mark.s3_gate_locking
@pytest.mark.parametrize("version_id", [None, "second"])
class TestS3GateLocking:
2023-10-31 14:51:09 +00:00
@allure.title("Retention period and legal lock on object (version_id={version_id}, s3_client={s3_client})")
def test_s3_object_locking(self, s3_client: S3ClientWrapper, bucket_w_lock: str, version_id: str, simple_object_size: ObjectSize):
file_path = generate_file(simple_object_size.value)
file_name = s3_helper.object_key_from_file_path(file_path)
retention_period = 2
with reporter.step("Put several versions of object into bucket"):
s3_client.put_object(bucket_w_lock, 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_w_lock, file_name_1)
if version_id:
version_id = version_id_2
with reporter.step(f"Put retention period {retention_period}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period)
retention = {
"Mode": "COMPLIANCE",
"RetainUntilDate": date_obj,
}
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "COMPLIANCE", date_obj, "OFF")
with reporter.step(f"Put legal hold to object {file_name}"):
s3_client.put_object_legal_hold(bucket_w_lock, file_name, "ON", version_id)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "COMPLIANCE", date_obj, "ON")
with reporter.step("Fail with deleting object with legal hold and retention period"):
if version_id:
with pytest.raises(Exception):
# An error occurred (AccessDenied) when calling the DeleteObject operation (reached max retries: 0): Access Denied.
s3_client.delete_object(bucket_w_lock, file_name, version_id)
with reporter.step("Check retention period is no longer set on the uploaded object"):
time.sleep((retention_period + 1) * 60)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "COMPLIANCE", date_obj, "ON")
with reporter.step("Fail with deleting object with legal hold and retention period"):
if version_id:
with pytest.raises(Exception):
# An error occurred (AccessDenied) when calling the DeleteObject operation (reached max retries: 0): Access Denied.
s3_client.delete_object(bucket_w_lock, file_name, version_id)
else:
s3_client.delete_object(bucket_w_lock, file_name, version_id)
2023-10-31 14:51:09 +00:00
@allure.title("Impossible to change retention mode COMPLIANCE (version_id={version_id}, s3_client={s3_client})")
def test_s3_mode_compliance(self, s3_client: S3ClientWrapper, bucket_w_lock: str, version_id: str, simple_object_size: ObjectSize):
file_path = generate_file(simple_object_size.value)
file_name = s3_helper.object_key_from_file_path(file_path)
retention_period = 2
retention_period_1 = 1
with reporter.step("Put object into bucket"):
obj_version = s3_client.put_object(bucket_w_lock, file_path)
if version_id:
version_id = obj_version
with reporter.step(f"Put retention period {retention_period}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period)
retention = {
"Mode": "COMPLIANCE",
"RetainUntilDate": date_obj,
}
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "COMPLIANCE", date_obj, "OFF")
with reporter.step(f"Try to change retention period {retention_period_1}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period_1)
retention = {
"Mode": "COMPLIANCE",
"RetainUntilDate": date_obj,
}
with pytest.raises(Exception):
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
2023-10-31 14:51:09 +00:00
@allure.title("Change retention mode GOVERNANCE (version_id={version_id}, s3_client={s3_client})")
def test_s3_mode_governance(self, s3_client: S3ClientWrapper, bucket_w_lock: str, version_id: str, simple_object_size: ObjectSize):
file_path = generate_file(simple_object_size.value)
file_name = s3_helper.object_key_from_file_path(file_path)
retention_period = 3
retention_period_1 = 2
retention_period_2 = 5
with reporter.step("Put object into bucket"):
obj_version = s3_client.put_object(bucket_w_lock, file_path)
if version_id:
version_id = obj_version
with reporter.step(f"Put retention period {retention_period}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period)
retention = {
"Mode": "GOVERNANCE",
"RetainUntilDate": date_obj,
}
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "GOVERNANCE", date_obj, "OFF")
with reporter.step(f"Try to change retention period {retention_period_1}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period_1)
retention = {
"Mode": "GOVERNANCE",
"RetainUntilDate": date_obj,
}
with pytest.raises(Exception):
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
with reporter.step(f"Try to change retention period {retention_period_1}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period_1)
retention = {
"Mode": "GOVERNANCE",
"RetainUntilDate": date_obj,
}
with pytest.raises(Exception):
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id)
with reporter.step(f"Put new retention period {retention_period_2}min to object {file_name}"):
date_obj = datetime.utcnow() + timedelta(minutes=retention_period_2)
retention = {
"Mode": "GOVERNANCE",
"RetainUntilDate": date_obj,
}
s3_client.put_object_retention(bucket_w_lock, file_name, retention, version_id, True)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "GOVERNANCE", date_obj, "OFF")
@allure.title("[NEGATIVE] Lock object in bucket with disabled locking (version_id={version_id}, s3_client={s3_client})")
def test_s3_legal_hold(self, s3_client: S3ClientWrapper, bucket_no_lock: str, version_id: str, simple_object_size: ObjectSize):
file_path = generate_file(simple_object_size.value)
file_name = s3_helper.object_key_from_file_path(file_path)
with reporter.step("Put object into bucket"):
obj_version = s3_client.put_object(bucket_no_lock, file_path)
if version_id:
version_id = obj_version
with reporter.step(f"Put legal hold to object {file_name}"):
with pytest.raises(Exception):
s3_client.put_object_legal_hold(bucket_no_lock, file_name, "ON", version_id)
@pytest.mark.nightly
@pytest.mark.s3_gate
class TestS3GateLockingBucket:
2023-09-08 10:35:34 +00:00
@allure.title("Bucket Lock (s3_client={s3_client})")
def test_s3_bucket_lock(self, s3_client: S3ClientWrapper, bucket_w_lock: str, simple_object_size: ObjectSize):
file_path = generate_file(simple_object_size.value)
file_name = s3_helper.object_key_from_file_path(file_path)
configuration = {"Rule": {"DefaultRetention": {"Mode": "COMPLIANCE", "Days": 1}}}
with reporter.step("PutObjectLockConfiguration with ObjectLockEnabled=False"):
s3_client.put_object_lock_configuration(bucket_w_lock, configuration)
with reporter.step("PutObjectLockConfiguration with ObjectLockEnabled=True"):
configuration["ObjectLockEnabled"] = "Enabled"
s3_client.put_object_lock_configuration(bucket_w_lock, configuration)
with reporter.step("GetObjectLockConfiguration"):
config = s3_client.get_object_lock_configuration(bucket_w_lock)
configuration["Rule"]["DefaultRetention"]["Years"] = 0
assert config == configuration, f"Configurations must be equal {configuration}"
with reporter.step("Put object into bucket"):
s3_client.put_object(bucket_w_lock, file_path)
s3_helper.assert_object_lock_mode(s3_client, bucket_w_lock, file_name, "COMPLIANCE", None, "OFF", 1)