frostfs-testcases/pytest_tests/testsuites/ape/test_ape_local_object_allow.py

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.cli import FrostfsCli
from frostfs_testlib.resources.error_patterns import NO_RULE_FOUND_OBJECT
from frostfs_testlib.steps.cli.object import delete_object, get_object, get_range, get_range_hash, head_object, put_object, search_object
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.testing.test_control import expect_not_raises
from frostfs_testlib.utils.file_utils import generate_file

from ...helpers.container_request import ContainerRequest

REP1_MSK = ContainerRequest("REP 1 IN MOW CBF 1 SELECT 1 FROM MSK AS MOW FILTER SubDivCode EQ MOW AS MSK", short_name="REP1_MSK_no_ape")


@pytest.mark.ape
@pytest.mark.ape_local
@pytest.mark.ape_object
@pytest.mark.ape_allow
@pytest.mark.parametrize("container_request", [REP1_MSK], indirect=True)
@pytest.mark.parametrize("user_tag", ["ApeLocalOverrideAllow"], indirect=True)  # provide dedicated user with no APE side-policies
class TestApeLocalOverrideAllow(ClusterTestBase):
    @allure.title("LocalOverride: Allow to GetObject in root tenant")
    def test_local_override_allow_to_get_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowGetObject",
                rule=f"allow Object.Get *",
            )

        with reporter.step("Check get object in container on the first node, expected allow"):
            with expect_not_raises():
                get_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check get object in container on the second node, epxected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                get_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowGetObject",
            )

    @allure.title("LocalOverride: Allow to PutObject in root tenant")
    def test_local_override_allow_to_put_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        simple_object_size: ObjectSize,
        container: str,
    ):
        test_file = generate_file(simple_object_size.value)

        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPutObject",
                rule=f"allow Object.Put *",
            )

        with reporter.step("Check put object in container on the first node, expected allow"):
            with expect_not_raises():
                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check get object in container on the second node, epxected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPutObject",
            )

    @allure.title("LocalOverride: Allow to HeadObject in root tenant")
    def test_local_override_allow_to_head_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHeadObject",
                rule=f"allow Object.Head *",
            )

        with reporter.step("Check head object in container on the first node, expected allow"):
            with expect_not_raises():
                head_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check head object in container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                head_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHeadObject",
            )

    @allure.title("LocalOverride: Allow to SearchObject in root tenant")
    def test_local_override_allow_to_search_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowSearchObject",
                rule=f"allow Object.Search *",
            )

        with reporter.step("Check search object in container on the first node, expected allow"):
            with expect_not_raises():
                search_object(default_wallet, container, 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=NO_RULE_FOUND_OBJECT):
                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowSearchObject",
            )

    @allure.title("LocalOverride: Allow to RangeObject in root tenant")
    def test_local_override_allow_to_range_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowRangeObject",
                rule=f"allow Object.Range *",
            )

        with reporter.step("Check get range object in container on the first node, expected allow"):
            with expect_not_raises():
                get_range(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check range object in container on the second node. expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                get_range(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowRangeObject",
            )

    @allure.title("LocalOverride: Allow to HashObject in root tenant")
    def test_local_override_allow_to_hash_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHashObject",
                rule=f"allow Object.Hash *",
            )

        with reporter.step("Check get range hash object in container on the first node, expected allow"):
            with expect_not_raises():
                get_range_hash(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check get range hash object in container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                get_range_hash(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHashObject",
            )

    @allure.title("LocalOverride: Allow to DeleteObject in root tenant")
    def test_local_override_allow_to_delete_object_root(
        self,
        default_wallet: WalletInfo,
        frostfs_cli_on_first_node: FrostfsCli,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowDeleteObject",
                rule=f"allow Object.Head Object.Delete *",
            )

        with reporter.step("Check delete object from container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                delete_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Check delete object in container on the first node, expected allow"):
            with expect_not_raises():
                delete_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowDeleteObject",
            )