[#246] Fix ACL and Policy tests

Signed-off-by: a.berezin <a.berezin@yadro.com>

pytest_tests/testsuites/acl/test_eacl_filters.py

@@ -387,6 +387,7 @@ class TestEACLFilters(ClusterTestBase):
 
     @allure.title("Operations with allow eACL user headers filters (match_type={match_type}, obj_size={object_size})")
     @pytest.mark.parametrize("match_type", [EACLMatchType.STRING_EQUAL, EACLMatchType.STRING_NOT_EQUAL])
+    @pytest.mark.parametrize("object_size", ["simple"])
     def test_extended_acl_allow_filters_object(
         self,
         wallets: Wallets,

pytest_tests/testsuites/conftest.py

@@ -268,12 +268,19 @@ def healthcheck(cluster: Cluster) -> Healthcheck:
     return healthcheck_cls()
 
 
-@pytest.fixture
-def cluster_state_controller(client_shell: Shell, cluster: Cluster, healthcheck: Healthcheck) -> ClusterStateController:
+@pytest.fixture(scope="session")
+def cluster_state_controller_session(
+    client_shell: Shell, cluster: Cluster, healthcheck: Healthcheck
+) -> ClusterStateController:
     controller = ClusterStateController(client_shell, cluster, healthcheck)
-    yield controller
-    controller.start_stopped_hosts()
-    controller.start_all_stopped_services()
+    return controller
+
+
+@pytest.fixture
+def cluster_state_controller(cluster_state_controller_session: ClusterStateController) -> ClusterStateController:
+    yield cluster_state_controller_session
+    cluster_state_controller_session.start_stopped_hosts()
+    cluster_state_controller_session.start_all_stopped_services()
 
 
 @pytest.fixture(scope="session")

pytest_tests/testsuites/container/test_policy.py

@@ -39,10 +39,10 @@ class TestPolicy(ClusterTestBase):
         return True
 
     @pytest.fixture(scope="module")
-    def fill_field_price(self, cluster_state_controller: ClusterStateController):
+    def fill_field_price(self, cluster_state_controller_session: ClusterStateController):
         prices = ["15", "10", "65", "55"]
 
-        config_manager = cluster_state_controller.manager(ConfigStateManager)
+        config_manager = cluster_state_controller_session.manager(ConfigStateManager)
         for i in zip(self.cluster.cluster_nodes, prices):
             config_manager.set_on_node(i[0], StorageNode, {"node:attribute_5": f"Price:{i[1]}"})
 
@@ -51,7 +51,7 @@ class TestPolicy(ClusterTestBase):
 
         yield
 
-        cluster_state_controller.manager(ConfigStateManager).revert_all()
+        cluster_state_controller_session.manager(ConfigStateManager).revert_all()
 
     @allure.title("[NEGATIVE] Placement policy: Can't parse placement policy")
     def test_placement_policy_negative(self, default_wallet):

pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py

@@ -1,6 +1,8 @@
 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
@@ -16,47 +18,31 @@ class TestS3GateACL:
         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, Check ACL is empty"):
+        with reporter.step("Put object into bucket"):
             s3_client.put_object(bucket, file_path)
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            assert obj_acl == [], f"Expected ACL is empty, got {obj_acl}"
 
-        with reporter.step("Put object ACL = public-read"):
-            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")
+        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("Put object ACL = private"):
-            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")
-
-        with reporter.step("Put object with grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"):
-            s3_client.put_object_acl(
-                bucket,
-                file_name,
-                grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
+        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 ACL = public-read-write"):
+        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_acl = s3_client.get_bucket_acl(bucket)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
+            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_acl = s3_client.get_bucket_acl(bucket)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser")
+            bucket_grants = s3_client.get_bucket_acl(bucket)
+            s3_helper.verify_acl_permissions(bucket_grants, PRIVATE_GRANTS)
 
-        with reporter.step("Change bucket acl to --grant-write uri=http://acs.amazonaws.com/groups/global/AllUsers"):
-            s3_client.put_bucket_acl(
-                bucket,
-                grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            bucket_acl = s3_client.get_bucket_acl(bucket)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
+        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)

pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py

@@ -3,6 +3,7 @@ from datetime import datetime, timedelta
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.resources.s3_acl_grants import PRIVATE_GRANTS, PUBLIC_READ_GRANTS, PUBLIC_READ_WRITE_GRANTS
 from frostfs_testlib.s3 import S3ClientWrapper
 from frostfs_testlib.steps.s3 import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
@@ -17,50 +18,18 @@ class TestS3GateBucket:
 
         with reporter.step("Create bucket with ACL private"):
             bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="private")
-            bucket_acl = s3_client.get_bucket_acl(bucket)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser")
+            bucket_grants = s3_client.get_bucket_acl(bucket)
+            s3_helper.verify_acl_permissions(bucket_grants, PRIVATE_GRANTS)
 
-        with reporter.step("Create bucket with ACL = public-read"):
-            bucket_1 = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read")
-            bucket_acl_1 = s3_client.get_bucket_acl(bucket_1)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers")
+        with reporter.step("Create bucket with ACL public-read"):
+            read_bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read")
+            bucket_grants = s3_client.get_bucket_acl(read_bucket)
+            s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_GRANTS)
 
         with reporter.step("Create bucket with ACL public-read-write"):
-            bucket_2 = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read-write")
-            bucket_acl_2 = s3_client.get_bucket_acl(bucket_2)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers")
-
-        with reporter.step("Create bucket with ACL = authenticated-read"):
-            bucket_3 = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="authenticated-read")
-            bucket_acl_3 = s3_client.get_bucket_acl(bucket_3)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl_3, permitted_users="AllUsers")
-
-    @allure.title("Create Bucket with different ACL by grant (s3_client={s3_client})")
-    def test_s3_create_bucket_with_grands(self, s3_client: S3ClientWrapper):
-
-        with reporter.step("Create bucket with  --grant-read"):
-            bucket = s3_client.create_bucket(
-                object_lock_enabled_for_bucket=True,
-                grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            bucket_acl = s3_client.get_bucket_acl(bucket)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
-
-        with reporter.step("Create bucket with --grant-wtite"):
-            bucket_1 = s3_client.create_bucket(
-                object_lock_enabled_for_bucket=True,
-                grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            bucket_acl_1 = s3_client.get_bucket_acl(bucket_1)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers")
-
-        with reporter.step("Create bucket with --grant-full-control"):
-            bucket_2 = s3_client.create_bucket(
-                object_lock_enabled_for_bucket=True,
-                grant_full_control="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            bucket_acl_2 = s3_client.get_bucket_acl(bucket_2)
-            s3_helper.assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers")
+            public_rw_bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read-write")
+            bucket_grants = s3_client.get_bucket_acl(public_rw_bucket)
+            s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_WRITE_GRANTS)
 
     @allure.title("Create bucket with object lock (s3_client={s3_client})")
     def test_s3_bucket_object_lock(self, s3_client: S3ClientWrapper, simple_object_size: ObjectSize):

pytest_tests/testsuites/services/s3_gate/test_s3_object.py

@@ -9,7 +9,8 @@ import allure
 import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.resources.common import ASSETS_DIR, DEFAULT_WALLET_PASS
-from frostfs_testlib.resources.error_patterns import S3_MALFORMED_XML_REQUEST
+from frostfs_testlib.resources.error_patterns import S3_BUCKET_DOES_NOT_ALLOW_ACL, S3_MALFORMED_XML_REQUEST
+from frostfs_testlib.resources.s3_acl_grants import PRIVATE_GRANTS
 from frostfs_testlib.s3 import AwsCliClient, S3ClientWrapper, VersioningStatus
 from frostfs_testlib.steps.s3 import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
@@ -115,20 +116,23 @@ class TestS3GateObject:
 
     @allure.title("Copy with acl (s3_client={s3_client})")
     def test_s3_copy_acl(self, s3_client: S3ClientWrapper, bucket: str, simple_object_size: ObjectSize):
-        version_1_content = "Version 1"
-        file_name_simple = generate_file_with_content(simple_object_size.value, content=version_1_content)
-        obj_key = os.path.basename(file_name_simple)
+        file_path = generate_file_with_content(simple_object_size.value)
+        file_name = os.path.basename(file_path)
 
         s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus.ENABLED)
 
-        with reporter.step("Put several versions of object into bucket"):
-            s3_client.put_object(bucket, file_name_simple)
-            s3_helper.check_objects_in_bucket(s3_client, bucket, [obj_key])
+        with reporter.step("Put object into bucket"):
+            s3_client.put_object(bucket, file_path)
+            s3_helper.check_objects_in_bucket(s3_client, bucket, [file_name])
 
-        with reporter.step("Copy object and check acl attribute"):
-            copy_obj_path = s3_client.copy_object(bucket, obj_key, acl="public-read-write")
-            obj_acl = s3_client.get_object_acl(bucket, copy_obj_path)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
+        with reporter.step("[NEGATIVE] Copy object with public-read-write ACL"):
+            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
+                copy_path = s3_client.copy_object(bucket, file_name, acl="public-read-write")
+
+        with reporter.step("Copy object with private ACL"):
+            copy_path = s3_client.copy_object(bucket, file_name, acl="private")
+            object_grants = s3_client.get_object_acl(bucket, copy_path)
+            s3_helper.verify_acl_permissions(object_grants, PRIVATE_GRANTS)
 
     @allure.title("Copy object with metadata (s3_client={s3_client})")
     def test_s3_copy_metadate(self, s3_client: S3ClientWrapper, bucket: str, simple_object_size: ObjectSize):
@@ -638,71 +642,38 @@ class TestS3GateObject:
         simple_object_size: ObjectSize,
         second_wallet_public_key: str,
     ):
