FiWiControl/tests/test_inventory_config.py

# 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, resolve_fabric_concentrator
from fiwicontrol.lab.inventory_verify import compare_host_inventory, dump_inventory_discovery_ini_sync


def test_resolve_fabric_concentrator_by_machine_name(tmp_path: Path) -> None:
    p = tmp_path / "lab.ini"
    p.write_text(
        textwrap.dedent(
            """
            [machine.a]
            machine.name = bench-workstation
            ipaddr = 10.0.0.1
            sshtype = ssh
            usb = local
            acroname =
            hvpm = HVPM:0

            [machine.b]
            machine.name = bench-pi
            ipaddr = 10.0.0.2
            sshtype = ssh
            usb = remote
            acroname =
            hvpm = HVPM:0
            """
        ).strip(),
        encoding="utf-8",
    )
    doc = load_inventory_ini(p)
    sid, n = resolve_fabric_concentrator(doc.nodes, "bench-workstation")
    assert sid == "a" and n.ipaddr == "10.0.0.1"


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 == "UAX-24"
    assert "raspberry_pi5" in doc.nodes
    names = {h.name for h in doc.hosts}
    assert names == {"fedora42", "raspberry_pi5"}
    local = next(h for h in doc.hosts if h.name == "fedora42")
    assert local.mode == "local"
    assert local.expected_acronames == (("USBHub2x4", 4),)
    assert local.hvpm_count == 0
    pi = next(h for h in doc.hosts if h.name == "raspberry_pi5")
    assert pi.mode == "relay"
    assert pi.relay_node_id == "raspberry_pi5"
    assert pi.ipaddr == "192.168.1.39"
    assert pi.expected_acronames == (("USBHub3p", 8), ("USBHub3p", 8), ("USBHub2x4", 4))
    assert pi.hvpm_count == 1


def test_compare_host_inventory_match() -> None:
    spec = HostInventorySpec(
        name="t",
        mode="local",
        relay_node_id=None,
        ipaddr=None,
        sshtype="ssh",
        expected_acronames=(("USBHub2x4", 4), ("USBHub2x4", 4)),
        hvpm_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",
        relay_node_id=None,
        ipaddr=None,
        sshtype="ssh",
        expected_acronames=(("USBHub3p", 8),),
        hvpm_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_hvpm_invalid(tmp_path: Path) -> None:
    p = tmp_path / "x.ini"
    p.write_text(
        textwrap.dedent(
            """
            [host.x]
            mode = local
            acroname =
            hvpm = Monsoon:1
            """
        ).strip(),
        encoding="utf-8",
    )
    with pytest.raises(ValueError, match="stem must be HVPM"):
        load_inventory_ini(p)


def test_parse_acroname_invalid(tmp_path: Path) -> None:
    p = tmp_path / "x.ini"
    p.write_text(
        textwrap.dedent(
            """
            [host.x]
            mode = local
            acroname = BadEntry
            hvpm = HVPM: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(
            """
            [host.solo]
            mode = local
            acroname =
            hvpm = HVPM: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