[#297] Refactore APE tests

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

pytest.ini

@@ -13,6 +13,8 @@ markers =
 #   controlling markers
     order: manual control of test order
     logs_after_session: Make the last test in session
+#   parametrizing markers
+    container: specify container details for container creation
 #   functional markers
     maintenance: tests for change mode node
     container: tests for container creation

pytest_tests/helpers/container_spec.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+from frostfs_testlib.steps.cli.container import DEFAULT_PLACEMENT_RULE
+from frostfs_testlib.storage.cluster import Cluster
+
+
+@dataclass
+class ContainerSpec:
+    rule: str = DEFAULT_PLACEMENT_RULE
+    basic_acl: str = None
+    allow_owner_via_ape: bool = False
+
+    def parsed_rule(self, cluster: Cluster):
+        if self.rule is None:
+            return None
+
+        substitutions = {"%NODE_COUNT%": str(len(cluster.cluster_nodes))}
+
+        parsed_rule = self.rule
+        for sub, replacement in substitutions.items():
+            parsed_rule = parsed_rule.replace(sub, replacement)
+
+        return parsed_rule

pytest_tests/testsuites/access/acl/test_acl.py

@@ -3,7 +3,6 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.resources.wellknown_acl import PRIVATE_ACL_F, PUBLIC_ACL_F, READONLY_ACL_F
 from frostfs_testlib.shell import Shell
-from frostfs_testlib.steps.cli.container import create_container
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
 from frostfs_testlib.storage.cluster import Cluster
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
@@ -11,39 +10,21 @@ from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 
 from pytest_tests.helpers.container_access import assert_full_access_to_container, assert_no_access_to_container, assert_read_only_container
 
+from ....helpers.container_spec import ContainerSpec
+
 
 @pytest.mark.sanity
 @pytest.mark.smoke
 @pytest.mark.acl
 class TestACLBasic(ClusterTestBase):
-    @pytest.fixture(scope="module")
-    def public_container(self, default_wallet: WalletInfo):
-        with reporter.step("Create public container"):
-            cid_public = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl=PUBLIC_ACL_F)
-
-        return cid_public
-
-    @pytest.fixture(scope="module")
-    def private_container(self, default_wallet: WalletInfo):
-        with reporter.step("Create private container"):
-            cid_private = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl=PRIVATE_ACL_F)
-
-        return cid_private
-
-    @pytest.fixture(scope="module")
-    def readonly_container(self, default_wallet: WalletInfo):
-        with reporter.step("Create public readonly container"):
-            cid_read_only = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl=READONLY_ACL_F)
-
-        return cid_read_only
-
     @allure.title("Operations in public container available to everyone (obj_size={object_size})")
+    @pytest.mark.container(ContainerSpec(basic_acl=PUBLIC_ACL_F))
     def test_basic_acl_public(
         self,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
         client_shell: Shell,
-        public_container: str,
+        container: str,
         file_path: str,
         cluster: Cluster,
     ):
@@ -58,7 +39,7 @@ class TestACLBasic(ClusterTestBase):
                 owner_object_oid = put_object_to_random_node(
                     default_wallet,
                     file_path,
-                    public_container,
+                    container,
                     shell=self.shell,
                     cluster=self.cluster,
                     attributes={"created": "owner"},
@@ -66,23 +47,24 @@ class TestACLBasic(ClusterTestBase):
                 other_object_oid = put_object_to_random_node(
                     other_wallet,
                     file_path,
-                    public_container,
+                    container,
                     shell=self.shell,
                     cluster=self.cluster,
                     attributes={"created": "other"},
                 )
 
             with reporter.step(f"Check {role} has full access to public container"):
-                assert_full_access_to_container(wallet, public_container, owner_object_oid, file_path, client_shell, cluster)
+                assert_full_access_to_container(wallet, container, owner_object_oid, file_path, client_shell, cluster)
-                assert_full_access_to_container(wallet, public_container, other_object_oid, file_path, client_shell, cluster)
+                assert_full_access_to_container(wallet, container, other_object_oid, file_path, client_shell, cluster)
 
     @allure.title("Operations in private container only available to owner (obj_size={object_size})")
+    @pytest.mark.container(ContainerSpec(basic_acl=PRIVATE_ACL_F))
     def test_basic_acl_private(
         self,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
         client_shell: Shell,
-        private_container: str,
+        container: str,
         file_path: str,
         cluster: Cluster,
     ):
@@ -91,21 +73,22 @@ class TestACLBasic(ClusterTestBase):
         """
 
         with reporter.step("Put object to container"):
-            owner_object_oid = put_object_to_random_node(default_wallet, file_path, private_container, client_shell, cluster)
+            owner_object_oid = put_object_to_random_node(default_wallet, file_path, container, client_shell, cluster)
 
         with reporter.step("Check no one except owner has access to operations with container"):
-            assert_no_access_to_container(other_wallet, private_container, owner_object_oid, file_path, client_shell, cluster)
+            assert_no_access_to_container(other_wallet, container, owner_object_oid, file_path, client_shell, cluster)
 
         with reporter.step("Check owner has full access to private container"):
-            assert_full_access_to_container(default_wallet, private_container, owner_object_oid, file_path, self.shell, cluster)
+            assert_full_access_to_container(default_wallet, container, owner_object_oid, file_path, self.shell, cluster)
 
     @allure.title("Read operations in readonly container available to others (obj_size={object_size})")
+    @pytest.mark.container(ContainerSpec(basic_acl=READONLY_ACL_F))
     def test_basic_acl_readonly(
         self,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
         client_shell: Shell,
-        readonly_container: str,
+        container: str,
         file_path: str,
         cluster: Cluster,
     ):
@@ -114,10 +97,10 @@ class TestACLBasic(ClusterTestBase):
         """
 
         with reporter.step("Put object to container"):
-            object_oid = put_object_to_random_node(default_wallet, file_path, readonly_container, client_shell, cluster)
+            object_oid = put_object_to_random_node(default_wallet, file_path, container, client_shell, cluster)
 
         with reporter.step("Check others has read-only access to operations with container"):
-            assert_read_only_container(other_wallet, readonly_container, object_oid, file_path, client_shell, cluster)
+            assert_read_only_container(other_wallet, container, object_oid, file_path, client_shell, cluster)
 
         with reporter.step("Check owner has full access to public container"):
-            assert_full_access_to_container(default_wallet, readonly_container, object_oid, file_path, client_shell, cluster)
+            assert_full_access_to_container(default_wallet, container, object_oid, file_path, client_shell, cluster)

pytest_tests/testsuites/access/ape/test_ape.py

@@ -3,7 +3,6 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.resources.wellknown_acl import PUBLIC_ACL
-from frostfs_testlib.steps.cli.container import create_container
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
 from frostfs_testlib.steps.node_management import drop_object
 from frostfs_testlib.storage.dataclasses import ape
@@ -20,67 +19,62 @@ from pytest_tests.helpers.container_access import (
     assert_no_access_to_container,
 )
 
+from ....helpers.container_spec import ContainerSpec
+
+
+@pytest.fixture
+def denied_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.Role) -> WalletInfo:
+    return other_wallet if role == ape.Role.OTHERS else default_wallet
+
+
+@pytest.fixture
+def allowed_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.Role) -> WalletInfo:
+    return default_wallet if role == ape.Role.OTHERS else other_wallet
+
 
 @pytest.mark.ape
 class TestApeContainer(ClusterTestBase):
