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

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.cli import FrostfsCli
from frostfs_testlib.resources.cli import FROSTFS_CLI_EXEC
from frostfs_testlib.shell import Shell
from frostfs_testlib.steps.cli.container import (
    REP_2_FOR_3_NODES_PLACEMENT_RULE,
    SINGLE_PLACEMENT_RULE,
    StorageContainer,
    StorageContainerInfo,
)
from frostfs_testlib.steps.cli.object import delete_object, get_object
from frostfs_testlib.steps.storage_object import StorageObjectInfo
from frostfs_testlib.storage.cluster import Cluster
from frostfs_testlib.storage.dataclasses import ape
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper
from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.testing.test_control import expect_not_raises
from pytest import FixtureRequest

from ...helpers.bearer_token import create_bearer_token
from ...helpers.container_access import assert_full_access_to_container
from ...helpers.container_creation import create_container_with_ape
from ...helpers.container_request import ContainerRequest


@pytest.fixture(scope="session")
@allure.title("Create user container for bearer token usage")
def user_container(
    frostfs_cli: FrostfsCli, default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, rpc_endpoint: str, request: FixtureRequest
) -> StorageContainer:
    policy = request.param if "param" in request.__dict__ else SINGLE_PLACEMENT_RULE
    container_request = ContainerRequest(policy)
    container_id = create_container_with_ape(container_request, frostfs_cli, default_wallet, client_shell, cluster, rpc_endpoint)

    # Deliberately using s3gate wallet here to test bearer token
    s3_gate_wallet = WalletInfo.from_node(cluster.s3_gates[0])
    return StorageContainer(StorageContainerInfo(container_id, s3_gate_wallet), client_shell, cluster)


@pytest.fixture(scope="session")
@allure.title("Create bearer token with allowed put for container")
def bearer_token(frostfs_cli: FrostfsCli, temp_directory: str, user_container: StorageContainer, cluster: Cluster) -> str:
    rule = ape.Rule(ape.Verb.ALLOW, ape.ObjectOperations.WILDCARD_ALL)
    return create_bearer_token(frostfs_cli, temp_directory, user_container.get_id(), rule, cluster.default_rpc_endpoint)


@pytest.fixture(scope="session")
def grpc_client_with_other_wallet(client_shell: Shell, other_wallet: WalletInfo) -> GrpcClientWrapper:
    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, other_wallet.config_path))


@pytest.fixture()
def storage_objects(
    user_container: StorageContainer,
    bearer_token: str,
    object_size: ObjectSize,
    cluster: Cluster,
) -> list[StorageObjectInfo]:
    storage_objects: list[StorageObjectInfo] = []
    for node in cluster.storage_nodes:
        storage_objects.append(
            user_container.generate_object(
                object_size.value,
                bearer_token=bearer_token,
                endpoint=node.get_rpc_endpoint(),
            )
        )
    return storage_objects


@pytest.mark.nightly
@pytest.mark.bearer
@pytest.mark.ape
@pytest.mark.grpc_api
class TestObjectApiWithBearerToken(ClusterTestBase):
    @allure.title("Object can be deleted from any node using s3gate wallet with bearer token (obj_size={object_size})")
    @pytest.mark.parametrize(
        "user_container",
        [SINGLE_PLACEMENT_RULE],
        indirect=True,
    )
    def test_delete_object_with_s3_wallet_bearer(
        self,
        storage_objects: list[StorageObjectInfo],
        bearer_token: str,
    ):
        s3_gate_wallet = WalletInfo.from_node(self.cluster.s3_gates[0])
        with reporter.step("Delete each object from first storage node"):
            for storage_object in storage_objects:
                with expect_not_raises():
                    delete_object(
                        s3_gate_wallet,
                        storage_object.cid,
                        storage_object.oid,
                        self.shell,
                        endpoint=self.cluster.default_rpc_endpoint,
                        bearer=bearer_token,
                    )

    @allure.title("Object can be fetched from any node using s3gate wallet with bearer token (obj_size={object_size})")
    @pytest.mark.parametrize(
        "user_container",
        [REP_2_FOR_3_NODES_PLACEMENT_RULE],
        indirect=True,
    )
    def test_get_object_with_s3_wallet_bearer_from_all_nodes(
        self,
        user_container: StorageContainer,
        object_size: ObjectSize,
        bearer_token: str,
    ):
        s3_gate_wallet = WalletInfo.from_node(self.cluster.s3_gates[0])
        with reporter.step("Put object to container"):
            storage_object = user_container.generate_object(
                object_size.value,
                bearer_token=bearer_token,
                endpoint=self.cluster.default_rpc_endpoint,
            )

        with reporter.step("Get object from each storage node"):
            for node in self.cluster.storage_nodes:
                with expect_not_raises():
                    get_object(
                        s3_gate_wallet,
                        storage_object.cid,
                        storage_object.oid,
                        self.shell,
                        node.get_rpc_endpoint(),
                        bearer_token,
                    )

    # TODO: Without PATCH operation,
    # since it requires specific permissions that do not apply when testing all operations at once
    @allure.title("Wildcard APE rule contains all permissions (obj_size={object_size})")
    def test_ape_wildcard_contains_all_rules(
        self,
        other_wallet: WalletInfo,
        storage_objects: list[StorageObjectInfo],
        bearer_token: str,
    ):
        obj = storage_objects.pop()
        with reporter.step("Assert all operations available with object"):
            assert_full_access_to_container(other_wallet, obj.cid, obj.oid, obj.file_path, self.shell, self.cluster, bearer_token)

    # ^
    @allure.title("Wildcard APE rule contains PATCH permission (obj_size={object_size})")
    def test_ape_wildcard_contains_patch_rule(
        self,
        grpc_client_with_other_wallet: GrpcClientWrapper,
        storage_objects: list[StorageObjectInfo],
        bearer_token: str,
    ):
        obj = storage_objects.pop()
        with reporter.step("Verify patch is available"):
            patched_oid = grpc_client_with_other_wallet.object.patch(
                obj.cid,
                obj.oid,
                self.cluster.default_rpc_endpoint,
                ranges=["99:88"],
                payloads=[obj.file_path],
                new_attrs="test-attribute=100",
                bearer=bearer_token,
                timeout="200s",
            )
            assert patched_oid != obj.oid, "OID of patched object must be different from original one"