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

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.resources.error_patterns import S3_BUCKET_DOES_NOT_ALLOW_ACL
from frostfs_testlib.resources.s3_acl_grants import PRIVATE_GRANTS, PUBLIC_READ_GRANTS, PUBLIC_READ_WRITE_GRANTS
from frostfs_testlib.s3 import AwsCliClient, Boto3ClientWrapper, S3ClientWrapper
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


@pytest.mark.acl
@pytest.mark.s3_gate
class TestS3GateACL:
    @allure.title("Object ACL (s3_client={s3_client})")
    @pytest.mark.parametrize("s3_client", [AwsCliClient], indirect=True)
    def test_s3_object_ACL(self, s3_client: S3ClientWrapper, bucket: 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"):
            s3_client.put_object(bucket, file_path)

        with reporter.step("Verify private ACL is default"):
            object_grants = s3_client.get_object_acl(bucket, file_name)
            s3_helper.verify_acl_permissions(object_grants, PRIVATE_GRANTS)

        with reporter.step("Verify put object ACL is restricted"):
            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
                object_grants = s3_client.put_object_acl(bucket, file_name, acl="public-read")

    @allure.title("Bucket ACL (s3_client={s3_client})")
    @pytest.mark.parametrize("s3_client", [AwsCliClient, Boto3ClientWrapper], indirect=True)
    def test_s3_bucket_ACL(self, s3_client: S3ClientWrapper):
        with reporter.step("Create bucket with public-read-write ACL"):
            bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read-write")
            bucket_grants = s3_client.get_bucket_acl(bucket)
            s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_WRITE_GRANTS)

        with reporter.step("Change bucket ACL to private"):
            s3_client.put_bucket_acl(bucket, acl="private")
            bucket_grants = s3_client.get_bucket_acl(bucket)
            s3_helper.verify_acl_permissions(bucket_grants, PRIVATE_GRANTS)

        with reporter.step("Change bucket ACL to public-read"):
            s3_client.put_bucket_acl(bucket, acl="public-read")
            bucket_grants = s3_client.get_bucket_acl(bucket)
            s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_GRANTS)