rjmcmahon
/
FiWiControl
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
FiWiControl/tests/test_fabric_instantiation.py

235 lines
8.9 KiB
Python
Raw Permalink Blame History

# Copyright (c) 2026 Umber
#
# Licensed under the Apache License, Version 2.0; see LICENSE.
"""Runtime :class:`~fiwicontrol.fabric.fabric.Fabric` construction only — never :meth:`~fiwicontrol.fabric.fabric.Fabric.realize`.
INI-backed test: ``configs/default.ini`` unless ``FIWI_TEST_INI`` is set to another path.
Run as a script: ``python3 tests/test_fabric_instantiation.py [INI]`` (default INI when argv omitted) prints a short report (INI path, site, machines, :class:`FabricDefinition` fields, then runtime :class:`Fabric`).
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
_REPO = Path(__file__).resolve().parents[1]
_SRC = str(_REPO / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from fiwicontrol.fabric import Fabric
from fiwicontrol.fabric.definition_from_ini import compose_definition, read_inventory_ini
from fiwicontrol.fabric.fabric import FabricDefinition, FabricRRHBinding
from fiwicontrol.fabric.ini_merge import FabricIniOverlay
from fiwicontrol.lab.inventory_config import LabIniDocument
from fiwicontrol.fabric.fingerprint import acroname_modules_fingerprint
from fiwicontrol.lab.discovery import AcronameModuleInfo
def _one_module() -> list[AcronameModuleInfo]:
return [
AcronameModuleInfo(
transport="USB",
serial_number=111,
module_address=0,
model_id=42,
model_name="USBHub3p",
model_description="",
stem_class="USBHub3p",
downstream_usb_ports=4,
hub_port_entities=4,
)
]
def test_instantiate_fabric_from_definition_without_realize() -> None:
fp = acroname_modules_fingerprint(_one_module())
definition = FabricDefinition(
fabric_id="bench-a",
discovery_fingerprint=fp,
rrhs=(FabricRRHBinding(radio_id="rrh-1", acroname_port=2, acroname_module_serial=111),),
concentrator_name="head",
concentrator_ipaddr="10.0.0.5",
)
fab = Fabric.from_definition(definition)
try:
assert fab.fabric_id == "bench-a"
assert fab.expected_discovery_fingerprint == fp
assert fab.rrh_power_ports == {"rrh-1": 2}
assert fab.concentrator is not None
assert fab.concentrator.name == "head" and fab.concentrator.ipaddr == "10.0.0.5"
assert len(fab.rrhs) == 1 and fab.rrhs[0].radio_id == "rrh-1"
assert fab in Fabric.get_instances()
finally:
fab.destroy()
assert fab not in Fabric.get_instances()
def test_instantiate_fabric_from_json_path_without_realize(tmp_path: Path) -> None:
fp = acroname_modules_fingerprint(_one_module())
path = tmp_path / "fabric.json"
FabricDefinition(
fabric_id="json-id",
discovery_fingerprint=fp,
rrhs=(FabricRRHBinding(radio_id="z", acroname_port=1, acroname_module_serial=111),),
concentrator_ipaddr="192.168.0.1",
lab_ini=str(tmp_path / "source.ini"),
).save(path)
fab = Fabric.from_json_path(path)
try:
assert fab.fabric_id == "json-id"
assert fab.rrh_power_ports == {"z": 1}
assert fab.concentrator is not None and fab.concentrator.ipaddr == "192.168.0.1"
finally:
fab.destroy()
def test_instantiate_fabric_from_definition_no_concentrator_without_realize() -> None:
fp = acroname_modules_fingerprint(_one_module())
definition = FabricDefinition(
fabric_id="no-ssh",
discovery_fingerprint=fp,
rrhs=(FabricRRHBinding(radio_id="only-rrh", acroname_port=0, acroname_module_serial=111),),
concentrator_name=None,
concentrator_ipaddr=None,
)
fab = Fabric.from_definition(definition)
try:
assert fab.concentrator is None
assert len(fab.rrhs) == 1
finally:
fab.destroy()
def _compose_fabric_from_ini(
config_path: Path,
) -> tuple[Fabric, FabricDefinition, LabIniDocument, FabricIniOverlay]:
doc, overlay, rrhs = read_inventory_ini(config_path)
definition = compose_definition(
doc,
overlay,
rrhs,
_one_module(),
lab_ini=str(config_path.resolve()),
)
return Fabric.from_definition(definition), definition, doc, overlay
def _instantiate_from_ini_path(config_path: Path) -> Fabric:
fab, _, _, _ = _compose_fabric_from_ini(config_path)
return fab
def test_instantiate_fabric_from_config_without_realize(capsys) -> None:
"""Load INI from ``FIWI_TEST_INI`` if set, else ``configs/default.ini`` under the repo; then ``Fabric.from_definition``."""
raw = os.environ.get("FIWI_TEST_INI")
config_path = Path(raw).expanduser() if raw else (_REPO / "configs" / "default.ini")
assert config_path.is_file(), "missing INI: {} (set FIWI_TEST_INI or add configs/default.ini)".format(config_path)
fab = _instantiate_from_ini_path(config_path)
try:
assert fab.fabric_id.strip()
assert len(fab.rrhs) >= 1
text = str(fab)
print(fab)
assert capsys.readouterr().out.strip() == text.strip()
finally:
fab.destroy()
def test_instantiate_with_concentrator_override_without_realize() -> None:
fp = acroname_modules_fingerprint(_one_module())
definition = FabricDefinition(
fabric_id="override-me",
discovery_fingerprint=fp,
rrhs=(FabricRRHBinding(radio_id="r", acroname_port=0, acroname_module_serial=111),),
)
fab = Fabric.from_definition(definition)
fab2: Fabric | None = None
try:
assert fab.concentrator is None
fab2 = fab.with_concentrator_override(name="c", ipaddr="10.10.10.10")
assert fab2.concentrator is not None
assert fab2.concentrator.name == "c" and fab2.concentrator.ipaddr == "10.10.10.10"
assert fab2.rrhs == fab.rrhs and fab2.fabric_id == fab.fabric_id
finally:
if fab2 is not None:
fab2.destroy()
fab.destroy()
def _print_fabric_instantiation_report(config_path: Path) -> None:
"""Human-readable summary for ``python3 tests/test_fabric_instantiation.py`` (no ``realize``)."""
fab, definition, doc, ini_overlay = _compose_fabric_from_ini(config_path)
try:
print("=== Fabric instantiation (no realize) ===")
print("INI:", config_path.resolve())
print("Site:", doc.site_name if doc.site_name else "(none)")
print("Machines:", ", ".join(sorted(doc.nodes)) if doc.nodes else "(none)")
print()
print("[fabric] INI overlay (plant / harness metadata, not in FabricDefinition JSON)")
print(
" concentrator_adnacom_adapter_count (Adnacom adapters on concentrator):",
ini_overlay.concentrator_adnacom_adapter_count
if ini_overlay.concentrator_adnacom_adapter_count is not None
else "(not set)",
)
print(
" patch_panel_ports:",
ini_overlay.patch_panel_ports if ini_overlay.patch_panel_ports is not None else "(not set)",
)
print(" fabric.rrh.* override sections:", len(ini_overlay.rrh_overrides))
print()
print(
"Expected Acroname + Monsoon/HVPM (per [machine.*] row — ``monsoon`` / ``hvpm`` in INI → hvpm_count; vs discovery when you verify)"
)
for h in doc.hosts:
ac = ", ".join("{}:{}".format(stem, n) for stem, n in h.expected_acronames) or "(none)"
relay = h.relay_node_id or "-"
print(
" machine {!r}: mode={} relay={} acroname={} monsoon_hvpm_expected={}".format(
h.name,
h.mode,
relay,
ac,
h.hvpm_count,
)
)
print()
print("FabricDefinition")
print(" fabric_id:", definition.fabric_id)
print(" discovery_fingerprint:", definition.discovery_fingerprint)
print(" concentrator_name:", definition.concentrator_name)
print(" concentrator_ipaddr:", definition.concentrator_ipaddr)
print(" lab_ini:", definition.lab_ini)
print(" rrhs ({} rows):".format(len(definition.rrhs)))
for b in definition.rrhs:
print(
" - radio_id={!r} acroname_port={} acroname_module_serial={} patch_panel_port={}".format(
b.radio_id,
b.acroname_port,
b.acroname_module_serial,
b.patch_panel_port,
)
)
print()
print("Runtime Fabric")
print(" ", fab)
print(" rrh_power_ports:", fab.rrh_power_ports)
if fab.concentrator is not None:
print(" concentrator ssh:", fab.concentrator.name, fab.concentrator.ipaddr)
else:
print(" concentrator ssh: (none)")
finally:
fab.destroy()
if __name__ == "__main__":
_cfg = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else (_REPO / "configs" / "default.ini").resolve()
if not _cfg.is_file():
print("INI not found:", _cfg, file=sys.stderr)
raise SystemExit(2)
_print_fabric_instantiation_report(_cfg)
Reference in New Issue View Git Blame Copy Permalink