2022-09-23 11:09:41 +00:00
|
|
|
import allure
|
|
|
|
import pytest
|
2023-05-15 09:59:33 +00:00
|
|
|
from frostfs_testlib.s3 import AwsCliClient, Boto3ClientWrapper, S3ClientWrapper
|
|
|
|
from frostfs_testlib.steps.s3 import s3_helper
|
2023-08-02 11:54:03 +00:00
|
|
|
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
|
2023-05-15 09:59:33 +00:00
|
|
|
from frostfs_testlib.utils.file_utils import generate_file
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
|
2022-11-10 05:27:52 +00:00
|
|
|
@pytest.mark.sanity
|
|
|
|
@pytest.mark.acl
|
2022-09-23 11:09:41 +00:00
|
|
|
@pytest.mark.s3_gate
|
2023-05-15 09:59:33 +00:00
|
|
|
class TestS3GateACL:
|
2023-09-08 10:35:34 +00:00
|
|
|
@allure.title("Object ACL (s3_client={s3_client})")
|
2023-05-15 09:59:33 +00:00
|
|
|
@pytest.mark.parametrize("s3_client", [AwsCliClient], indirect=True)
|
2023-08-02 11:54:03 +00:00
|
|
|
def test_s3_object_ACL(
|
|
|
|
self, s3_client: S3ClientWrapper, bucket: str, simple_object_size: ObjectSize
|
|
|
|
):
|
|
|
|
file_path = generate_file(simple_object_size.value)
|
2023-05-15 09:59:33 +00:00
|
|
|
file_name = s3_helper.object_key_from_file_path(file_path)
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
with allure.step("Put object into bucket, Check ACL is empty"):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_object(bucket, file_path)
|
|
|
|
obj_acl = s3_client.get_object_acl(bucket, file_name)
|
2022-09-23 11:09:41 +00:00
|
|
|
assert obj_acl == [], f"Expected ACL is empty, got {obj_acl}"
|
|
|
|
|
|
|
|
with allure.step("Put object ACL = public-read"):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_object_acl(bucket, file_name, "public-read")
|
|
|
|
obj_acl = s3_client.get_object_acl(bucket, file_name)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
with allure.step("Put object ACL = private"):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_object_acl(bucket, file_name, "private")
|
|
|
|
obj_acl = s3_client.get_object_acl(bucket, file_name)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
with allure.step(
|
|
|
|
"Put object with grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
|
|
|
):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_object_acl(
|
2022-09-23 11:09:41 +00:00
|
|
|
bucket,
|
|
|
|
file_name,
|
|
|
|
grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
|
|
|
)
|
2023-05-15 09:59:33 +00:00
|
|
|
obj_acl = s3_client.get_object_acl(bucket, file_name)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
2022-09-23 11:09:41 +00:00
|
|
|
|
2023-09-08 10:35:34 +00:00
|
|
|
@allure.title("Bucket ACL (s3_client={s3_client})")
|
2023-05-15 09:59:33 +00:00
|
|
|
@pytest.mark.parametrize("s3_client", [AwsCliClient, Boto3ClientWrapper], indirect=True)
|
|
|
|
def test_s3_bucket_ACL(self, s3_client: S3ClientWrapper):
|
2022-09-23 11:09:41 +00:00
|
|
|
with allure.step("Create bucket with ACL = public-read-write"):
|
2023-05-15 09:59:33 +00:00
|
|
|
bucket = s3_client.create_bucket(
|
|
|
|
object_lock_enabled_for_bucket=True, acl="public-read-write"
|
|
|
|
)
|
|
|
|
bucket_acl = s3_client.get_bucket_acl(bucket)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
with allure.step("Change bucket ACL to private"):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_bucket_acl(bucket, acl="private")
|
|
|
|
bucket_acl = s3_client.get_bucket_acl(bucket)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser")
|
2022-09-23 11:09:41 +00:00
|
|
|
|
|
|
|
with allure.step(
|
|
|
|
"Change bucket acl to --grant-write uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
|
|
|
):
|
2023-05-15 09:59:33 +00:00
|
|
|
s3_client.put_bucket_acl(
|
2022-09-23 11:09:41 +00:00
|
|
|
bucket,
|
|
|
|
grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
|
|
|
)
|
2023-05-15 09:59:33 +00:00
|
|
|
bucket_acl = s3_client.get_bucket_acl(bucket)
|
|
|
|
s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
|