import allure 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 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 @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})") def test_basic_acl_public( self, default_wallet: WalletInfo, other_wallet: WalletInfo, client_shell: Shell, public_container: str, file_path: str, cluster: Cluster, ): """ Test access to object operations in public container. """ for wallet, role in ((default_wallet, "owner"), (other_wallet, "others")): with reporter.step("Put objects to container"): # We create new objects for each wallet because assert_full_access_to_container # deletes the object owner_object_oid = put_object_to_random_node( default_wallet, file_path, public_container, shell=self.shell, cluster=self.cluster, attributes={"created": "owner"}, ) other_object_oid = put_object_to_random_node( other_wallet, file_path, public_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, public_container, other_object_oid, file_path, client_shell, cluster) @allure.title("Operations in private container only available to owner (obj_size={object_size})") def test_basic_acl_private( self, default_wallet: WalletInfo, other_wallet: WalletInfo, client_shell: Shell, private_container: str, file_path: str, cluster: Cluster, ): """ Test access to object operations in private container. """ with reporter.step("Put object to container"): owner_object_oid = put_object_to_random_node(default_wallet, file_path, private_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) 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) @allure.title("Read operations in readonly container available to others (obj_size={object_size})") def test_basic_acl_readonly( self, default_wallet: WalletInfo, other_wallet: WalletInfo, client_shell: Shell, readonly_container: str, file_path: str, cluster: Cluster, ): """ Test access to object operations in readonly container. """ with reporter.step("Put object to container"): object_oid = put_object_to_random_node(default_wallet, file_path, readonly_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) 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)