Use one Host object per ContainerizedService

get_shell() method is bound to the Host object. If we put all
ContainerizedService's into a single Host object we won't be able to
shell into specific containers.

Signed-off-by: Vitaliy Potyarkin <v.potyarkin@yadro.com>
This commit is contained in:
Vitaliy Potyarkin 2025-05-12 12:32:28 +03:00
parent cf950f6313
commit adbcd91253

View file

@ -4,7 +4,7 @@ Based on testcontainers and Docker.
"""
from pathlib import Path
from typing import Any
from typing import Any, Mapping
from frostfs_testlib.credentials.interfaces import GrpcCredentialsProvider, User
from frostfs_testlib.hosting.interfaces import Host
@ -20,9 +20,14 @@ def dynamic_hosting_config(**fixtures) -> dict[str, Any]:
"""
Translate ContainerizedService fixtures into a hosting configuration tree.
"""
config = {
"hosts": [
{
config = {"hosts": []}
for name, containers in fixtures.items():
config["hosts"].extend(_host_config(name, containers, fixtures))
return config
def _host_config(name: str, containers: list, fixtures: Mapping[str, Any]) -> None:
host_template = {
"plugin_name": "component_tests",
"grpc_creds_plugin_name": "component_tests",
"healthcheck_plugin_name": "basic",
@ -37,25 +42,38 @@ def dynamic_hosting_config(**fixtures) -> dict[str, Any]:
"services": [],
"clis": [],
}
],
}
services = config["hosts"][0]["services"]
for name, nodes in fixtures.items():
services.extend(_services(name, nodes))
return config
def _services(name: str, nodes: list) -> None:
if name == "storage":
for index, node in enumerate(nodes, 1):
yield {
for index, container in enumerate(containers, 1):
host = _from_template(host_template)
host["services"].append(
{
"name": f"frostfs-storage_{index:02}",
"attributes": {
"control_endpoint": f"{node.ip}:8801",
"endpoint_data0": f"{node.ip}:8802",
"endpoint_prometheus": f"{node.ip}:9090",
"control_endpoint": f"{container.ip}:8801",
"endpoint_data0": f"{container.ip}:8802",
"endpoint_prometheus": f"{container.ip}:9090",
},
}
)
yield host
def _from_template(template: Mapping[str, Any]) -> Mapping[str, Any]:
"""Like deepcopy but does not dive into custom objects with pickle."""
result = {}
for key, value in template.items():
if isinstance(value, Mapping):
result[key] = _from_template(value)
continue
try:
copy = value.copy()
if not isinstance(copy, type(value)):
raise ValueError("copy() method did not return an object of the same type")
except Exception:
copy = value
result[key] = copy
return result
class ContainerHost(Host):