-        file_path_1 = generate_file(complex_object_size.value)
-        file_name = s3_helper.object_key_from_file_path(file_path_1)
-        if bucket_versioning == "ENABLED":
-            status = VersioningStatus.ENABLED
-        elif bucket_versioning == "SUSPENDED":
-            status = VersioningStatus.SUSPENDED
-        s3_helper.set_bucket_versioning(s3_client, bucket, status)
+        file_path = generate_file(complex_object_size.value)
+        file_name = s3_helper.object_key_from_file_path(file_path)
+        s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus[bucket_versioning])
 
         with reporter.step("Put object with acl private"):
-            s3_client.put_object(bucket, file_path_1, acl="private")
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
-            object_1 = s3_client.get_object(bucket, file_name)
-            assert get_file_hash(file_path_1) == get_file_hash(object_1), "Hashes must be the same"
+            s3_client.put_object(bucket, file_path, acl="private")
+            object_grants = s3_client.get_object_acl(bucket, file_name)
+            s3_helper.verify_acl_permissions(object_grants, PRIVATE_GRANTS)
+            object = s3_client.get_object(bucket, file_name)
+            assert get_file_hash(file_path) == get_file_hash(object), "Hashes must be the same"
 
-        with reporter.step("Put object with acl public-read"):
-            file_path_2 = generate_file_with_content(simple_object_size.value, file_path=file_path_1)
-            s3_client.put_object(bucket, file_path_2, acl="public-read")
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
-            object_2 = s3_client.get_object(bucket, file_name)
-            assert get_file_hash(file_path_2) == get_file_hash(object_2), "Hashes must be the same"
+        with reporter.step("[NEGATIVE] Put object with acl public-read"):
+            generate_file_with_content(simple_object_size.value, file_path)
+            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
+                s3_client.put_object(bucket, file_path, acl="public-read")
 
