frostfs-testcases/pytest_tests/testsuites/session_token/test_static_session_token_container.py

import allure
import pytest
from frostfs_testlib.resources.common import PUBLIC_ACL
from frostfs_testlib.shell import Shell

from pytest_tests.helpers.acl import (
    EACLAccess,
    EACLOperation,
    EACLRole,
    EACLRule,
    create_eacl,
    set_eacl,
    wait_for_cache_expired,
)
from pytest_tests.helpers.container import (
    create_container,
    delete_container,
    get_container,
    list_containers,
)
from pytest_tests.helpers.file_helper import generate_file
from pytest_tests.helpers.object_access import can_put_object
from pytest_tests.helpers.wallet import WalletFile
from pytest_tests.steps.cluster_test_base import ClusterTestBase
from pytest_tests.steps.session_token import ContainerVerb, get_container_signed_token


@pytest.mark.static_session_container
class TestSessionTokenContainer(ClusterTestBase):
    @pytest.fixture(scope="module")
    def static_sessions(
        self,
        owner_wallet: WalletFile,
        user_wallet: WalletFile,
        client_shell: Shell,
        temp_directory: str,
    ) -> dict[ContainerVerb, str]:
        """
        Returns dict with static session token file paths for all verbs with default lifetime
        """
        return {
            verb: get_container_signed_token(
                owner_wallet, user_wallet, verb, client_shell, temp_directory
            )
            for verb in ContainerVerb
        }

    def test_static_session_token_container_create(
        self,
        owner_wallet: WalletFile,
        user_wallet: WalletFile,
        static_sessions: dict[ContainerVerb, str],
    ):
        """
        Validate static session with create operation
        """
        with allure.step("Create container with static session token"):
            cid = create_container(
                user_wallet.path,
                session_token=static_sessions[ContainerVerb.CREATE],
                shell=self.shell,
                endpoint=self.cluster.default_rpc_endpoint,
                wait_for_creation=False,
            )

        container_info: dict[str, str] = get_container(
            owner_wallet.path, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
        )
        assert container_info["ownerID"] == owner_wallet.get_address()

        assert cid not in list_containers(
            user_wallet.path, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
        )
        assert cid in list_containers(
            owner_wallet.path, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
        )

    def test_static_session_token_container_create_with_other_verb(
        self,
        user_wallet: WalletFile,
        static_sessions: dict[ContainerVerb, str],
    ):
        """
        Validate static session without create operation
        """
        with allure.step("Try create container with static session token without PUT rule"):
            for verb in [verb for verb in ContainerVerb if verb != ContainerVerb.CREATE]:
                with pytest.raises(Exception):
                    create_container(
                        user_wallet.path,
                        session_token=static_sessions[verb],
                        shell=self.shell,
                        endpoint=self.cluster.default_rpc_endpoint,
                        wait_for_creation=False,
                    )

    def test_static_session_token_container_create_with_other_wallet(
        self,
        stranger_wallet: WalletFile,
        static_sessions: dict[ContainerVerb, str],
    ):
        """
        Validate static session with create operation for other wallet
        """
        with allure.step("Try create container with static session token without PUT rule"):
            with pytest.raises(Exception):
                create_container(
                    stranger_wallet.path,
                    session_token=static_sessions[ContainerVerb.CREATE],
                    shell=self.shell,
                    endpoint=self.cluster.default_rpc_endpoint,
                    wait_for_creation=False,
                )

    def test_static_session_token_container_delete(
        self,
        owner_wallet: WalletFile,
        user_wallet: WalletFile,
        static_sessions: dict[ContainerVerb, str],
    ):
        """
        Validate static session with delete operation
        """
        with allure.step("Create container"):
            cid = create_container(
                owner_wallet.path,
                shell=self.shell,
                endpoint=self.cluster.default_rpc_endpoint,
                wait_for_creation=False,
            )
        with allure.step("Delete container with static session token"):
            delete_container(
                wallet=user_wallet.path,
                cid=cid,
                session_token=static_sessions[ContainerVerb.DELETE],
                shell=self.shell,
                endpoint=self.cluster.default_rpc_endpoint,
                await_mode=True,
            )

        assert cid not in list_containers(
            owner_wallet.path, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
        )

    def test_static_session_token_container_set_eacl(
        self,
        owner_wallet: WalletFile,
        user_wallet: WalletFile,
        stranger_wallet: WalletFile,
        static_sessions: dict[ContainerVerb, str],
        simple_object_size,
    ):
        """
        Validate static session with set eacl operation
        """
        with allure.step("Create container"):
            cid = create_container(
                owner_wallet.path,
                basic_acl=PUBLIC_ACL,
                shell=self.shell,
                endpoint=self.cluster.default_rpc_endpoint,
            )
        file_path = generate_file(simple_object_size)
        assert can_put_object(stranger_wallet.path, cid, file_path, self.shell, self.cluster)

        with allure.step(f"Deny all operations for other via eACL"):
            eacl_deny = [
                EACLRule(access=EACLAccess.DENY, role=EACLRole.OTHERS, operation=op)
                for op in EACLOperation
            ]
            set_eacl(
                user_wallet.path,
                cid,
                create_eacl(cid, eacl_deny, shell=self.shell),
                shell=self.shell,
                endpoint=self.cluster.default_rpc_endpoint,
                session_token=static_sessions[ContainerVerb.SETEACL],
            )
            wait_for_cache_expired()

        assert not can_put_object(stranger_wallet.path, cid, file_path, self.shell, self.cluster)