-    @pytest.fixture(scope="function")
-    def full_placement_container_with_object(self, default_wallet: WalletInfo, file_path: str) -> tuple[str, str, str]:
-        storage_nodes = self.cluster.storage_nodes
-        node_count = len(storage_nodes)
-        with reporter.step("Create public container with full placement rule"):
-            full_placement_rule = f"REP {node_count} IN X CBF 1 SELECT {node_count} FROM * AS X"
-            cid = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, full_placement_rule, PUBLIC_ACL)
-
-        with reporter.step("Put object to container"):
-            oid = put_object_to_random_node(default_wallet, file_path, cid, shell=self.shell, cluster=self.cluster)
-            wait_object_replication(cid, oid, node_count, shell=self.shell, nodes=storage_nodes)
-
-        yield cid, oid
-
     @pytest.mark.sanity
     @allure.title("Deny operations via APE by role (role={role}, obj_size={object_size})")
     @pytest.mark.parametrize("role", [ape.Role.OWNER, ape.Role.OTHERS], indirect=True)
     def test_deny_operations_via_ape_by_role(
         self,
-        default_wallet: WalletInfo,
+        denied_wallet: WalletInfo,
-        other_wallet: WalletInfo,
+        allowed_wallet: WalletInfo,
         frostfs_cli: FrostfsCli,
-        container_with_objects: tuple[str, list[str], str],
+        container: str,
+        objects: list[str],
         role: ape.Role,
         file_path: TestFile,
+        rpc_endpoint: str,
     ):
-        denied_wallet = other_wallet if role == ape.Role.OTHERS else default_wallet
-        allowed_wallet = default_wallet if role == ape.Role.OTHERS else other_wallet
-        allowed_role = ape.Role.OWNER if role == ape.Role.OTHERS else ape.Role.OTHERS
-        cid, object_oids, file_path = container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
-        role_condition = ape.Condition.by_role(role.value)
-
         with reporter.step(f"Deny all operations for {role} via APE"):
