From 2ce8f43e9c37a55896dfb2f95e0f0c5d34b73858 Mon Sep 17 00:00:00 2001 From: Ekaterina Chernitsyna Date: Thu, 31 Oct 2024 13:09:42 +0300 Subject: [PATCH] [#303] add local deny ape tests --- pytest_tests/testsuites/ape/test_ape.py | 905 ------------------------ 1 file changed, 905 deletions(-) diff --git a/pytest_tests/testsuites/ape/test_ape.py b/pytest_tests/testsuites/ape/test_ape.py index c28cd6a..467f05f 100644 --- a/pytest_tests/testsuites/ape/test_ape.py +++ b/pytest_tests/testsuites/ape/test_ape.py @@ -35,18 +35,6 @@ def remote_frostfs_cli_first_node(cluster: Cluster): return cli -@pytest.fixture(scope="session") -def remote_frostfs_adm_first_node(cluster: Cluster): - node = cluster.cluster_nodes[0] - shell = node.host.get_shell() - adm = FrostfsAdm( - shell=shell, - frostfs_adm_exec_path=FROSTFS_ADM_EXEC, - config_file=FROSTFS_ADM_CONFIG_PATH, - ) - return adm - - def local_overrides_on_node(node: ClusterNode): target = "Chain ID" shell: Shell = node.host.get_shell() @@ -123,899 +111,6 @@ def pre_create_container_object_cli( return cid, oid -def pre_create_container_object_adm( - default_user: User, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - simple_object_size: ObjectSize, - shell: Shell, - cluster: Cluster, -): - - 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=cluster.storage_nodes[0].get_rpc_endpoint(), - policy="REP 4", - name="dcl1", - await_mode=True, - basic_acl="0", - ) - .stdout.split(" ")[1] - .strip() - .split("\n")[0] - ) - - with reporter.step("Create a namespace 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("Put objects in container on the first node"): - oid_1 = put_object(default_user.wallet, test_file, cid, shell, cluster.storage_nodes[0].get_rpc_endpoint()) - oid_2 = put_object(default_user.wallet, test_file, cid, shell, cluster.storage_nodes[0].get_rpc_endpoint(), copies_number=3) - - with reporter.step("Create a namespace rule for the first node"): - remote_frostfs_adm_first_node.morph.remove_rule( - target_type="container", - target_name=f"{cid}", - chain_id="allowPutObject", - ) - - return cid, (oid_1, oid_2) - - -@pytest.mark.ape -@pytest.mark.ape_morph_rule -class TestApeMorphRuleChain(ClusterTestBase): - @allure.title("MorphRuleChain: Deny to GetContainer in root tenant") - def test_morph_rule_chain_deny_to_get_container_root( - self, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - ): - - 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 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .stdout.split(" ")[1] - .strip() - .split("\n")[0] - ) - - with reporter.step("Create a namespace rule for the first node"): - remote_frostfs_adm_first_node.morph.add_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerGet", - rule="deny Container.Get *", - ) - - with reporter.step("Check get the container property on the first node, expected access drnied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.GET_CONTAINER)): - frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid) - - with reporter.step("Check et the container property on the second node, expected access drnied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.GET_CONTAINER)): - frostfs_cli.container.get(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid) - - with reporter.step("Delete a rule"): - remote_frostfs_adm_first_node.morph.remove_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerGet", - ) - - with reporter.step("Check get the container property on the first node, expected allow"): - with expect_not_raises(): - frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid) - - @allure.title("MorphRuleChain: Deny to PutContainer in root tenant") - def test_morph_rule_chain_deny_to_put_container_root( - self, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - ): - - with reporter.step("Create a namespace rule for the first node"): - remote_frostfs_adm_first_node.morph.add_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerPut", - rule="deny Container.Put *", - ) - - with reporter.step("Check create a container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.PUT_CONTAINER)): - frostfs_cli.container.create( - rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), - policy="REP 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - - with reporter.step("Check create a container on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.PUT_CONTAINER)): - frostfs_cli.container.create( - rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(), - policy="REP 4", - name="dcl2", - await_mode=True, - basic_acl="public-read-write", - ) - - with reporter.step("Delete a rule"): - remote_frostfs_adm_first_node.morph.remove_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerPut", - ) - - with reporter.step("Check create a container on the first node, expected allow"): - with expect_not_raises(): - frostfs_cli.container.create( - rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), - policy="REP 4", - name="dcl3", - await_mode=True, - basic_acl="public-read-write", - ) - - @allure.title("Deny to ListContainer in root tenant") - def test_morph_rule_chain_deny_to_list_container_root( - self, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - ): - - 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 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .stdout.split(" ")[1] - .strip() - .split("\n")[0] - ) - - with reporter.step("Create a namespace rule for the first node"): - remote_frostfs_adm_first_node.morph.add_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerList", - rule="deny Container.List *", - ) - - with reporter.step("Check list the container properties on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.LIST_CONTAINER)): - frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1) - - with reporter.step("Check list the container properties on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.LIST_CONTAINER)): - frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(), ttl=1) - - with reporter.step("Delete a rule"): - remote_frostfs_adm_first_node.morph.remove_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerList", - ) - - with reporter.step("Check display a list of containers on the first node, expected allow"): - with expect_not_raises(): - frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1) - - @allure.title("MorphRuleChain: Deny to DeleteContainer in root tenant") - def test_morph_rule_chain_deny_to_delete_container_root( - self, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - ): - with reporter.step("Create a namespace rule for the first node"): - remote_frostfs_adm_first_node.morph.add_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerDelete", - rule="deny Container.Delete *", - ) - - with reporter.step("Create container on the first node"): - cid = ( - frostfs_cli.container.create( - rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), - policy="REP 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .stdout.split(" ")[1] - .strip() - .split("\n")[0] - ) - - with reporter.step("Check delete first container from the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.DELETE_CONTAINER)): - frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid, ttl=1) - - with reporter.step("Check delete a second container from the second node, expected access error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.DELETE_CONTAINER)): - frostfs_cli.container.delete(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid, ttl=1) - - with reporter.step("Delete a rule"): - remote_frostfs_adm_first_node.morph.remove_rule( - target_type="namespace", - target_name="root", - chain_id="denyContainerDelete", - ) - - with reporter.step("Check delete first container from the first node,expected allow"): - with expect_not_raises(): - frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid, ttl=1) - - @allure.title("MorphRuleChain: Deny to GetObject in root tenant") - def test_morph_rule_chain_deny_to_get_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 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyGetObject", - rule=f"deny Object.Get /{cid}/*", - ) - - with reporter.step("Put object in container on the first node"): - oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check get object from container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.GET_OBJECT)): - get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check get object from container on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.GET_OBJECT)): - get_object(default_user.wallet, cid, oid, 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="denyGetObject", - ) - - with reporter.step("Check get object in container on the first node, expected allow"): - with expect_not_raises(): - get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to PutObject in root tenant") - def test_morph_rule_chain_deny_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 2 CBF 1", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyPutObject", - rule=f"deny Object.Put /{cid}/*", - ) - - with reporter.step("Check put object from container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED): - put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check put object from container on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED): - 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="denyPutObject", - ) - - with reporter.step("Check put object in container on the first node, expected allow"): - with expect_not_raises(): - put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to HeadObject in root tenant") - def test_morph_rule_chain_deny_to_head_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 4", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyHeadObject", - rule=f"deny Object.Head /{cid}/*", - ) - - with reporter.step("Put object in container on the first node"): - oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check head object from container on the first nodel expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HEAD_OBJECT)): - head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check head object from container on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HEAD_OBJECT)): - head_object(default_user.wallet, cid, oid, 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="denyHeadObject", - ) - - with reporter.step("Check head object in container on the first node, expected allow"): - with expect_not_raises(): - head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to SearchObject in root tenant") - def test_morph_rule_chain_deny_to_search_object_root( - self, - default_user: User, - remote_frostfs_adm_first_node: FrostfsAdm, - frostfs_cli: FrostfsCli, - simple_object_size: ObjectSize, - ): - - 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 2", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denySearchObject", - rule=f"deny Object.Search /{cid}/*", - ) - - with reporter.step("Check search object from container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)): - 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, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)): - 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="container", - target_name=f"{cid}", - chain_id="denySearchObject", - ) - - with reporter.step("Check search object in container on the first node, expected allow"): - with expect_not_raises(): - search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to RangeObject in root tenant") - def test_morph_rule_chain_deny_to_range_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 2", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyRangeObject", - rule=f"deny Object.Range /{cid}/*", - ) - - with reporter.step("Put object in container on the first node"): - oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check range object from container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)): - get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check range object from container on the second node, expected denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)): - get_range(default_user.wallet, cid, oid, "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="container", - target_name=f"{cid}", - chain_id="denyRangeObject", - ) - - with reporter.step("Check range object in container on the first node,expected allow"): - with expect_not_raises(): - get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to HashObject in root tenant") - def test_morph_rule_chain_deny_to_hash_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 2", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyHashObject", - rule=f"deny Object.Hash /{cid}/*", - ) - - with reporter.step("Put object in container on the first node"): - oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Check range hash object from container on the first node, expected denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)): - get_range_hash(default_user.wallet, cid, oid, "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, expected denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)): - get_range_hash(default_user.wallet, cid, oid, "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="container", - target_name=f"{cid}", - chain_id="denyHashObject", - ) - - with reporter.step("Check range hash object in container on the first node, expected denied error"): - with expect_not_raises(): - get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - @allure.title("MorphRuleChain: Deny to DeleteObject in root tenant") - def test_morph_rule_chain_deny_to_delete_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 2", - name="dcl1", - await_mode=True, - basic_acl="public-read-write", - ) - .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="denyDeleteObject", - rule=f"deny Object.Delete /{cid}/*", - ) - - with reporter.step("Put two objects in container on the first node"): - oid_1 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - oid_2 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Search object in container on the first node"): - search_object_in_container_1 = search_object( - default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint() - ) - assert oid_1 in search_object_in_container_1, f"Object {oid_1} was not found" - assert oid_2 in search_object_in_container_1, f"Object {oid_2} was not found" - - with reporter.step("Search object from container on the second node"): - search_object_in_container_2 = search_object( - default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint() - ) - assert oid_1 in search_object_in_container_2, f"Object {oid_1} was not found" - assert oid_2 in search_object_in_container_2, f"Object {oid_2} was not found" - - with reporter.step("Check delete object from container on the first node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.DELETE_OBJECT)): - delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) - - with reporter.step("Delete object from container on the second node, expected access denied error"): - with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.DELETE_OBJECT)): - delete_object(default_user.wallet, cid, oid_2, self.shell, self.cluster.storage_nodes[0].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="denyDeleteObject", - ) - - with reporter.step("Check delete object in container on the first node, expected allow"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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 namespace 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, allow expected"): - 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, allow expected"): - 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", - ) - - @pytest.mark.ape @pytest.mark.ape_local class TestApeLocalOverride(ClusterTestBase):