140 lines
4.0 KiB
Python
140 lines
4.0 KiB
Python
# Copyright (c) 2026 Umber
|
|
#
|
|
# Licensed under the Apache License, Version 2.0; see LICENSE.
|
|
|
|
from __future__ import annotations
|
|
|
|
import textwrap
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fiwicontrol.lab.inventory_config import HostInventorySpec, load_inventory_ini
|
|
from fiwicontrol.lab.inventory_verify import compare_host_inventory, dump_inventory_discovery_ini_sync
|
|
|
|
|
|
def test_load_default_lab_ini(tmp_path: Path) -> None:
|
|
src = Path(__file__).resolve().parents[1] / "configs" / "default.ini"
|
|
doc = load_inventory_ini(src)
|
|
assert doc.site_name == "ClubHouse"
|
|
assert "raspberry_pi5" in doc.nodes
|
|
names = {h.name for h in doc.hosts}
|
|
assert names == {"localhost", "pi5"}
|
|
local = next(h for h in doc.hosts if h.name == "localhost")
|
|
assert local.mode == "local"
|
|
assert local.expected_acronames == (("USBHub2x4", 4),)
|
|
assert local.monsoon_count == 0
|
|
pi = next(h for h in doc.hosts if h.name == "pi5")
|
|
assert pi.mode == "remote"
|
|
assert pi.ipaddr == "192.168.1.39"
|
|
assert pi.expected_acronames == (("USBHub3p", 8), ("USBHub3p", 8), ("USBHub2x4", 4))
|
|
assert pi.monsoon_count == 1
|
|
|
|
|
|
def test_compare_host_inventory_match() -> None:
|
|
spec = HostInventorySpec(
|
|
name="t",
|
|
mode="local",
|
|
node_section_id=None,
|
|
ipaddr=None,
|
|
sshtype="ssh",
|
|
expected_acronames=(("USBHub2x4", 4), ("USBHub2x4", 4)),
|
|
monsoon_count=0,
|
|
)
|
|
payload = {
|
|
"acroname": [
|
|
{
|
|
"transport": "USB",
|
|
"serial_number": 1,
|
|
"module_address": 6,
|
|
"model_id": 17,
|
|
"model_name": "USBHub2x4",
|
|
"model_description": "",
|
|
"stem_class": "USBHub2x4",
|
|
"downstream_usb_ports": 4,
|
|
"hub_port_entities": 6,
|
|
},
|
|
{
|
|
"transport": "USB",
|
|
"serial_number": 2,
|
|
"module_address": 6,
|
|
"model_id": 17,
|
|
"model_name": "USBHub2x4",
|
|
"model_description": "",
|
|
"stem_class": "USBHub2x4",
|
|
"downstream_usb_ports": 4,
|
|
"hub_port_entities": 6,
|
|
},
|
|
],
|
|
"monsoon": [],
|
|
}
|
|
assert compare_host_inventory(spec, payload) == []
|
|
|
|
|
|
def test_compare_host_inventory_mismatch() -> None:
|
|
spec = HostInventorySpec(
|
|
name="t",
|
|
mode="local",
|
|
node_section_id=None,
|
|
ipaddr=None,
|
|
sshtype="ssh",
|
|
expected_acronames=(("USBHub3p", 8),),
|
|
monsoon_count=1,
|
|
)
|
|
payload = {
|
|
"acroname": [
|
|
{
|
|
"transport": "USB",
|
|
"serial_number": 1,
|
|
"module_address": 6,
|
|
"model_id": 17,
|
|
"model_name": "USBHub2x4",
|
|
"model_description": "",
|
|
"stem_class": "USBHub2x4",
|
|
"downstream_usb_ports": 4,
|
|
"hub_port_entities": 6,
|
|
},
|
|
],
|
|
"monsoon": [],
|
|
}
|
|
errs = compare_host_inventory(spec, payload)
|
|
assert len(errs) == 2
|
|
|
|
|
|
def test_parse_acroname_invalid(tmp_path: Path) -> None:
|
|
p = tmp_path / "x.ini"
|
|
p.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
[inventory.host.x]
|
|
mode = local
|
|
acroname = BadEntry
|
|
monsoon_count = 0
|
|
"""
|
|
).strip(),
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(ValueError, match="Stem:ports"):
|
|
load_inventory_ini(p)
|
|
|
|
|
|
def test_dump_inventory_discovery_local_only(tmp_path: Path) -> None:
|
|
ini = tmp_path / "solo.ini"
|
|
ini.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
[inventory.host.solo]
|
|
mode = local
|
|
acroname =
|
|
monsoon_count = 0
|
|
"""
|
|
).strip(),
|
|
encoding="utf-8",
|
|
)
|
|
buf = StringIO()
|
|
dump_inventory_discovery_ini_sync(ini, file=buf)
|
|
text = buf.getvalue()
|
|
assert "=== solo (mode=local;" in text
|
|
assert '"acroname"' in text and '"monsoon"' in text
|