255 lines
12 KiB
Python
255 lines
12 KiB
Python
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",
|
|
)
|