[#327] add allow morph tests ape

pytest_tests/testsuites/ape/test_ape.py

@@ -27,18 +27,6 @@ from frostfs_testlib.utils.file_utils import generate_file
 reporter = get_reporter()
 
 
-@pytest.fixture(scope="session")
-def remote_frostfs_cli_first_node(cluster: Cluster):
-    node = cluster.cluster_nodes[0]
-    shell = node.host.get_shell()
-    cli = FrostfsCli(
-        shell=shell,
-        frostfs_cli_exec_path=FROSTFS_CLI_EXEC,
-        config_file=node.storage_node.get_remote_wallet_config_path(),
-    )
-    return cli
-
-
 @pytest.fixture(scope="session")
 def remote_frostfs_adm_first_node(cluster: Cluster):
     node = cluster.cluster_nodes[0]
@@ -51,22 +39,6 @@ def remote_frostfs_adm_first_node(cluster: Cluster):
     return adm
 
 
-def local_overrides_on_node(node: ClusterNode, id_chains_for_remove: list[str]):
-    shell: Shell = node.host.get_shell()
-    remote_config: str = node.storage_node.get_remote_wallet_config_path()
-    cli = FrostfsCli(shell=shell, frostfs_cli_exec_path=FROSTFS_CLI_EXEC, config_file=remote_config)
-    with reporter.step(f"Delete rules on {node.storage_node.id} node"):
-        for chain_id in id_chains_for_remove:
-            cli.control.remove_rule(
-                endpoint=node.storage_node.get_control_endpoint(),
-                target_type="namespace",
-                target_name="root",
-                chain_id=chain_id,
-            )
-        with reporter.step("Wait for one block"):
-            time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME))
-
-
 def morph_on_node(node: ClusterNode, id_chains_for_remove: list[str]):
     shell: Shell = node.host.get_shell()
     adm = FrostfsAdm(
@@ -74,7 +46,7 @@ def morph_on_node(node: ClusterNode, id_chains_for_remove: list[str]):
         frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
         config_file=FROSTFS_ADM_CONFIG_PATH,
     )
-    rules = adm.morph.list_rules(target_name="kapusta", target_type="namespace").stdout
+    rules = adm.morph.list_rules(target_name="root", target_type="namespace").stdout
     chain_ids = [i.split(" ")[2].strip() for i in rules.split("\n") if "Chain ID" in i]
     need_wait = False
     with reporter.step(f"Delete rules on {node.storage_node.id} node"):
@@ -92,14 +64,6 @@ def morph_on_node(node: ClusterNode, id_chains_for_remove: list[str]):
                 time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME))
 
 
-@pytest.fixture(scope="session")
-def remove_rule_ape_in_local_override(cluster: Cluster) -> None:
-    id_chains_for_remove = []
-    yield id_chains_for_remove
-    with reporter.step("Check local overrides on nodes."):
-        parallel(local_overrides_on_node, cluster.cluster_nodes, id_chains_for_remove)
-
-
 @pytest.fixture(scope="session")
 def remove_rule_ape_in_morph(cluster: Cluster) -> None:
     id_chains_for_remove = []
@@ -766,3 +730,261 @@ class TestApeMorphRuleChain(ClusterTestBase):
         with reporter.step("Check delete object in container on the first node"):
             with expect_not_raises():
                 delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+    @allure.title("MorphRuleChain: Allow to GetObject in root tenant")