-        with reporter.step("Put object with acl public-read-write"):
-            file_path_3 = generate_file_with_content(simple_object_size.value, file_path=file_path_1)
-            s3_client.put_object(bucket, file_path_3, acl="public-read-write")
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
-            object_3 = s3_client.get_object(bucket, file_name)
-            assert get_file_hash(file_path_3) == get_file_hash(object_3), "Hashes must be the same"
+        with reporter.step("[NEGATIVE] Put object with acl public-read-write"):
+            generate_file_with_content(simple_object_size.value, file_path)
+            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
+                s3_client.put_object(bucket, file_path, acl="public-read-write")
 
-        with reporter.step("Put object with acl authenticated-read"):
-            file_path_4 = generate_file_with_content(simple_object_size.value, file_path=file_path_1)
-            s3_client.put_object(bucket, file_path_4, acl="authenticated-read")
-            obj_acl = s3_client.get_object_acl(bucket, file_name)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
-            object_4 = s3_client.get_object(bucket, file_name)
-            assert get_file_hash(file_path_4) == get_file_hash(object_4), "Hashes must be the same"
+        with reporter.step("[NEGATIVE] Put object with --grant-full-control id=mycanonicaluserid"):
+            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
+                s3_client.put_object(bucket, file_path, grant_full_control=f"id={second_wallet_public_key}")
 