-            deny_rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, role_condition)
+            deny_rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, ape.Condition.by_role(role.value))
-            frostfs_cli.ape_manager.add(endpoint, deny_rule.chain_id, target_name=cid, target_type="container", rule=deny_rule.as_string())
+            frostfs_cli.ape_manager.add(
+                rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container", rule=deny_rule.as_string()
+            )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
-        with reporter.step(f"Assert {role} have no access to public container"):
+        with reporter.step(f"Assert denied role have no access to public container"):
-            assert_no_access_to_container(denied_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
+            assert_no_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
-        with reporter.step(f"Assert {allowed_role} have full access to public container"):
+        with reporter.step(f"Assert allowed role have full access to public container"):
-            assert_full_access_to_container(allowed_wallet, cid, object_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
         with reporter.step(f"Remove deny rule from APE"):
-            frostfs_cli.ape_manager.remove(endpoint, deny_rule.chain_id, target_name=cid, target_type="container")
+            frostfs_cli.ape_manager.remove(rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container")
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
-        with reporter.step("Assert owner have full access to public container"):
+        with reporter.step("Assert allowed role have full access to public container"):
-            assert_full_access_to_container(default_wallet, cid, object_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
-        with reporter.step("Assert others have full access to public container"):
+        with reporter.step("Assert denied role have full access to public container"):
-            assert_full_access_to_container(other_wallet, cid, object_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
     @allure.title("Deny operations for others via APE excluding single pubkey (obj_size={object_size})")
     def test_deny_opeartions_excluding_pubkey(
@@ -89,11 +83,11 @@ class TestApeContainer(ClusterTestBase):
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
         other_wallet_2: WalletInfo,
-        container_with_objects: tuple[str, list[str], str],
+        container: str,
+        objects: list[str],
+        rpc_endpoint: str,
+        file_path: TestFile,
     ):
-        endpoint = self.cluster.default_rpc_endpoint
-        cid, object_oids, file_path = container_with_objects
-
         with reporter.step("Add deny APE rules for others except single wallet"):
             rule_conditions = [
                 ape.Condition.by_role(ape.Role.OTHERS),
@@ -103,29 +97,36 @@ class TestApeContainer(ClusterTestBase):
                 ),
             ]
             rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, rule_conditions)
-            frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step("Assert others have no access to public container"):
-            assert_no_access_to_container(other_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
+            assert_no_access_to_container(other_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
         with reporter.step("Assert owner have full access to public container"):
-            assert_full_access_to_container(default_wallet, cid, object_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(default_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
         with reporter.step("Assert allowed wallet have full access to public container"):
-            assert_full_access_to_container(other_wallet_2, cid, object_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(other_wallet_2, container, objects.pop(), file_path, self.shell, self.cluster)
 
     @allure.title("Replication works with APE deny rules on OWNER and OTHERS (obj_size={object_size})")
+    @pytest.mark.container(ContainerSpec(f"REP %NODE_COUNT% IN X CBF 1 SELECT %NODE_COUNT% FROM * AS X", PUBLIC_ACL))
     def test_replication_works_with_deny_rules(
         self,
+        default_wallet: WalletInfo,
         frostfs_cli: FrostfsCli,
-        full_placement_container_with_object: tuple[str, list[str], str],
+        container: str,
+        rpc_endpoint: str,
+        file_path: TestFile,
     ):
-        endpoint = self.cluster.default_rpc_endpoint
+        with reporter.step("Put object to container"):
-        cid, oid = full_placement_container_with_object
+            oid = put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster)
-        storage_nodes = self.cluster.storage_nodes
+
+        with reporter.step("Wait for object replication after upload"):
+            wait_object_replication(container, oid, len(self.cluster.cluster_nodes), self.shell, self.cluster.storage_nodes)
 
         with reporter.step("Add deny APE rules for owner and others"):
             rule_conditions = [
@@ -134,24 +135,23 @@ class TestApeContainer(ClusterTestBase):
             ]
             for rule_condition in rule_conditions:
                 rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, rule_condition)
-                frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+                frostfs_cli.ape_manager.add(
+                    rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string()
+                )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step("Drop object"):
-            drop_object(storage_nodes[0], cid, oid)
+            drop_object(self.cluster.storage_nodes[0], container, oid)
 
         with reporter.step("Wait for dropped object to be replicated"):
-            wait_object_replication(cid, oid, len(storage_nodes), self.shell, storage_nodes)
+            wait_object_replication(container, oid, len(self.cluster.storage_nodes), self.shell, self.cluster.storage_nodes)
 
     @allure.title("Deny operations via APE by role (role=ir, obj_size={object_size})")
     def test_deny_operations_via_ape_by_role_ir(
-        self, frostfs_cli: FrostfsCli, ir_wallet: WalletInfo, container_with_objects: tuple[str, list[str], str]
+        self, frostfs_cli: FrostfsCli, ir_wallet: WalletInfo, container: str, objects: list[str], rpc_endpoint: str, file_path: TestFile
     ):
-        cid, object_oids, file_path = container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
         default_ir_access = {
             ape.ObjectOperations.PUT: False,
             ape.ObjectOperations.GET: True,
@@ -163,36 +163,38 @@ class TestApeContainer(ClusterTestBase):
         }
 
         with reporter.step("Assert IR wallet access in default state"):
-            assert_access_to_container(default_ir_access, ir_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
         with reporter.step("Add deny APE rule with deny all operations for IR role"):
             rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, [ape.Condition.by_role(ape.Role.IR.value)])
-            frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step("Assert IR wallet ignores APE rules"):
-            assert_access_to_container(default_ir_access, ir_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
         with reporter.step("Remove APE rule"):
-            frostfs_cli.ape_manager.remove(endpoint, rule.chain_id, target_name=cid, target_type="container")
+            frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step("Assert IR wallet access is restored"):
-            assert_access_to_container(default_ir_access, ir_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
     @allure.title("Deny operations via APE by role (role=container, obj_size={object_size})")
     def test_deny_operations_via_ape_by_role_container(
-        self, frostfs_cli: FrostfsCli, storage_wallet: WalletInfo, container_with_objects: tuple[str, list[str], str]
+        self,
+        frostfs_cli: FrostfsCli,
+        container_node_wallet: WalletInfo,
+        container: str,
+        objects: list[str],
+        rpc_endpoint: str,
+        file_path: TestFile,
     ):
-
+        access_matrix = {
-        cid, object_oids, file_path = container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
-        default_container_access = {
             ape.ObjectOperations.PUT: True,
             ape.ObjectOperations.GET: True,
             ape.ObjectOperations.HEAD: True,
@@ -203,24 +205,24 @@ class TestApeContainer(ClusterTestBase):
         }
 
         with reporter.step("Assert CONTAINER wallet access in default state"):
-            assert_access_to_container(default_container_access, storage_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
         rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, ape.Condition.by_role(ape.Role.CONTAINER.value))
 
         with reporter.step(f"Add APE rule with deny all operations for CONTAINER and IR roles"):
-            frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step("Assert CONTAINER wallet ignores APE rule"):
-            assert_access_to_container(default_container_access, storage_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)
 
-        with reporter.step("Remove APE rules"):
+        with reporter.step("Remove APE rule"):
-            frostfs_cli.ape_manager.remove(endpoint, rule.chain_id, target_name=cid, target_type="container")
+            frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
-        with reporter.step("Assert CONTAINER wallet access is restored"):
+        with reporter.step("Assert CONTAINER wallet access after rule was removed"):
-            assert_access_to_container(default_container_access, storage_wallet, cid, object_oids[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)

pytest_tests/testsuites/access/ape/test_ape_filters.py

@@ -2,11 +2,8 @@ import allure
 import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
-from frostfs_testlib.resources.error_patterns import OBJECT_ACCESS_DENIED, OBJECT_NOT_FOUND
+from frostfs_testlib.resources.error_patterns import OBJECT_ACCESS_DENIED
-from frostfs_testlib.resources.wellknown_acl import PUBLIC_ACL
-from frostfs_testlib.steps.cli.container import create_container
 from frostfs_testlib.steps.cli.object import get_object_from_random_node, head_object, put_object_to_random_node
-from frostfs_testlib.storage.cluster import Cluster
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
@@ -23,6 +20,8 @@ from pytest_tests.helpers.container_access import (
 )
 from pytest_tests.helpers.object_access import OBJECT_NO_ACCESS
 
+from ....helpers.container_spec import ContainerSpec
+
 
 @pytest.mark.ape
 class TestApeFilters(ClusterTestBase):
@@ -47,61 +46,27 @@ class TestApeFilters(ClusterTestBase):
         ape.ObjectOperations.PUT,
     ]
 
-    @pytest.fixture(scope="function")
+    @pytest.fixture
-    def public_container_with_objects(self, default_wallet: WalletInfo, file_path: TestFile):
+    def objects_with_attributes(self, default_wallet: WalletInfo, file_path: TestFile, container: str):
-        with reporter.step("Create public container"):
+        return [
-            cid = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl=PUBLIC_ACL)
+            put_object_to_random_node(
-
+                default_wallet, file_path, container, self.shell, self.cluster, attributes={**self.ATTRIBUTES, "key": val}
-        objects_with_header, objects_with_other_header, objects_without_header = self._fill_container(default_wallet, file_path, cid)
-
-        return cid, objects_with_header, objects_with_other_header, objects_without_header, file_path
-
-    @pytest.fixture(scope="function")
-    def private_container(self, default_wallet: WalletInfo, frostfs_cli: FrostfsCli, cluster: Cluster):
-        with reporter.step("Create private container"):
-            cid = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl="0")
-
-        with reporter.step("Create allow APE rule for container owner"):
-            role_condition = ape.Condition.by_role(ape.Role.OWNER)
-            deny_rule = ape.Rule(ape.Verb.ALLOW, ape.ObjectOperations.WILDCARD_ALL, role_condition)
-
-            frostfs_cli.ape_manager.add(
-                cluster.default_rpc_endpoint,
-                deny_rule.chain_id,
-                target_name=cid,
-                target_type="container",
-                rule=deny_rule.as_string(),
             )
-
-        with reporter.step("Wait for one block"):
-            self.wait_for_blocks()
-
-        return cid
-
-    @pytest.fixture(scope="function")
-    def container_with_objects(self, private_container: str, default_wallet: WalletInfo, file_path: TestFile):
-        objects_with_header, objects_with_other_header, objects_without_header = self._fill_container(
-            default_wallet, file_path, private_container
-        )
-        return private_container, objects_with_header, objects_with_other_header, objects_without_header, file_path
-
-    @reporter.step("Add objects to container")
-    def _fill_container(self, wallet: WalletInfo, test_file: TestFile, cid: str):
-        objects_with_header = [
-            put_object_to_random_node(wallet, test_file, cid, self.shell, self.cluster, attributes={**self.ATTRIBUTES, "key": val})
             for val in range(self.OBJECT_COUNT)
         ]
 
-        objects_with_other_header = [
+    @pytest.fixture
-            put_object_to_random_node(wallet, test_file, cid, self.shell, self.cluster, attributes={**self.OTHER_ATTRIBUTES, "key": val})
+    def objects_with_other_attributes(self, default_wallet: WalletInfo, file_path: TestFile, container: str):
+        return [
+            put_object_to_random_node(
+                default_wallet, file_path, container, self.shell, self.cluster, attributes={**self.OTHER_ATTRIBUTES, "key": val}
+            )
             for val in range(self.OBJECT_COUNT)
         ]
 
-        objects_without_header = [
+    @pytest.fixture
-            put_object_to_random_node(wallet, test_file, cid, self.shell, self.cluster) for _ in range(self.OBJECT_COUNT)
+    def objects_without_attributes(self, default_wallet: WalletInfo, file_path: TestFile, container: str):
-        ]
+        return [put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster) for _ in range(self.OBJECT_COUNT)]
-
-        return objects_with_header, objects_with_other_header, objects_without_header
 
     @pytest.mark.sanity
     @allure.title("Operations with request filter (match_type={match_type}, obj_size={object_size})")
@@ -112,24 +77,22 @@ class TestApeFilters(ClusterTestBase):
         frostfs_cli: FrostfsCli,
         temp_directory: str,
         other_wallet: WalletInfo,
-        public_container_with_objects: tuple[str, list[str], str],
+        container: str,
+        objects_with_attributes: list[str],
+        objects_with_other_attributes: list[str],
+        objects_without_attributes: list[str],
         match_type: ape.MatchType,
+        file_path: TestFile,
+        rpc_endpoint: str,
     ):
-        (
-            cid,
-            objects_with_header,
-            objects_with_other_header,
-            objects_without_header,
-            file_path,
-        ) = public_container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
         with reporter.step("Deny all operations for others via APE with request condition"):
             request_condition = ape.Condition('"frostfs:xheader/check_key"', '"check_value"', ape.ConditionType.REQUEST, match_type)
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             deny_rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, [request_condition, role_condition])
 
-            frostfs_cli.ape_manager.add(endpoint, deny_rule.chain_id, target_name=cid, target_type="container", rule=deny_rule.as_string())
+            frostfs_cli.ape_manager.add(
+                rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container", rule=deny_rule.as_string()
+            )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
@@ -137,7 +100,7 @@ class TestApeFilters(ClusterTestBase):
         with reporter.step("Create bearer token with everything allowed for others role"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, role_condition)
-            bearer = create_bearer_token(frostfs_cli, temp_directory, cid, rule, endpoint)
+            bearer = create_bearer_token(frostfs_cli, temp_directory, container, rule, rpc_endpoint)
 
         # Filter denies requests where "check_key {match_type} ATTRIBUTE", so when match_type
         # is STRING_EQUAL, then requests with "check_key=OTHER_ATTRIBUTE" will be allowed while
@@ -147,18 +110,22 @@ class TestApeFilters(ClusterTestBase):
 
         # We test on 3 groups of objects with various headers,
         # but APE rule should ignore object headers and only work based on request headers
-        for oids in [objects_with_header, objects_with_other_header, objects_without_header]:
+        for oids in [objects_with_attributes, objects_with_other_attributes, objects_without_attributes]:
             with reporter.step("Check others has full access when sending request without headers"):
-                assert_full_access_to_container(other_wallet, cid, oids.pop(), file_path, self.shell, self.cluster)
+                assert_full_access_to_container(other_wallet, container, oids.pop(), file_path, self.shell, self.cluster)
 
             with reporter.step("Check others has full access when sending request with allowed headers"):
-                assert_full_access_to_container(other_wallet, cid, oids.pop(), file_path, self.shell, self.cluster, xhdr=allow_headers)
+                assert_full_access_to_container(
+                    other_wallet, container, oids.pop(), file_path, self.shell, self.cluster, xhdr=allow_headers
+                )
 
             with reporter.step("Check others has no access when sending request with denied headers"):
-                assert_no_access_to_container(other_wallet, cid, oids.pop(), file_path, self.shell, self.cluster, xhdr=deny_headers)
+                assert_no_access_to_container(other_wallet, container, oids.pop(), file_path, self.shell, self.cluster, xhdr=deny_headers)
 
             with reporter.step("Check others has full access when sending request with denied headers and using bearer token"):
-                assert_full_access_to_container(other_wallet, cid, oids.pop(), file_path, self.shell, self.cluster, bearer, deny_headers)
+                assert_full_access_to_container(
+                    other_wallet, container, oids.pop(), file_path, self.shell, self.cluster, bearer, deny_headers
+                )
 
     @allure.title("Operations with deny user headers filter (match_type={match_type}, obj_size={object_size})")
     @pytest.mark.parametrize("match_type", [ape.MatchType.EQUAL, ape.MatchType.NOT_EQUAL])
@@ -168,19 +135,14 @@ class TestApeFilters(ClusterTestBase):
         frostfs_cli: FrostfsCli,
         temp_directory: str,
         other_wallet: WalletInfo,
-        public_container_with_objects: tuple[str, list[str], str],
+        container: str,
+        objects_with_attributes: list[str],
+        objects_with_other_attributes: list[str],
+        objects_without_attributes: list[str],
         match_type: ape.MatchType,
+        rpc_endpoint: str,
+        file_path: TestFile,
     ):
-        # TODO: Refactor this to be fixtures, not test logic
-        (
-            cid,
-            objects_with_attributes,
-            objects_with_other_attributes,
-            objs_without_attributes,
-            file_path,
-        ) = public_container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
         allow_objects = objects_with_other_attributes if match_type == ape.MatchType.EQUAL else objects_with_attributes
         deny_objects = objects_with_attributes if match_type == ape.MatchType.EQUAL else objects_with_other_attributes
 
@@ -210,15 +172,15 @@ class TestApeFilters(ClusterTestBase):
                 ape.ObjectOperations.DELETE: False,  # Because delete needs to put a tombstone without attributes
             },
         }
-        # End of refactor
 
         with reporter.step("Deny operations for others via APE with resource condition"):
             resource_condition = ape.Condition('"check_key"', '"check_value"', ape.ConditionType.RESOURCE, match_type)
-            not_a_tombstone_condition = ape.Condition.by_object_type("TOMBSTONE", ape.ConditionType.RESOURCE, ape.MatchType.NOT_EQUAL)
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             deny_rule = ape.Rule(ape.Verb.DENY, self.RESOURCE_OPERATIONS, [resource_condition, role_condition])
 
-            frostfs_cli.ape_manager.add(endpoint, deny_rule.chain_id, target_name=cid, target_type="container", rule=deny_rule.as_string())
+            frostfs_cli.ape_manager.add(
+                rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container", rule=deny_rule.as_string()
+            )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
@@ -226,12 +188,12 @@ class TestApeFilters(ClusterTestBase):
         with reporter.step("Create bearer token with everything allowed for others role"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, role_condition)
-            bearer = create_bearer_token(frostfs_cli, temp_directory, cid, rule, endpoint)
+            bearer = create_bearer_token(frostfs_cli, temp_directory, container, rule, rpc_endpoint)
 
         with reporter.step("Create bearer token with allowed put for others role"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             rule = ape.Rule(ape.Verb.ALLOW, ape.ObjectOperations.PUT, role_condition)
-            bearer_put = create_bearer_token(frostfs_cli, temp_directory, cid, rule, endpoint)
+            bearer_put = create_bearer_token(frostfs_cli, temp_directory, container, rule, rpc_endpoint)
 
         # We will attempt requests with various headers,
         # but APE rule should ignore request headers and validate only object headers
@@ -240,8 +202,8 @@ class TestApeFilters(ClusterTestBase):
                 assert_access_to_container(
                     no_attributes_access[match_type],
                     other_wallet,
-                    cid,
+                    container,
-                    objs_without_attributes.pop(),
+                    objects_without_attributes.pop(),
                     file_path,
                     self.shell,
                     self.cluster,
@@ -250,52 +212,50 @@ class TestApeFilters(ClusterTestBase):
 
             with reporter.step("Check others have full access to objects without deny attribute"):
                 assert_access_to_container(
-                    allowed_access[match_type], other_wallet, cid, allow_objects.pop(), file_path, self.shell, self.cluster, xhdr=xhdr
+                    allowed_access[match_type], other_wallet, container, allow_objects.pop(), file_path, self.shell, self.cluster, xhdr=xhdr
                 )
 
             with reporter.step("Check others have no access to objects with deny attribute"):
                 with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                    head_object(other_wallet, cid, deny_objects[0], self.shell, endpoint, xhdr=xhdr)
+                    head_object(other_wallet, container, deny_objects[0], self.shell, rpc_endpoint, xhdr=xhdr)
 
                 with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                    get_object_from_random_node(other_wallet, cid, deny_objects[0], self.shell, self.cluster, xhdr=xhdr)
+                    get_object_from_random_node(other_wallet, container, deny_objects[0], self.shell, self.cluster, xhdr=xhdr)
 
             with reporter.step("Check others have access to objects with deny attribute and using bearer token"):
-                assert_full_access_to_container(other_wallet, cid, deny_objects.pop(), file_path, self.shell, self.cluster, bearer, xhdr)
+                assert_full_access_to_container(
+                    other_wallet, container, deny_objects.pop(), file_path, self.shell, self.cluster, bearer, xhdr
+                )
 
         allow_attribute = self.OTHER_HEADER if match_type == ape.MatchType.EQUAL else self.HEADER
         with reporter.step("Check others can PUT objects without denied attribute"):
-            put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, attributes=allow_attribute)
+            put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, attributes=allow_attribute)
 
         deny_attribute = self.HEADER if match_type == ape.MatchType.EQUAL else self.OTHER_HEADER
         with reporter.step("Check others can not PUT objects with denied attribute"):
             with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
-                put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, attributes=deny_attribute)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, attributes=deny_attribute)
 
         with reporter.step("Check others can PUT objects with denied attribute and using bearer token"):
-            put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, bearer_put, attributes=deny_attribute)
+            put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, bearer_put, attributes=deny_attribute)
 
     @allure.title("Operations with allow APE rule with resource filters (match_type={match_type}, obj_size={object_size})")
     @pytest.mark.parametrize("match_type", [ape.MatchType.EQUAL, ape.MatchType.NOT_EQUAL])
     @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
