frostfs-testcases/pytest_tests/testsuites/access/acl/test_acl.py

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.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

from ....helpers.container_spec import ContainerSpec


@pytest.mark.sanity
@pytest.mark.smoke
@pytest.mark.acl
class TestACLBasic(ClusterTestBase):
    @allure.title("Operations in public container available to everyone (obj_size={object_size})")
    @pytest.mark.container(ContainerSpec(basic_acl=PUBLIC_ACL_F))
    def test_basic_acl_public(
        self,
        default_wallet: WalletInfo,
        other_wallet: WalletInfo,
        client_shell: Shell,
        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,
                    container,
                    shell=self.shell,
                    cluster=self.cluster,
                    attributes={"created": "owner"},
                )
                other_object_oid = put_object_to_random_node(
                    other_wallet,
                    file_path,
                    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, container, owner_object_oid, file_path, client_shell, cluster)
                assert_full_access_to_container(wallet, container, other_object_oid, file_path, client_shell, cluster)

    @allure.title("Operations in private container only available to owner (obj_size={object_size})")
    @pytest.mark.container(ContainerSpec(basic_acl=PRIVATE_ACL_F))
    def test_basic_acl_private(
        self,
        default_wallet: WalletInfo,
        other_wallet: WalletInfo,
        client_shell: Shell,
        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, 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, 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, container, owner_object_oid, file_path, self.shell, cluster)

    @allure.title("Read operations in readonly container available to others (obj_size={object_size})")
    @pytest.mark.container(ContainerSpec(basic_acl=READONLY_ACL_F))
    def test_basic_acl_readonly(
        self,
        default_wallet: WalletInfo,
        other_wallet: WalletInfo,
        client_shell: Shell,
        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, container, client_shell, cluster)

        with reporter.step("Check others has read-only access to operations with container"):
            assert_read_only_container(other_wallet, 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, container, object_oid, file_path, client_shell, cluster)