Skip to content

Commit

Permalink
tests: adapt s3 acl tests to new default behavior (#894)
Browse files Browse the repository at this point in the history 
closes #885
  • Loading branch information
@roman-khimov
roman-khimov authored Dec 4, 2024
2 parents 442a854 + 965d4fc commit a5ff395
Show file tree
Hide file tree
Showing 3 changed files with 135 additions and 0 deletions.
9 changes: 9 additions & 0 deletions pytest_tests/lib/helpers/aws_cli_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,15 @@ def get_object_lock_configuration(self, Bucket):
output = _cmd_run(cmd)
return self._to_json(output)

def put_bucket_ownership_controls(self, Bucket, OwnershipControls):
cmd = (
f"aws {self.common_flags} s3api put-bucket-ownership-controls --bucket {Bucket} "
f"--ownership-controls '{{\"Rules\": [{{\"ObjectOwnership\": \"{OwnershipControls['Rules'][0]['ObjectOwnership']}\"}}]}}' "
f"--endpoint-url {self.s3gate_endpoint}"
)
output = _cmd_run(cmd)
return self._to_json(output)

@staticmethod
def _to_json(output: str) -> dict:
json_output = {}
Expand Down
19 changes: 19 additions & 0 deletions pytest_tests/lib/s3/s3_bucket.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,12 @@ class VersioningStatus(Enum):
SUSPENDED = "Suspended"


class ObjectOwnership(Enum):
BUCKET_OWNER_PREFERRED = "BucketOwnerPreferred"
BUCKET_OWNER_ENFORCED = "BucketOwnerEnforced"
OBJECT_WRITER = "ObjectWriter"


@allure.step("Create bucket S3")
def create_bucket_s3(
s3_client,
Expand Down Expand Up @@ -309,3 +315,16 @@ def delete_bucket_cors(s3_client, bucket: str):
f'Error Message: {err.response["Error"]["Message"]}\n'
f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}'
) from err


def put_bucket_ownership_controls(s3_client, bucket: str, object_ownership: ObjectOwnership):
params = {"Bucket": bucket, "OwnershipControls": {"Rules": [{"ObjectOwnership": object_ownership.value}]}}
try:
response = s3_client.put_bucket_ownership_controls(**params)
log_command_execution("S3 put_bucket_ownership_controls result", response)
return response
except ClientError as err:
raise Exception(
f'Error Message: {err.response["Error"]["Message"]}\n'
f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}'
) from err
107 changes: 107 additions & 0 deletions pytest_tests/tests/s3/test_s3_ACL.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ class TestS3ACL(TestNeofsS3Base):
@pytest.mark.sanity
@allure.title("Test S3: Object ACL")
def test_s3_object_ACL(self, bucket, simple_object_size):
if self.neofs_env.s3_gw._get_version() <= "0.32.0":
pytest.skip("This test runs only on post 0.32.0 S3 gw version")
file_path = generate_file(simple_object_size)
file_name = object_key_from_file_path(file_path)

Expand All @@ -25,6 +27,15 @@ def test_s3_object_ACL(self, bucket, simple_object_size):

with allure.step("Put object ACL = public-read"):
acl = "public-read"
with allure.step("By default ACLs are disabled"):
with pytest.raises(Exception, match=r".*The bucket does not allow ACLs.*"):
s3_object.put_object_acl_s3(self.s3_client, bucket, file_name, acl)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PRIVATE)
with allure.step("Enable ACLs"):
s3_bucket.put_bucket_ownership_controls(
self.s3_client, bucket, s3_bucket.ObjectOwnership.BUCKET_OWNER_PREFERRED
)
s3_object.put_object_acl_s3(self.s3_client, bucket, file_name, acl)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PUBLIC_READ)
Expand All @@ -45,6 +56,48 @@ def test_s3_object_ACL(self, bucket, simple_object_size):
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PUBLIC_READ)

with allure.step("Disable ACL"):
s3_bucket.put_bucket_ownership_controls(
self.s3_client, bucket, s3_bucket.ObjectOwnership.BUCKET_OWNER_ENFORCED
)

with allure.step("Put object ACL = public-read"):
acl = "public-read"
with pytest.raises(Exception, match=r".*The bucket does not allow ACLs.*"):
s3_object.put_object_acl_s3(self.s3_client, bucket, file_name, acl)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PRIVATE)

