frostfs-testcases/pytest_tests/testsuites/object/test_object_api_bearer.py

import allure
import pytest
from cluster import Cluster
from common import COMPLEX_OBJ_SIZE, SIMPLE_OBJ_SIZE
from container import REP_2_FOR_3_NODES_PLACEMENT_RULE, SINGLE_PLACEMENT_RULE, create_container
from epoch import get_epoch
from neofs_testlib.shell import Shell
from neofs_verbs import delete_object, get_object
from pytest import FixtureRequest
from python_keywords.acl import EACLAccess, EACLOperation, EACLRole, EACLRule, form_bearertoken_file
from wellknown_acl import EACL_PUBLIC_READ_WRITE

from helpers.container import StorageContainer, StorageContainerInfo
from helpers.test_control import expect_not_raises
from helpers.wallet import WalletFile
from steps.cluster_test_base import ClusterTestBase
from steps.storage_object import StorageObjectInfo


@pytest.fixture(scope="module")
@allure.title("Create bearer token for OTHERS with all operations allowed for all containers")
def bearer_token_file_all_allow(default_wallet: str, client_shell: Shell, cluster: Cluster) -> str:
    bearer = form_bearertoken_file(
        default_wallet,
        "",
        [
            EACLRule(operation=op, access=EACLAccess.ALLOW, role=EACLRole.OTHERS)
            for op in EACLOperation
        ],
        shell=client_shell,
        endpoint=cluster.default_rpc_endpoint,
    )

    return bearer


@pytest.fixture(scope="module")
@allure.title("Create user container for bearer token usage")
def user_container(
    default_wallet: str, client_shell: Shell, cluster: Cluster, request: FixtureRequest
) -> StorageContainer:
    container_id = create_container(
        default_wallet,
        shell=client_shell,
        rule=request.param,
        basic_acl=EACL_PUBLIC_READ_WRITE,
        endpoint=cluster.default_rpc_endpoint,
    )
    # Deliberately using s3gate wallet here to test bearer token
    s3gate = cluster.s3gates[0]
    return StorageContainer(
        StorageContainerInfo(container_id, WalletFile.from_node(s3gate)),
        client_shell,
        cluster,
    )


@pytest.fixture()
def storage_objects(
    user_container: StorageContainer,
    bearer_token_file_all_allow: str,
    request: FixtureRequest,
    client_shell: Shell,
    cluster: Cluster,
) -> list[StorageObjectInfo]:
    epoch = get_epoch(client_shell, cluster)
    storage_objects: list[StorageObjectInfo] = []
    for node in cluster.storage_nodes:
        storage_objects.append(
            user_container.generate_object(
                request.param,
                epoch + 3,
                bearer_token=bearer_token_file_all_allow,
                endpoint=node.get_rpc_endpoint(),
            )
        )
    return storage_objects


@pytest.mark.smoke
@pytest.mark.bearer
class TestObjectApiWithBearerToken(ClusterTestBase):
    @pytest.mark.parametrize(
        "user_container",
        [SINGLE_PLACEMENT_RULE],
        ids=["single replica for all nodes placement rule"],
        indirect=True,
    )
    @pytest.mark.parametrize(
        "storage_objects",
        [SIMPLE_OBJ_SIZE, COMPLEX_OBJ_SIZE],
        ids=["simple object", "complex object"],
        indirect=True,
    )
    def test_delete_object_with_s3_wallet_bearer(
        self,
        storage_objects: list[StorageObjectInfo],
        bearer_token_file_all_allow: str,
        request: FixtureRequest,
    ):
        allure.dynamic.title(
            f"Check that objects can be deleted from any node using s3gate wallet with bearer token for {request.node.callspec.id}"
        )

        s3_gate_wallet = self.cluster.s3gates[0]
        with allure.step("Try to delete each object from first storage node"):
            for storage_object in storage_objects:
                with expect_not_raises():
                    delete_object(
                        s3_gate_wallet.get_wallet_path(),
                        storage_object.cid,
                        storage_object.oid,
                        self.shell,
                        endpoint=self.cluster.default_rpc_endpoint,
                        bearer=bearer_token_file_all_allow,
                        wallet_config=s3_gate_wallet.get_wallet_config_path(),
                    )

    @pytest.mark.parametrize(
        "user_container",
        [REP_2_FOR_3_NODES_PLACEMENT_RULE],
        ids=["2 replicas for 3 nodes placement rule"],
        indirect=True,
    )
    @pytest.mark.parametrize(
        "file_size",
        [SIMPLE_OBJ_SIZE, COMPLEX_OBJ_SIZE],
        ids=["simple object", "complex object"],
    )
    def test_get_object_with_s3_wallet_bearer_from_all_nodes(
        self,
        user_container: StorageContainer,
        file_size: int,
        bearer_token_file_all_allow: str,
        request: FixtureRequest,
    ):
        allure.dynamic.title(
            f"Check that objects can be deleted from any node using s3gate wallet with bearer token for {request.node.callspec.id}"
        )

        s3_gate_wallet = self.cluster.s3gates[0]
        with allure.step("Put one object to container"):
            epoch = self.get_epoch()
            storage_object = user_container.generate_object(
                file_size, epoch + 3, bearer_token=bearer_token_file_all_allow
            )

        with allure.step("Try to fetch object from each storage node"):
            for node in self.cluster.storage_nodes:
                with expect_not_raises():
                    get_object(
                        s3_gate_wallet.get_wallet_path(),
                        storage_object.cid,
                        storage_object.oid,
                        self.shell,
                        endpoint=node.get_rpc_endpoint(),
                        bearer=bearer_token_file_all_allow,
                        wallet_config=s3_gate_wallet.get_wallet_config_path(),
                    )