-        file_path_5 = generate_file(complex_object_size.value)
-        file_name_5 = s3_helper.object_key_from_file_path(file_path_5)
-
-        with reporter.step("Put object with --grant-full-control id=mycanonicaluserid"):
-            generate_file_with_content(simple_object_size.value, file_path=file_path_5)
-            s3_client.put_object(
-                bucket,
-                file_path_5,
-                grant_full_control=f"id={second_wallet_public_key}",
-            )
-            obj_acl = s3_client.get_object_acl(bucket, file_name_5)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
-            object_5 = s3_client.get_object(bucket, file_name_5)
-            assert get_file_hash(file_path_5) == get_file_hash(object_5), "Hashes must be the same"
-
-        with reporter.step("Put object with --grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"):
-            generate_file_with_content(simple_object_size.value, file_path=file_path_5)
-            s3_client.put_object(
-                bucket,
-                file_path_5,
-                grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
-            )
-            obj_acl = s3_client.get_object_acl(bucket, file_name_5)
-            s3_helper.assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
-            object_6 = s3_client.get_object(bucket, file_name_5)
-            assert get_file_hash(file_path_5) == get_file_hash(object_6), "Hashes must be the same"
+        with reporter.step(
+            "[NEGATIVE] Put object with --grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"
+        ):
+            with pytest.raises(Exception, match=S3_BUCKET_DOES_NOT_ALLOW_ACL):
+                s3_client.put_object(
+                    bucket, file_path, grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers"
+                )
 
     @allure.title("Put object with lock-mode (s3_client={s3_client})")
     def test_s3_put_object_lock_mode(
@@ -795,21 +766,11 @@ class TestS3GateObject:
         generate_file_with_content(simple_object_size.value, file_path=file_path_1)
         generate_file_with_content(simple_object_size.value, file_path=file_path_2)
         s3_helper.set_bucket_versioning(s3_client, bucket, VersioningStatus.ENABLED)
-        # TODO: return ACL, when https://github.com/nspcc-dev/neofs-s3-gw/issues/685 will be closed
+
         if sync_type == "sync":
-            s3_client.sync(
-                bucket=bucket,
-                dir_path=os.path.dirname(file_path_1),
-                # acl="public-read-write",
-                metadata=object_metadata,
-            )
+            s3_client.sync(bucket=bucket, dir_path=os.path.dirname(file_path_1), metadata=object_metadata)
         elif sync_type == "cp":
-            s3_client.cp(
-                bucket=bucket,
-                dir_path=os.path.dirname(file_path_1),
-                # acl="public-read-write",
-                metadata=object_metadata,
-            )
+            s3_client.cp(bucket=bucket, dir_path=os.path.dirname(file_path_1), metadata=object_metadata)
 
         with reporter.step("Check objects are synced"):
             objects = s3_client.list_objects(bucket)
@@ -823,9 +784,8 @@ class TestS3GateObject:
                 ), "Expected hashes are the same"
                 obj_head = s3_client.head_object(bucket, obj_key)
                 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_client.get_object_acl(bucket, obj_key)
-                # s3_helper.assert_s3_acl(acl_grants = obj_acl, permitted_users = "AllUsers")
+                object_grants = s3_client.get_object_acl(bucket, obj_key)
+                s3_helper.verify_acl_permissions(object_grants, PRIVATE_GRANTS)
 
     @allure.title("Put 10 nested level object (s3_client={s3_client})")
     def test_s3_put_10_folder(