+    @pytest.mark.container(ContainerSpec(basic_acl="0", allow_owner_via_ape=True))
     def test_ape_allow_filters_object(
         self,
         frostfs_cli: FrostfsCli,
         other_wallet: WalletInfo,
-        container_with_objects: tuple[str, list[str], str],
+        container: str,
-        temp_directory: str,
+        objects_with_attributes: list[str],
+        objects_with_other_attributes: list[str],
+        objects_without_attributes: list[str],
         match_type: ape.MatchType,
+        rpc_endpoint: str,
+        file_path: TestFile,
+        temp_directory: str,
     ):
-        # TODO: Refactor this to be fixtures, not test logic!
-        (
-            cid,
-            objects_with_attributes,
-            objects_with_other_attributes,
-            objects_without_attributes,
-            file_path,
-        ) = container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
         if match_type == ape.MatchType.EQUAL:
             allow_objects = objects_with_attributes
             deny_objects = objects_with_other_attributes
@@ -308,14 +268,15 @@ class TestApeFilters(ClusterTestBase):
             allow_attribute = self.OTHER_HEADER
             deny_attribute = self.HEADER
             no_attributes_match_context = expect_not_raises()
-        # End of refactor block
 
         with reporter.step("Allow operations for others except few operations by resource condition via APE"):
             resource_condition = ape.Condition('"check_key"', '"check_value"', ape.ConditionType.RESOURCE, match_type)
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             deny_rule = ape.Rule(ape.Verb.ALLOW, self.RESOURCE_OPERATIONS, [resource_condition, role_condition])
 
