53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
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)
|