# 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)