-            frostfs_cli.ape_manager.add(endpoint, deny_rule.chain_id, target_name=cid, target_type="container", rule=deny_rule.as_string())
+            frostfs_cli.ape_manager.add(
+                rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container", rule=deny_rule.as_string()
+            )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
@@ -323,113 +284,115 @@ class TestApeFilters(ClusterTestBase):
         with reporter.step("Check GET, PUT and HEAD operations with objects without attributes for OTHERS role"):
             oid = objects_without_attributes.pop()
             with no_attributes_match_context:
-                assert head_object(other_wallet, cid, oid, self.shell, endpoint)
+                assert head_object(other_wallet, container, oid, self.shell, rpc_endpoint)
 
             with no_attributes_match_context:
-                assert get_object_from_random_node(other_wallet, cid, oid, self.shell, self.cluster)
+                assert get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster)
 
             with no_attributes_match_context:
-                assert put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster)
+                assert put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster)
 
         with reporter.step("Create bearer token with everything allowed for others role"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, role_condition)
-            bearer = create_bearer_token(frostfs_cli, temp_directory, cid, rule, endpoint)
+            bearer = create_bearer_token(frostfs_cli, temp_directory, container, rule, rpc_endpoint)
 
         with reporter.step("Check others can get and put objects without attributes and using bearer token"):
             oid = objects_without_attributes[0]
             with expect_not_raises():
