frostfs-testcases/pytest_tests/testsuites/access/conftest.py

import json

import pytest
from frostfs_testlib import reporter
from frostfs_testlib.cli.frostfs_cli.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 search_nodes_with_container
from frostfs_testlib.steps.cli.object import put_object_to_random_node
from frostfs_testlib.storage.cluster import Cluster, ClusterNode
from frostfs_testlib.storage.dataclasses import ape
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.parallel import parallel


@pytest.fixture(scope="session")
def ir_wallet(cluster: Cluster) -> WalletInfo:
    return WalletInfo.from_node(cluster.ir_nodes[0])


@pytest.fixture(scope="session")
def storage_wallet(cluster: Cluster) -> WalletInfo:
    return WalletInfo.from_node(cluster.storage_nodes[0])


@pytest.fixture(scope="session")
def role(request: pytest.FixtureRequest):
    return request.param


@pytest.fixture(scope="session")
def test_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.Role):
    role_to_wallet_map = {
        ape.Role.OWNER: default_wallet,
        ape.Role.OTHERS: other_wallet,
    }

    assert role in role_to_wallet_map, "Missing wallet with role {role}"

    return role_to_wallet_map[role]


@pytest.fixture(scope="function", params=[5])
def objects(
    container: str,
    default_wallet: WalletInfo,
    client_shell: Shell,
    cluster: Cluster,
    test_file: str,
    request: pytest.FixtureRequest,
):
    object_count = request.param

    with reporter.step("Add test objects to container"):
        put_results = parallel(
            [put_object_to_random_node] * object_count,
            wallet=default_wallet,
            path=test_file,
            cid=container,
            shell=client_shell,
            cluster=cluster,
        )
        objects_oids = [put_result.result() for put_result in put_results]

    return objects_oids


@pytest.fixture
def container_nodes(default_wallet: WalletInfo, container: str, client_shell: Shell, cluster: Cluster) -> list[ClusterNode]:
    cid = container
    container_holder_nodes = search_nodes_with_container(default_wallet, cid, client_shell, cluster.default_rpc_endpoint, cluster)

    report_data = {node.id: node.host_ip for node in container_holder_nodes}
    reporter.attach(json.dumps(report_data, indent=2), "container_nodes.json")

    return container_holder_nodes


@pytest.fixture
def container_node_wallet(container_nodes: list[ClusterNode]) -> WalletInfo:
    return WalletInfo.from_node(container_nodes[0].storage_node)


@pytest.fixture
def grpc_client_with_container_wallet(client_shell: Shell, container_node_wallet: WalletInfo) -> GrpcClientWrapper:
    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, container_node_wallet.config_path))


@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(scope="session")
def grpc_client_with_ir_wallet(client_shell: Shell, ir_wallet: WalletInfo) -> GrpcClientWrapper:
    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, ir_wallet.config_path))