+    def test_morph_rule_chain_allow_to_get_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowGetObject",
+                rule=f"allow Object.Get *",
+            )
+
+        with reporter.step("Check get object from container on the first node"):
+            with expect_not_raises():
+                get_object(default_user.wallet, cid, oids[0], self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check get object from container on the second node"):
+            with expect_not_raises():
+                get_object(default_user.wallet, cid, oids[0], self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowGetObject",
+            )
+
+    @allure.title("MorphRuleChain: allow to PutObject in root tenant")
+    def test_morph_rule_chain_allow_to_put_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+
+        test_file = generate_file(simple_object_size.value)
+
+        with reporter.step("Create a container on the first node"):
+            cid = (
+                frostfs_cli.container.create(
+                    rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+                    policy="REP 1 IN MOW CBF 1 SELECT 1 FROM MSK AS MOW FILTER SubDivCode EQ MOW AS MSK",
+                    name="dcl1",
+                    await_mode=True,
+                    basic_acl="0",
+                )
+                .stdout.split(" ")[1]
+                .strip()
+                .split("\n")[0]
+            )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowPutObject",
+                rule=f"allow Object.Put *",
+            )
+
+        with reporter.step("Check put object to container on the first node"):
+            with expect_not_raises():
+                put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check put object to container on the second node"):
+            with expect_not_raises():
+                put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowPutObject",
+            )
+
+    @allure.title("MorphRuleChain: Allow to HeadObject in root tenant")
+    def test_morph_rule_chain_allow_to_head_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowHeadObject",
+                rule=f"allow Object.Head *",
+            )
+
+        with reporter.step("Check head object from container on the first node"):
+            with expect_not_raises():
+                head_object(default_user.wallet, cid, oids[0], self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check head object from container on the second node"):
+            with expect_not_raises():
+                head_object(default_user.wallet, cid, oids[0], self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowHeadObject",
+            )
+
+    @allure.title("MorphRuleChain: Allow to SearchObject in root tenant")
+    def test_morph_rule_chain_allow_to_search_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowSearchObject",
+                rule=f"allow Object.Search *",
+            )
+
+        with reporter.step("Check search object from container on the first node"):
+            with expect_not_raises():
+                search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check search object from container on the second node"):
+            with expect_not_raises():
+                search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowSearchObject",
+            )
+
+    @allure.title("MorphRuleChain: Allow to RangehObject in root tenant")
+    def test_morph_rule_chain_allow_to_range_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowRangeObject",
+                rule=f"allow Object.Range *",
+            )
+
+        with reporter.step("Check range object from container on the first node"):
+            with expect_not_raises():
+                get_range(default_user.wallet, cid, oids[0], "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check range object from container on the second node"):
+            with expect_not_raises():
+                get_range(default_user.wallet, cid, oids[0], "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowRangeObject",
+            )
+
+    @allure.title("MorphRuleChain: Allow to Hash Object in root tenant")
+    def test_morph_rule_chain_allow_to_hash_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowHashObject",
+                rule=f"allow Object.Hash *",
+            )
+
+        with reporter.step("Check range hash object from container on the first node"):
+            with expect_not_raises():
+                get_range_hash(default_user.wallet, cid, oids[0], "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check range hash object from container on the second node"):
+            with expect_not_raises():
+                get_range_hash(default_user.wallet, cid, oids[0], "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowHashObject",
+            )
+
+    @allure.title("MorphRuleChain: Allow to Delete Object in root tenant")
+    def test_morph_rule_chain_allow_to_delete_object_root(
+        self,
+        default_user: User,
+        remote_frostfs_adm_first_node: FrostfsAdm,
+        frostfs_cli: FrostfsCli,
+        simple_object_size: ObjectSize,
+    ):
+        cid, oids = pre_create_container_object_adm(
+            default_user, remote_frostfs_adm_first_node, frostfs_cli, simple_object_size, self.shell, self.cluster
+        )
+
+        with reporter.step("Create a container rule for the first node"):
+            remote_frostfs_adm_first_node.morph.add_rule(
+                target_type="container",
+                target_name=f"{cid}",
+                chain_id="allowDeleteObject",
+                rule=f"allow Object.Head Object.Delete *",
+            )
+
+        with reporter.step("Check delete object from container on the first node"):
+            with expect_not_raises():
+                delete_object(default_user.wallet, cid, oids[0], self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+        with reporter.step("Check delete object from container on the second node"):
+            with expect_not_raises():
+                delete_object(default_user.wallet, cid, oids[1], self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+        with reporter.step("Delete a rule"):
+            remote_frostfs_adm_first_node.morph.remove_rule(
+                target_type="namespace",
+                target_name=f"kapusta",
+                chain_id="allowDeleteObject",
+            )