-                head_object(other_wallet, cid, oid, self.shell, endpoint, bearer)
+                head_object(other_wallet, container, oid, self.shell, rpc_endpoint, bearer)
 
             with expect_not_raises():
-                get_object_from_random_node(other_wallet, cid, oid, self.shell, self.cluster, bearer)
+                get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster, bearer)
 
             with expect_not_raises():
-                put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, bearer)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, bearer)
 
         with reporter.step("Check others can get and put objects with attributes matching the filter"):
             oid = allow_objects.pop()
             with expect_not_raises():
-                head_object(other_wallet, cid, oid, self.shell, endpoint)
+                head_object(other_wallet, container, oid, self.shell, rpc_endpoint)
 
             with expect_not_raises():
-                get_object_from_random_node(other_wallet, cid, oid, self.shell, self.cluster)
+                get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster)
 
             with expect_not_raises():
-                put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, attributes=allow_attribute)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, attributes=allow_attribute)
 
         with reporter.step("Check others cannot get and put objects without attributes matching the filter"):
             oid = deny_objects[0]
             with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                head_object(other_wallet, cid, oid, self.shell, endpoint)
+                head_object(other_wallet, container, oid, self.shell, rpc_endpoint)
 
             with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                assert get_object_from_random_node(other_wallet, cid, oid, self.shell, self.cluster)
+                assert get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster)
 
             with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                assert put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, attributes=deny_attribute)
+                assert put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, attributes=deny_attribute)
 
         with reporter.step("Check others can get and put objects without attributes matching the filter with bearer token"):
             oid = deny_objects.pop()
             with expect_not_raises():
-                head_object(other_wallet, cid, oid, self.shell, endpoint, bearer)
+                head_object(other_wallet, container, oid, self.shell, rpc_endpoint, bearer)
 
             with expect_not_raises():
-                get_object_from_random_node(other_wallet, cid, oid, self.shell, self.cluster, bearer)
+                get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster, bearer)
 
             with expect_not_raises():
-                put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, bearer, attributes=allow_attribute)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, bearer, attributes=allow_attribute)
 
     @allure.title("PUT and GET object using bearer with objectID in filter (obj_size={object_size}, match_type=NOT_EQUAL)")
+    @pytest.mark.container(ContainerSpec(basic_acl="0", allow_owner_via_ape=True))
     def test_ape_filter_object_id_not_equals(
         self,
         frostfs_cli: FrostfsCli,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
-        private_container: str,
+        container: str,
         temp_directory: str,
         file_path: TestFile,
     ):
         with reporter.step("Put object to container"):
-            oid = put_object_to_random_node(default_wallet, file_path, private_container, self.shell, self.cluster)
+            oid = put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster)
 
         with reporter.step("Create bearer token with objectID filter"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             object_condition = ape.Condition.by_object_id(oid, ape.ConditionType.RESOURCE, ape.MatchType.NOT_EQUAL)
             rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, [role_condition, object_condition])
-            bearer = create_bearer_token(frostfs_cli, temp_directory, private_container, rule, self.cluster.default_rpc_endpoint)
+            bearer = create_bearer_token(frostfs_cli, temp_directory, container, rule, self.cluster.default_rpc_endpoint)
 
         with reporter.step("Others should be able to put object using bearer token"):
             with expect_not_raises():
-                put_object_to_random_node(other_wallet, file_path, private_container, self.shell, self.cluster, bearer)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, bearer)
 
         with reporter.step("Others should not be able to get object matching the filter"):
             with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                get_object_from_random_node(other_wallet, private_container, oid, self.shell, self.cluster, bearer)
