529 lines
17 KiB
Python
529 lines
17 KiB
Python
# Copyright (c) 2026 Umber
|
|
#
|
|
# Licensed under the Apache License, Version 2.0; see LICENSE.
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from fiwicontrol.fabric import Fabric
|
|
from fiwicontrol.fabric.definition_from_ini import _rrhs_with_resolved_serials, compose_definition, read_inventory_ini
|
|
from fiwicontrol.fabric.fingerprint import acroname_modules_fingerprint
|
|
from fiwicontrol.fabric.ini_merge import fabric_rrh_bindings_from_ini, merge_fabric_definition_with_ini, parse_fabric_ini_overlay
|
|
from fiwicontrol.fabric.patch_panel_json import load_patch_panel_bdf_map
|
|
from fiwicontrol.fabric.fabric import (
|
|
FabricBindingStatus,
|
|
FabricDefinition,
|
|
FabricRRHBinding,
|
|
)
|
|
from fiwicontrol.lab.discovery import AcronameModuleInfo
|
|
from fiwicontrol.lab.inventory_config import load_inventory_ini
|
|
|
|
|
|
def _sample_modules() -> 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_fabric_rrh_bindings_from_ini(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[host.h]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric.rrh.z]",
|
|
"acroname_port = 2",
|
|
"acroname_module_serial = 999",
|
|
"[fabric.rrh.a]",
|
|
"acroname_port = 0",
|
|
"patch_panel_port = 5",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
rrhs = fabric_rrh_bindings_from_ini(ini)
|
|
assert [h.radio_id for h in rrhs] == ["a", "z"]
|
|
assert rrhs[0].acroname_port == 0 and rrhs[0].patch_panel_port == 5
|
|
assert rrhs[1].acroname_module_serial == 999
|
|
|
|
|
|
def test_fabric_definition_lab_ini_roundtrip(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="t",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
lab_ini="/tmp/lab.ini",
|
|
)
|
|
p = tmp_path / "f.json"
|
|
d.save(p)
|
|
d2 = FabricDefinition.load(p)
|
|
assert d2.lab_ini == "/tmp/lab.ini"
|
|
|
|
|
|
def test_fabric_definition_roundtrip(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="t",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(
|
|
FabricRRHBinding(radio_id="rrh-01", acroname_port=0, acroname_module_serial=111, patch_panel_port=10),
|
|
FabricRRHBinding(radio_id="rrh-02", acroname_port=1, acroname_module_serial=111),
|
|
),
|
|
concentrator_name="rig",
|
|
concentrator_ipaddr="10.0.0.1",
|
|
)
|
|
path = tmp_path / "fabric.json"
|
|
d.save(path)
|
|
d2 = FabricDefinition.load(path)
|
|
assert d2.fabric_id == "t"
|
|
assert d2.discovery_fingerprint == fp
|
|
assert len(d2.rrhs) == 2
|
|
assert d2.rrhs[0].radio_id == "rrh-01" and d2.rrhs[0].acroname_port == 0
|
|
assert d2.concentrator_ipaddr == "10.0.0.1"
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
assert raw["schema_version"] == FabricDefinition.JSON_SCHEMA_VERSION
|
|
|
|
|
|
def test_fabric_from_json_path(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
FabricDefinition(
|
|
fabric_id="t2",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
concentrator_ipaddr="192.168.1.2",
|
|
).save(tmp_path / "f.json")
|
|
|
|
f = Fabric.from_json_path(tmp_path / "f.json")
|
|
assert f.fabric_id == "t2"
|
|
assert f.rrh_power_ports == {"a": 0}
|
|
assert f.concentrator is not None and f.concentrator.ipaddr == "192.168.1.2"
|
|
assert f.expected_discovery_fingerprint == fp
|
|
|
|
|
|
def test_binding_cache_status_missing(tmp_path: Path) -> None:
|
|
assert Fabric.binding_cache_status(tmp_path / "nope.json") is FabricBindingStatus.MISSING
|
|
|
|
|
|
def test_binding_cache_status_invalid(tmp_path: Path) -> None:
|
|
p = tmp_path / "bad.json"
|
|
p.write_text("{not json", encoding="utf-8")
|
|
assert Fabric.binding_cache_status(p) is FabricBindingStatus.INVALID
|
|
|
|
|
|
def test_validate_duplicate_radio_id() -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="x",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(
|
|
FabricRRHBinding(radio_id="same", acroname_port=0, acroname_module_serial=111),
|
|
FabricRRHBinding(radio_id="same", acroname_port=1, acroname_module_serial=111),
|
|
),
|
|
)
|
|
with pytest.raises(ValueError, match="duplicate radio_id"):
|
|
d.validate()
|
|
|
|
|
|
def test_validate_duplicate_port_same_module() -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="x",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(
|
|
FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),
|
|
FabricRRHBinding(radio_id="b", acroname_port=0, acroname_module_serial=111),
|
|
),
|
|
)
|
|
with pytest.raises(ValueError, match="duplicate"):
|
|
d.validate()
|
|
|
|
|
|
def test_rrhs_with_resolved_serials_preserves_pcie_bdf() -> None:
|
|
rrhs = (
|
|
FabricRRHBinding(
|
|
radio_id="a",
|
|
acroname_port=0,
|
|
acroname_module_serial=None,
|
|
pcie_bdf="0000:03:00.0",
|
|
),
|
|
)
|
|
out = _rrhs_with_resolved_serials(rrhs, _sample_modules())
|
|
assert out[0].acroname_module_serial == 111
|
|
assert out[0].pcie_bdf == "0000:03:00.0"
|
|
|
|
|
|
def test_merge_lab_ini_concentrator(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[site]",
|
|
"name = SiteX",
|
|
"[node.pi]",
|
|
"ipaddr = 10.0.0.7",
|
|
"sshtype = ssh",
|
|
"[host.h1]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric]",
|
|
"fabric_id = FromIni",
|
|
"concentrator_node = pi",
|
|
"patch_panel_ports = 48",
|
|
"[fabric.rrh.a]",
|
|
"acroname_port = 3",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
FabricDefinition(
|
|
fabric_id="from-json",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(
|
|
FabricRRHBinding(
|
|
radio_id="a",
|
|
acroname_port=0,
|
|
acroname_module_serial=111,
|
|
pcie_bdf="0000:01:00.0",
|
|
),
|
|
),
|
|
concentrator_ipaddr="1.1.1.1",
|
|
patch_panel_ports=8,
|
|
lab_ini=str(tmp_path / "source.ini"),
|
|
).save(tmp_path / "f.json")
|
|
|
|
merged = FabricDefinition.load_json_merged_with_ini(tmp_path / "f.json", ini)
|
|
assert merged.fabric_id == "FromIni"
|
|
assert merged.concentrator_ipaddr == "10.0.0.7"
|
|
assert merged.rrhs[0].acroname_port == 3
|
|
assert merged.rrhs[0].pcie_bdf == "0000:01:00.0"
|
|
assert merged.patch_panel_ports == 48
|
|
assert merged.lab_ini == str(tmp_path / "source.ini")
|
|
|
|
|
|
def test_merge_lab_ini_keeps_json_patch_panel_when_ini_omits(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[site]",
|
|
"name = SiteX",
|
|
"[node.pi]",
|
|
"ipaddr = 10.0.0.7",
|
|
"sshtype = ssh",
|
|
"[host.h1]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric]",
|
|
"concentrator_node = pi",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
d = FabricDefinition(
|
|
fabric_id="j",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
patch_panel_ports=24,
|
|
)
|
|
merged = merge_fabric_definition_with_ini(d, ini)
|
|
assert merged.patch_panel_ports == 24
|
|
|
|
|
|
def test_fabric_realize_registered_runs_all_registered() -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="r",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
)
|
|
fab = Fabric.from_definition(d)
|
|
assert fab in Fabric.get_instances()
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"fiwicontrol.lab.discovery.discover_acroname_modules_async",
|
|
new=AsyncMock(return_value=_sample_modules()),
|
|
):
|
|
out = await Fabric.realize_registered(strict=True, fabrics="all", timeout=30.0)
|
|
assert len(out) >= 1 and fab in out
|
|
|
|
asyncio.run(_run())
|
|
fab.destroy()
|
|
assert fab not in Fabric.get_instances()
|
|
|
|
|
|
def test_fabric_realize_accepts_matching_fingerprint() -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
d = FabricDefinition(
|
|
fabric_id="r",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
)
|
|
fab = Fabric.from_definition(d)
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"fiwicontrol.lab.discovery.discover_acroname_modules_async",
|
|
new=AsyncMock(return_value=_sample_modules()),
|
|
):
|
|
out = await fab.realize()
|
|
assert out is fab
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_fabric_realize_strict_rejects_stale_fingerprint() -> None:
|
|
d = FabricDefinition(
|
|
fabric_id="r",
|
|
discovery_fingerprint="0" * 64,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
)
|
|
fab = Fabric.from_definition(d)
|
|
assert fab.expected_discovery_fingerprint == "0" * 64
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"fiwicontrol.lab.discovery.discover_acroname_modules_async",
|
|
new=AsyncMock(return_value=_sample_modules()),
|
|
):
|
|
with pytest.raises(ValueError, match="fingerprint"):
|
|
await fab.realize(strict=True)
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_fabric_realize_non_strict_logs_fdir_warning(caplog: pytest.LogCaptureFixture) -> None:
|
|
d = FabricDefinition(
|
|
fabric_id="r",
|
|
discovery_fingerprint="0" * 64,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
)
|
|
fab = Fabric.from_definition(d)
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"fiwicontrol.lab.discovery.discover_acroname_modules_async",
|
|
new=AsyncMock(return_value=_sample_modules()),
|
|
):
|
|
with caplog.at_level(logging.WARNING):
|
|
out = await fab.realize(strict=False)
|
|
assert out is fab
|
|
|
|
asyncio.run(_run())
|
|
assert "FiWi-FDIR" in caplog.text
|
|
assert "strict=False" in caplog.text
|
|
|
|
|
|
def test_fabric_realize_non_strict_ignores_mismatch() -> None:
|
|
d = FabricDefinition(
|
|
fabric_id="r",
|
|
discovery_fingerprint="0" * 64,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
)
|
|
fab = Fabric.from_definition(d)
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"fiwicontrol.lab.discovery.discover_acroname_modules_async",
|
|
new=AsyncMock(return_value=_sample_modules()),
|
|
):
|
|
out = await fab.realize(strict=False)
|
|
assert out is fab
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_compose_definition_fills_omitted_hub_serial(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[site]",
|
|
"name = LabX",
|
|
"[host.h]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric]",
|
|
"fabric_id = t",
|
|
"[fabric.rrh.a]",
|
|
"acroname_port = 0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
doc, overlay, rrhs = read_inventory_ini(ini)
|
|
assert rrhs[0].acroname_module_serial is None
|
|
d = compose_definition(doc, overlay, rrhs, _sample_modules(), lab_ini=str(ini.resolve()))
|
|
assert d.rrhs[0].acroname_module_serial == 111
|
|
|
|
|
|
def test_compose_definition_rejects_ambiguous_hub_serial(tmp_path: Path) -> None:
|
|
m1 = _sample_modules()[0]
|
|
m2 = AcronameModuleInfo(
|
|
transport="USB",
|
|
serial_number=222,
|
|
module_address=0,
|
|
model_id=42,
|
|
model_name="USBHub3p",
|
|
model_description="",
|
|
stem_class="USBHub3p",
|
|
downstream_usb_ports=4,
|
|
hub_port_entities=4,
|
|
)
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[host.h]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric.rrh.a]",
|
|
"acroname_port = 0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
doc, overlay, rrhs = read_inventory_ini(ini)
|
|
with pytest.raises(ValueError, match="Multiple Acroname"):
|
|
compose_definition(doc, overlay, rrhs, [m1, m2], lab_ini=str(ini.resolve()))
|
|
|
|
|
|
def test_compose_definition_pure(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[site]",
|
|
"name = LabX",
|
|
"[host.h]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric]",
|
|
"fabric_id = from-overlay",
|
|
"[fabric.rrh.z]",
|
|
"acroname_port = 2",
|
|
"acroname_module_serial = 999",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
doc, overlay, rrhs = read_inventory_ini(ini)
|
|
assert overlay.fabric_id == "from-overlay"
|
|
mods = _sample_modules()
|
|
d = compose_definition(
|
|
doc,
|
|
overlay,
|
|
rrhs,
|
|
mods,
|
|
lab_ini=str(ini.resolve()),
|
|
)
|
|
assert d.fabric_id == "from-overlay"
|
|
assert d.discovery_fingerprint == acroname_modules_fingerprint(mods)
|
|
assert len(d.rrhs) == 1 and d.rrhs[0].radio_id == "z"
|
|
|
|
|
|
def test_read_inventory_ini_matches_fabric_rrh_bindings_from_ini(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[host.h]",
|
|
"mode = local",
|
|
"acroname = USBHub2x4:4",
|
|
"hvpm = HVPM:0",
|
|
"[fabric.rrh.a]",
|
|
"acroname_port = 0",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
doc, overlay, rrhs = read_inventory_ini(ini)
|
|
assert load_inventory_ini(ini).hosts == doc.hosts
|
|
assert parse_fabric_ini_overlay(ini).fabric_id == overlay.fabric_id
|
|
assert rrhs == fabric_rrh_bindings_from_ini(ini)
|
|
|
|
|
|
def test_parse_fabric_ini_overlay_adnacom_adapter_count(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[fabric]",
|
|
"concentrator_adnacom_adapter_count = 6",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
o = parse_fabric_ini_overlay(ini)
|
|
assert o.concentrator_adnacom_adapter_count == 6
|
|
|
|
ini.write_text(
|
|
"\n".join(
|
|
[
|
|
"[fabric]",
|
|
"adnacom_adapter_count = 3",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert parse_fabric_ini_overlay(ini).concentrator_adnacom_adapter_count == 3
|
|
|
|
|
|
def test_patch_panel_json_rejects_oversize_file(tmp_path: Path) -> None:
|
|
p = tmp_path / "panel.json"
|
|
inner = ",".join('"k{}":"1"'.format(i) for i in range(5000))
|
|
p.write_text('{"bdf_to_patch": {' + inner + "}}", encoding="utf-8")
|
|
assert load_patch_panel_bdf_map(p) == {}
|
|
|
|
|
|
def test_parse_fabric_ini_overlay_patch_panel_ports(tmp_path: Path) -> None:
|
|
ini = tmp_path / "lab.ini"
|
|
ini.write_text("[fabric]\npatch_panel_ports = 24\n", encoding="utf-8")
|
|
assert parse_fabric_ini_overlay(ini).patch_panel_ports == 24
|
|
|
|
ini.write_text("[fabric]\npatch_panel_port_count = 12\n", encoding="utf-8")
|
|
assert parse_fabric_ini_overlay(ini).patch_panel_ports == 12
|
|
|
|
|
|
def test_with_concentrator_override(tmp_path: Path) -> None:
|
|
fp = acroname_modules_fingerprint(_sample_modules())
|
|
FabricDefinition(
|
|
fabric_id="ov",
|
|
discovery_fingerprint=fp,
|
|
rrhs=(FabricRRHBinding(radio_id="a", acroname_port=0, acroname_module_serial=111),),
|
|
).save(tmp_path / "f.json")
|
|
f = Fabric.from_json_path(tmp_path / "f.json")
|
|
assert f.concentrator is None
|
|
f2 = f.with_concentrator_override(name="rig", ipaddr="10.0.0.5")
|
|
assert f2.concentrator is not None and f2.concentrator.ipaddr == "10.0.0.5"
|