From 9395a8003ffea06881b6f66a37097e6fdd09a92e Mon Sep 17 00:00:00 2001 From: Aleskei Chetaev Date: Tue, 14 Feb 2023 09:09:37 +0100 Subject: [PATCH] Add assert_s3_acl Signed-off-by: Elizaveta Chichindaeva --- pytest_tests/helpers/s3_helper.py | 31 +++++++++++ .../services/s3_gate/test_s3_ACL.py | 36 +++---------- .../services/s3_gate/test_s3_bucket.py | 48 +++++------------ .../services/s3_gate/test_s3_object.py | 52 ++++++------------- 4 files changed, 66 insertions(+), 101 deletions(-) diff --git a/pytest_tests/helpers/s3_helper.py b/pytest_tests/helpers/s3_helper.py index 5c0ffe0..013ab39 100644 --- a/pytest_tests/helpers/s3_helper.py +++ b/pytest_tests/helpers/s3_helper.py @@ -1,4 +1,5 @@ import datetime +import logging import os from datetime import datetime, timedelta from typing import Optional @@ -8,6 +9,8 @@ import s3_gate_bucket import s3_gate_object from dateutil.parser import parse +logger = logging.getLogger("NeoLogger") + @allure.step("Expected all objects are presented in the bucket") def check_objects_in_bucket( @@ -127,3 +130,31 @@ def assert_object_lock_mode( assert ( retain_date - last_modify + timedelta(seconds=1) ).days == retain_period, f"Expected retention period is {retain_period} days" + + +def assert_s3_acl(acl_grants: list, permitted_users: str): + if permitted_users == "AllUsers": + grantees = {"AllUsers": 0, "CanonicalUser": 0} + for acl_grant in acl_grants: + if acl_grant.get("Grantee", {}).get("Type") == "Group": + uri = acl_grant.get("Grantee", {}).get("URI") + permission = acl_grant.get("Permission") + assert (uri, permission) == ( + "http://acs.amazonaws.com/groups/global/AllUsers", + "FULL_CONTROL", + ), "All Groups should have FULL_CONTROL" + grantees["AllUsers"] += 1 + if acl_grant.get("Grantee", {}).get("Type") == "CanonicalUser": + permission = acl_grant.get("Permission") + assert permission == "FULL_CONTROL", "Canonical User should have FULL_CONTROL" + grantees["CanonicalUser"] += 1 + assert grantees["AllUsers"] >= 1, "All Users should have FULL_CONTROL" + assert grantees["CanonicalUser"] >= 1, "Canonical User should have FULL_CONTROL" + + if permitted_users == "CanonicalUser": + for acl_grant in acl_grants: + if acl_grant.get("Grantee", {}).get("Type") == "CanonicalUser": + permission = acl_grant.get("Permission") + assert permission == "FULL_CONTROL", "Only CanonicalUser should have FULL_CONTROL" + else: + logger.error("FULL_CONTROL is given to All Users") diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py b/pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py index 188f7b4..e85d931 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py @@ -1,7 +1,7 @@ import allure import pytest from file_helper import generate_file -from s3_helper import object_key_from_file_path +from s3_helper import assert_s3_acl, object_key_from_file_path from steps import s3_gate_bucket, s3_gate_object from steps.s3_gate_base import TestS3GateBase @@ -29,19 +29,12 @@ class TestS3GateACL(TestS3GateBase): with allure.step("Put object ACL = public-read"): s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "public-read") obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") with allure.step("Put object ACL = private"): s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "private") obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - ], "Permission for Canonical User is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser") with allure.step( "Put object with grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers" @@ -53,30 +46,19 @@ class TestS3GateACL(TestS3GateBase): grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") @allure.title("Test S3: Bucket ACL") def test_s3_bucket_ACL(self): with allure.step("Create bucket with ACL = public-read-write"): bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read-write") bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket) - bucket_permission = [permission.get("Permission") for permission in bucket_acl] - assert bucket_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers") with allure.step("Change bucket ACL to private"): s3_gate_bucket.put_bucket_acl_s3(self.s3_client, bucket, acl="private") bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket) - bucket_permission = [permission.get("Permission") for permission in bucket_acl] - assert bucket_permission == [ - "FULL_CONTROL" - ], "Permission for CanonicalUser is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser") with allure.step( "Change bucket acl to --grant-write uri=http://acs.amazonaws.com/groups/global/AllUsers" @@ -87,8 +69,4 @@ class TestS3GateACL(TestS3GateBase): grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket) - bucket_permission = [permission.get("Permission") for permission in bucket_acl] - assert bucket_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers") diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py b/pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py index 2cca521..be9d6c5 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py @@ -3,7 +3,12 @@ from datetime import datetime, timedelta import allure import pytest from file_helper import generate_file -from s3_helper import assert_object_lock_mode, check_objects_in_bucket, object_key_from_file_path +from s3_helper import ( + assert_object_lock_mode, + assert_s3_acl, + check_objects_in_bucket, + object_key_from_file_path, +) from steps import s3_gate_bucket, s3_gate_object from steps.s3_gate_base import TestS3GateBase @@ -24,41 +29,26 @@ class TestS3GateBucket(TestS3GateBase): with allure.step("Create bucket with ACL private"): bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="private") bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket) - bucket_permission = [permission.get("Permission") for permission in bucket_acl] - assert bucket_permission == [ - "FULL_CONTROL" - ], "Permission for CanonicalUser is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser") with allure.step("Create bucket with ACL = public-read"): bucket_1 = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read") bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1) - bucket_permission_1 = [permission.get("Permission") for permission in bucket_acl_1] - assert bucket_permission_1 == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers") with allure.step("Create bucket with ACL public-read-write"): bucket_2 = s3_gate_bucket.create_bucket_s3( self.s3_client, True, acl="public-read-write" ) bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2) - bucket_permission_2 = [permission.get("Permission") for permission in bucket_acl_2] - assert bucket_permission_2 == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for CanonicalUser is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers") with allure.step("Create bucket with ACL = authenticated-read"): bucket_3 = s3_gate_bucket.create_bucket_s3( self.s3_client, True, acl="authenticated-read" ) bucket_acl_3 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_3) - bucket_permission_3 = [permission.get("Permission") for permission in bucket_acl_3] - assert bucket_permission_3 == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl_3, permitted_users="AllUsers") @allure.title("Test S3: Create Bucket with different ACL by grand") def test_s3_create_bucket_with_grands(self): @@ -70,11 +60,7 @@ class TestS3GateBucket(TestS3GateBase): grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket) - bucket_permission = [permission.get("Permission") for permission in bucket_acl] - assert bucket_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for CanonicalUser is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers") with allure.step("Create bucket with --grant-wtite"): bucket_1 = s3_gate_bucket.create_bucket_s3( @@ -83,11 +69,7 @@ class TestS3GateBucket(TestS3GateBase): grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1) - bucket_permission_1 = [permission.get("Permission") for permission in bucket_acl_1] - assert bucket_permission_1 == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers") with allure.step("Create bucket with --grant-full-control"): bucket_2 = s3_gate_bucket.create_bucket_s3( @@ -96,11 +78,7 @@ class TestS3GateBucket(TestS3GateBase): grant_full_control="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2) - bucket_permission_2 = [permission.get("Permission") for permission in bucket_acl_2] - assert bucket_permission_2 == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for CanonicalUser is FULL_CONTROL" + assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers") @allure.title("Test S3: create bucket with object lock") def test_s3_bucket_object_lock(self, simple_object_size): diff --git a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py index 16f0f03..ed6cefe 100644 --- a/pytest_tests/testsuites/services/s3_gate/test_s3_object.py +++ b/pytest_tests/testsuites/services/s3_gate/test_s3_object.py @@ -12,7 +12,12 @@ from data_formatters import get_wallet_public_key from file_helper import concat_files, generate_file, generate_file_with_content, get_file_hash from neofs_testlib.utils.wallet import init_wallet from python_keywords.payment_neogo import deposit_gas, transfer_gas -from s3_helper import assert_object_lock_mode, check_objects_in_bucket, set_bucket_versioning +from s3_helper import ( + assert_object_lock_mode, + assert_s3_acl, + check_objects_in_bucket, + set_bucket_versioning, +) from steps import s3_gate_bucket, s3_gate_object from steps.s3_gate_base import TestS3GateBase @@ -131,10 +136,7 @@ class TestS3GateObject(TestS3GateBase): self.s3_client, bucket, obj_key, ACL="public-read-write" ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, copy_obj_path) - for control in obj_acl: - assert ( - control.get("Permission") == "FULL_CONTROL" - ), "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser") @allure.title("Test S3: Copy object with metadata") def test_s3_copy_metadate(self, bucket, simple_object_size): @@ -703,8 +705,7 @@ class TestS3GateObject(TestS3GateBase): with allure.step("Put object with acl private"): s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_1, ACL="private") obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == ["FULL_CONTROL"], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser") object_1 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name) assert get_file_hash(file_path_1) == get_file_hash(object_1), "Hashes must be the same" @@ -712,11 +713,7 @@ class TestS3GateObject(TestS3GateBase): file_path_2 = generate_file_with_content(simple_object_size, file_path=file_path_1) s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_2, ACL="public-read") obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") object_2 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name) assert get_file_hash(file_path_2) == get_file_hash(object_2), "Hashes must be the same" @@ -726,11 +723,7 @@ class TestS3GateObject(TestS3GateBase): self.s3_client, bucket, file_path_3, ACL="public-read-write" ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") object_3 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name) assert get_file_hash(file_path_3) == get_file_hash(object_3), "Hashes must be the same" @@ -740,11 +733,7 @@ class TestS3GateObject(TestS3GateBase): self.s3_client, bucket, file_path_4, ACL="authenticated-read" ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name) assert get_file_hash(file_path_4) == get_file_hash(object_4), "Hashes must be the same" @@ -760,11 +749,7 @@ class TestS3GateObject(TestS3GateBase): GrantFullControl=f"id={self.other_public_key}", ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser") object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5) assert get_file_hash(file_path_5) == get_file_hash(object_4), "Hashes must be the same" @@ -779,11 +764,7 @@ class TestS3GateObject(TestS3GateBase): GrantRead="uri=http://acs.amazonaws.com/groups/global/AllUsers", ) obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5) - obj_permission = [permission.get("Permission") for permission in obj_acl] - assert obj_permission == [ - "FULL_CONTROL", - "FULL_CONTROL", - ], "Permission for all groups is FULL_CONTROL" + assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers") object_7 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5) assert get_file_hash(file_path_7) == get_file_hash(object_7), "Hashes must be the same" @@ -913,12 +894,9 @@ class TestS3GateObject(TestS3GateBase): assert ( obj_head.get("Metadata") == object_metadata ), f"Metadata of object is {object_metadata}" + # Uncomment after https://github.com/nspcc-dev/neofs-s3-gw/issues/685 is solved # obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, obj_key) - # obj_permission = [permission.get("Permission") for permission in obj_acl] - # assert obj_permission == [ - # "FULL_CONTROL", - # "FULL_CONTROL", - # ], "Permission for all groups is FULL_CONTROL" + # assert_s3_acl(acl_grants = obj_acl, permitted_users = "AllUsers") @allure.title("Test S3 Put 10 nested level object") def test_s3_put_10_folder(self, bucket, temp_directory, simple_object_size):