+                get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster, bearer)
 
     @allure.title("PUT and GET object using bearer with objectID in filter (obj_size={object_size}, match_type=EQUAL)")
+    @pytest.mark.container(ContainerSpec(basic_acl="0", allow_owner_via_ape=True))
     def test_ape_filter_object_id_equals(
         self,
         frostfs_cli: FrostfsCli,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
-        private_container: str,
+        container: str,
         temp_directory: str,
         file_path: TestFile,
     ):
         with reporter.step("Put object to container"):
-            oid = put_object_to_random_node(default_wallet, file_path, private_container, self.shell, self.cluster)
+            oid = put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster)
 
         with reporter.step("Create bearer token with objectID filter"):
             role_condition = ape.Condition.by_role(ape.Role.OTHERS)
             object_condition = ape.Condition.by_object_id(oid, ape.ConditionType.RESOURCE, ape.MatchType.EQUAL)
             rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, [role_condition, object_condition])
-            bearer = create_bearer_token(frostfs_cli, temp_directory, private_container, rule, self.cluster.default_rpc_endpoint)
+            bearer = create_bearer_token(frostfs_cli, temp_directory, container, rule, self.cluster.default_rpc_endpoint)
 
         with reporter.step("Others should not be able to put object using bearer token"):
             with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
-                put_object_to_random_node(other_wallet, file_path, private_container, self.shell, self.cluster, bearer)
+                put_object_to_random_node(other_wallet, file_path, container, self.shell, self.cluster, bearer)
 
         with reporter.step("Others should be able to get object matching the filter"):
             with expect_not_raises():
-                get_object_from_random_node(other_wallet, private_container, oid, self.shell, self.cluster, bearer)
+                get_object_from_random_node(other_wallet, container, oid, self.shell, self.cluster, bearer)

pytest_tests/testsuites/access/ape/test_bearer.py

@@ -5,6 +5,7 @@ from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
+from frostfs_testlib.utils.file_utils import TestFile
 
 from pytest_tests.helpers.bearer_token import create_bearer_token
 from pytest_tests.helpers.container_access import (
@@ -23,21 +24,21 @@ class TestApeBearer(ClusterTestBase):
     @pytest.mark.parametrize("role", [ape.Role.OWNER, ape.Role.OTHERS], indirect=True)
     def test_bearer_token_operations(
         self,
-        container_with_objects: tuple[str, list[str], str],
+        container: str,
+        objects: list[str],
         frostfs_cli: FrostfsCli,
         temp_directory: str,
         test_wallet: WalletInfo,
         role: ape.Role,
+        file_path: TestFile,
+        rpc_endpoint: str,
     ):
-        cid, objects_oids, file_path = container_with_objects
-        endpoint = self.cluster.default_rpc_endpoint
-
         with reporter.step(f"Check {role} has full access to container without bearer token"):
-            assert_full_access_to_container(test_wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
         with reporter.step(f"Deny all operations for everyone via APE"):
             rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS)
-            frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
@@ -46,25 +47,25 @@ class TestApeBearer(ClusterTestBase):
             bearer = create_bearer_token(
                 frostfs_cli,
                 temp_directory,
-                cid,
+                container,
                 rule=ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS),
-                endpoint=self.cluster.default_rpc_endpoint,
+                endpoint=rpc_endpoint,
             )
 
         with reporter.step(f"Check {role} without token has no access to all operations with container"):
-            assert_no_access_to_container(test_wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster)
+            assert_no_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
         with reporter.step(f"Check {role} with token has access to all operations with container"):
-            assert_full_access_to_container(test_wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster, bearer)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster, bearer)
 
         with reporter.step(f"Remove deny rule from APE"):