@allure.title("Test S3: Object eligible ACLs")
def test_s3_object_eligible_acls(self, bucket, simple_object_size):
"""
By default with disabled ACLs, user should be able to set object 'private'
and 'bucket-owner-full-control' ACLs
"""
if self.neofs_env.s3_gw._get_version() <= "0.32.0":
pytest.skip("This test runs only on post 0.32.0 S3 gw version")
file_path = generate_file(simple_object_size)
file_name = object_key_from_file_path(file_path)

with allure.step("Put object into bucket, Check ACL is empty"):
s3_object.put_object_s3(self.s3_client, bucket, file_path)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PRIVATE)

with allure.step("Put object ACL = bucket-owner-full-control"):
acl = "bucket-owner-full-control"
s3_object.put_object_acl_s3(self.s3_client, bucket, file_name, acl)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
assert len(obj_acl) == 1, f"Invalid number of grantee entries for {acl}"
assert obj_acl[0]["Permission"] == "FULL_CONTROL", f"Invalid permissions for {acl}"

with allure.step("Put object ACL = private"):
acl = "private"
s3_object.put_object_acl_s3(self.s3_client, bucket, file_name, acl)
obj_acl = s3_object.get_object_acl_s3(self.s3_client, bucket, file_name)
verify_acls(obj_acl, ACLType.PRIVATE)

@pytest.mark.sanity
@allure.title("Test S3: Bucket ACL")
def test_s3_bucket_ACL(self):
with allure.step("Create bucket with ACL = public-read-write"):
Expand Down Expand Up @@ -72,3 +125,57 @@ def test_s3_bucket_ACL(self):
)
bucket_acl = s3_bucket.get_bucket_acl(self.s3_client, bucket)
verify_acls(bucket_acl, ACLType.PUBLIC_WRITE)

@allure.title("Test S3: Bucket Enable Disable ACL")
def test_s3_bucket_disable_enable_ACL(self):
if self.neofs_env.s3_gw._get_version() <= "0.32.0":
pytest.skip("This test runs only on post 0.32.0 S3 gw version")
with allure.step("Create bucket"):
bucket = s3_bucket.create_bucket_s3(
self.s3_client,
bucket_configuration="rep-1",
)
bucket_acl = s3_bucket.get_bucket_acl(self.s3_client, bucket)
verify_acls(bucket_acl, ACLType.PRIVATE)

with allure.step("Try to change bucket acl to public-read-write"):
acl = "public-read-write"
with pytest.raises(Exception, match=r".*The bucket does not allow ACLs..*"):
s3_bucket.put_bucket_acl_s3(
self.s3_client,
bucket,
acl=acl,
)
bucket_acl = s3_bucket.get_bucket_acl(self.s3_client, bucket)
verify_acls(bucket_acl, ACLType.PRIVATE)

with allure.step("Enable ACLs"):
s3_bucket.put_bucket_ownership_controls(
self.s3_client, bucket, s3_bucket.ObjectOwnership.BUCKET_OWNER_PREFERRED
)

with allure.step("Change bucket acl to public-read-write"):
acl = "public-read-write"
s3_bucket.put_bucket_acl_s3(
self.s3_client,
bucket,
acl=acl,
)
bucket_acl = s3_bucket.get_bucket_acl(self.s3_client, bucket)
verify_acls(bucket_acl, ACLType.PUBLIC_READ_WRITE)

with allure.step("Disable ACL"):
s3_bucket.put_bucket_ownership_controls(
self.s3_client, bucket, s3_bucket.ObjectOwnership.BUCKET_OWNER_ENFORCED
)

with allure.step("Try to change bucket acl to public-read-write"):
acl = "public-read"
with pytest.raises(Exception, match=r".*The bucket does not allow ACLs..*"):
s3_bucket.put_bucket_acl_s3(
self.s3_client,
bucket,
acl=acl,
)
bucket_acl = s3_bucket.get_bucket_acl(self.s3_client, bucket)
verify_acls(bucket_acl, ACLType.PUBLIC_READ_WRITE)

0 comments on commit a5ff395

Please sign in to comment.