frostfs-testcases/pytest_tests/testsuites/shard/test_control_shard.py

import json
import pathlib

import allure
import pytest
from frostfs_testlib.cli import FrostfsCli
from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
from frostfs_testlib.resources.common import DEFAULT_WALLET_CONFIG
from frostfs_testlib.storage.cluster import Cluster, StorageNode
from frostfs_testlib.storage.dataclasses.shard import Shard


@pytest.mark.shard
class TestControlShard:
    @staticmethod
    def get_shards_from_config(node: StorageNode) -> list[Shard]:
        config_file = node.get_shard_config_path()
        file_type = pathlib.Path(config_file).suffix

        parser_method = {
            ".env": node.get_shards_from_env,
            ".yaml": node.get_shards,
            ".yml": node.get_shards,
        }

        shards = parser_method[file_type]()
        return shards

    @staticmethod
    def get_shards_from_cli(node: StorageNode) -> list[Shard]:
        wallet_path = node.get_remote_wallet_path()
        wallet_password = node.get_wallet_password()
        control_endpoint = node.get_control_endpoint()

        cli_config = node.host.get_cli_config("frostfs-cli")

        cli = FrostfsCli(node.host.get_shell(), cli_config.exec_path, DEFAULT_WALLET_CONFIG)
        result = cli.shards.list(
            endpoint=control_endpoint,
            wallet=wallet_path,
            wallet_password=wallet_password,
            json_mode=True,
            timeout=CLI_DEFAULT_TIMEOUT,
        )
        return [Shard.from_object(shard) for shard in json.loads(result.stdout.split(">", 1)[1])]

    @allure.title("All shards are available")
    def test_control_shard(self, cluster: Cluster):
        for storage_node in cluster.storage_nodes:
            shards_from_config = self.get_shards_from_config(storage_node)
            shards_from_cli = self.get_shards_from_cli(storage_node)

            assert set(shards_from_config) == set(shards_from_cli)