-            frostfs_cli.ape_manager.remove(endpoint, rule.chain_id, target_name=cid, target_type="container")
+            frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         with reporter.step(f"Check {role} without token has access to all operations with container"):
-            assert_full_access_to_container(test_wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
     @allure.title("BearerToken for compound operations (obj_size={object_size})")
     def test_bearer_token_compound_operations(
@@ -73,16 +74,16 @@ class TestApeBearer(ClusterTestBase):
         temp_directory: str,
         default_wallet: WalletInfo,
         other_wallet: WalletInfo,
-        container_with_objects: tuple[str, list[str], str],
+        container: tuple[str, list[str], str],
+        objects: list[str],
+        rpc_endpoint: str,
+        file_path: TestFile,
     ):
         """
         Bearer Token COMPLETLY overrides chains set for the specific target.
         Thus, any restictions or permissions should be explicitly defined in BT.
         """
 
-        endpoint = self.cluster.default_rpc_endpoint
-        cid, objects_oids, file_path = container_with_objects
-
         wallets_map = {
             ape.Role.OWNER: default_wallet,
             ape.Role.OTHERS: other_wallet,
@@ -167,24 +168,26 @@ class TestApeBearer(ClusterTestBase):
         for role, operations in deny_map.items():
             with reporter.step(f"Add APE deny rule for {role}"):
                 rule = ape.Rule(ape.Verb.DENY, operations, conditions_map[role])
-                frostfs_cli.ape_manager.add(endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string())
+                frostfs_cli.ape_manager.add(
+                    rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string()
+                )
 
         with reporter.step("Wait for one block"):
             self.wait_for_blocks()
 
         for role, wallet in wallets_map.items():
             with reporter.step(f"Assert access to container without bearer token for {role}"):
-                assert_access_to_container(access_map[role], wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster)
+                assert_access_to_container(access_map[role], wallet, container, objects.pop(), file_path, self.shell, self.cluster)
 
         bearer_tokens = {}
         for role in wallets_map.keys():
             with reporter.step(f"Create bearer token for {role}"):
                 rule = ape.Rule(verb_map[role], bearer_map[role], conditions_map[role])
-                bt = create_bearer_token(frostfs_cli, temp_directory, cid, rule, endpoint)
+                bt = create_bearer_token(frostfs_cli, temp_directory, container, rule, rpc_endpoint)
                 bearer_tokens[role] = bt
 
         for role, wallet in wallets_map.items():
             with reporter.step(f"Assert access to container with bearer token for {role}"):
                 assert_access_to_container(
-                    bt_access_map[role], wallet, cid, objects_oids.pop(), file_path, self.shell, self.cluster, bearer_tokens[role]
+                    bt_access_map[role], wallet, container, objects.pop(), file_path, self.shell, self.cluster, bearer_tokens[role]
                 )

pytest_tests/testsuites/access/conftest.py

@@ -1,13 +1,22 @@
+import json
+import time
+
+import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
+from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
 from frostfs_testlib.resources.wellknown_acl import PUBLIC_ACL
 from frostfs_testlib.shell import Shell
-from frostfs_testlib.steps.cli.container import create_container
+from frostfs_testlib.steps.cli.container import create_container, search_nodes_with_container
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
-from frostfs_testlib.storage.cluster import Cluster
+from frostfs_testlib.storage.cluster import Cluster, ClusterNode
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.parallel import parallel
+from frostfs_testlib.utils import datetime_utils
+
+from ...helpers.container_spec import ContainerSpec
 
 OBJECT_COUNT = 5
 
@@ -39,26 +48,108 @@ def test_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.
     return role_to_wallet_map[role]
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture
-def container_with_objects(default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, file_path: str) -> tuple[str, list[str], str]:
+def container(
+    default_wallet: WalletInfo,
+    frostfs_cli: FrostfsCli,
+    client_shell: Shell,
+    cluster: Cluster,
+    request: pytest.FixtureRequest,
+    rpc_endpoint: str,
+) -> str:
+    container_spec = _get_container_spec(request)
+    cid = _create_container_by_spec(default_wallet, client_shell, cluster, rpc_endpoint, container_spec)
+    if container_spec.allow_owner_via_ape:
+        _allow_owner_via_ape(frostfs_cli, cluster, cid)
 
-    with reporter.step("Create public container"):
+    return cid
+
+
+def _create_container_by_spec(
+    default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, rpc_endpoint: str, container_spec: ContainerSpec
+) -> str:
+    # TODO: add container spec to step message
+    with reporter.step("Create container"):
         cid = create_container(
-            default_wallet,
+            default_wallet, client_shell, rpc_endpoint, basic_acl=container_spec.basic_acl, rule=container_spec.parsed_rule(cluster)
-            shell=client_shell,
-            endpoint=cluster.default_rpc_endpoint,
-            basic_acl=PUBLIC_ACL,
         )
 
+    with reporter.step("Search nodes holding the container"):
+        container_holder_nodes = search_nodes_with_container(default_wallet, cid, client_shell, cluster.default_rpc_endpoint, cluster)
+        report_data = {node.id: node.host_ip for node in container_holder_nodes}
+
+    reporter.attach(json.dumps(report_data, indent=2), "container_nodes.json")
+
+    return cid
+
+
+def _get_container_spec(request: pytest.FixtureRequest) -> ContainerSpec:
+    container_marker = request.node.get_closest_marker("container")
+    # let default container to be public at the moment
+    container_spec = ContainerSpec(basic_acl=PUBLIC_ACL)
+
+    if container_marker:
+        if len(container_marker.args) != 1:
+            raise RuntimeError(f"Something wrong with container marker: {container_marker}")
+        container_spec = container_marker.args[0]
+
+    if "param" in request.__dict__:
+        container_spec = request.param
+
+    if not container_spec:
+        raise RuntimeError(
+            f"""Container specification is empty. 
+                            Either add @pytest.mark.container(ContainerSpec(...)) or 
+                            @pytest.mark.parametrize(\"container\", [ContainerSpec(...)], indirect=True) decorator"""
+        )
+
+    return container_spec
+
+
+def _allow_owner_via_ape(frostfs_cli: FrostfsCli, cluster: Cluster, container: str):
+    with reporter.step("Create allow APE rule for container owner"):
+        role_condition = ape.Condition.by_role(ape.Role.OWNER)
+        deny_rule = ape.Rule(ape.Verb.ALLOW, ape.ObjectOperations.WILDCARD_ALL, role_condition)
+
+        frostfs_cli.ape_manager.add(
+            cluster.default_rpc_endpoint,
+            deny_rule.chain_id,
+            target_name=container,
+            target_type="container",
+            rule=deny_rule.as_string(),
+        )
+
+    with reporter.step("Wait for one block"):
+        time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME))
+
+
+@pytest.fixture
+def objects(container: str, default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, file_path: str):
     with reporter.step("Add test objects to container"):
         put_results = parallel(
             [put_object_to_random_node] * OBJECT_COUNT,
             wallet=default_wallet,
             path=file_path,
-            cid=cid,
+            cid=container,
             shell=client_shell,
             cluster=cluster,
         )
         objects_oids = [put_result.result() for put_result in put_results]
 
-    return cid, objects_oids, file_path
+    return objects_oids
+
+
+@pytest.fixture
+def container_nodes(default_wallet: WalletInfo, container: str, client_shell: Shell, cluster: Cluster) -> list[ClusterNode]:
+    cid = container
+    container_holder_nodes = search_nodes_with_container(default_wallet, cid, client_shell, cluster.default_rpc_endpoint, cluster)
+
+    report_data = {node.id: node.host_ip for node in container_holder_nodes}
+    reporter.attach(json.dumps(report_data, indent=2), "container_nodes.json")
+
+    return container_holder_nodes
+
+
+@pytest.fixture
+def container_node_wallet(container_nodes: list[ClusterNode]) -> WalletInfo:
+    return WalletInfo.from_node(container_nodes[0].storage_node)

pytest_tests/testsuites/conftest.py

@@ -401,6 +401,11 @@ def after_deploy_healthcheck(cluster: Cluster):
         parallel(readiness_on_node, cluster.cluster_nodes)
 
 
+@pytest.fixture(scope="session")
+def rpc_endpoint(cluster: Cluster):
+    return cluster.default_rpc_endpoint
+
+
 @wait_for_success(60 * SERVICE_ACTIVE_TIME * 3, 60, title="Wait for {cluster_node} readiness")
 def readiness_on_node(cluster_node: ClusterNode):
     if "skip_readiness_check" in cluster_node.host.config.attributes and cluster_node.host.config.attributes["skip_readiness_check"]: