From ed5bda83e0bb389f36a40aa70b3bb3650c5047cb Mon Sep 17 00:00:00 2001 From: sjat Date: Wed, 24 Jun 2026 15:02:34 +0200 Subject: [PATCH] feat(rack): validate network links against peer files and ports --- scripts/gen_rack.py | 57 +++++++++++++++++++++++++++++++ tests/test_gen_rack.py | 77 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+) diff --git a/scripts/gen_rack.py b/scripts/gen_rack.py index ab48a92..ac716a8 100644 --- a/scripts/gen_rack.py +++ b/scripts/gen_rack.py @@ -120,6 +120,61 @@ def check_overlaps(items: list[dict]) -> None: occupied[key] = name +def load_hardware_index(hardware_dir: Path) -> dict[str, dict]: + """Map hostname -> frontmatter for every hardware file (global peer lookup).""" + index: dict[str, dict] = {} + for path in sorted(hardware_dir.glob("*.md")): + if path.name == "index.md": + continue + fm = parse_frontmatter(path) + if fm is None: + continue + name = fm.get("hostname") + if isinstance(name, str) and name: + index[name] = fm + return index + + +def validate_links(items: list[dict], hw_index: dict[str, dict]) -> None: + """Validate `links` cable declarations (rule 4). + + Every links[].peer must resolve to a real hardware file (global lookup via + hw_index); peer_port must fall within the peer's declared `ports` when it + declares an integer count. + """ + for fm in items: + links = fm.get("links") + if links is None: + continue + name = fm.get("hostname", "?") + if not isinstance(links, list): + raise SchemaError(f"{name}: links must be a list") + for link in links: + if not isinstance(link, dict): + raise SchemaError(f"{name}: links entry must be a mapping") + local = link.get("local") + peer = link.get("peer") + peer_port = link.get("peer_port") + if not isinstance(local, str) or not local: + raise SchemaError(f"{name}: links entry needs a non-empty 'local'") + if not isinstance(peer, str) or not peer: + raise SchemaError(f"{name}: links entry needs a non-empty 'peer'") + if not isinstance(peer_port, int): + raise SchemaError( + f"{name}: links entry for {peer} needs an integer 'peer_port'" + ) + target = hw_index.get(peer) + if target is None: + raise SchemaError( + f"{name}: links peer={peer!r} is not a known hardware file" + ) + ports = target.get("ports") + if isinstance(ports, int) and (peer_port < 1 or peer_port > ports): + raise SchemaError( + f"{name}: peer_port {peer_port} out of range 1..{ports} on {peer}" + ) + + def _pdu_index(items: list[dict]) -> dict[str, dict]: """Map hostname -> frontmatter for every kind:pdu item.""" return { @@ -383,6 +438,7 @@ def render_page(rack: str, items: list[dict]) -> str: def generate(hardware_dir: Path, output_dir: Path) -> int: items = load_rack_items(hardware_dir) + hw_index = load_hardware_index(hardware_dir) errors: list[str] = [] for fm in items: @@ -400,6 +456,7 @@ def generate(hardware_dir: Path, output_dir: Path) -> int: try: check_overlaps(ritems) validate_power(ritems) + validate_links(ritems, hw_index) except SchemaError as e: errors.append(f"{rack}: {e}") diff --git a/tests/test_gen_rack.py b/tests/test_gen_rack.py index 6ecd93c..f1a5e7e 100644 --- a/tests/test_gen_rack.py +++ b/tests/test_gen_rack.py @@ -342,3 +342,80 @@ def test_generate_includes_power_section(tmp_path): assert "## Power" in page assert "```mermaid" in page assert "outlet 1" in page + + +def test_load_hardware_index_maps_all_hostnames(tmp_path): + hw = tmp_path / "hardware" + hw.mkdir() + _write_item( + hw, "sw01", + "---\nhostname: sw01\nkind: switch\nstatus: in-use\nports: 24\n---\n", + ) + _write_item( + hw, "mf00", + "---\nhostname: mf00\nkind: server\nstatus: in-use\n" + "rack: rack01\nrack_u: 1\nu_height: 1\nrack_face: front\n---\n", + ) + idx = gen_rack.load_hardware_index(hw) + assert set(idx) == {"sw01", "mf00"} + assert idx["sw01"]["ports"] == 24 + + +def test_validate_links_accepts_valid_link(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=[{"local": "eth0", "peer": "sw01", + "peer_port": 1, "speed_gbps": 1}])] + hw_index = {"sw01": item(hostname="sw01", kind="switch", ports=24)} + gen_rack.validate_links(items, hw_index) + + +def test_validate_links_rejects_unknown_peer(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=[{"local": "eth0", "peer": "ghost", "peer_port": 1}])] + with pytest.raises(gen_rack.SchemaError): + gen_rack.validate_links(items, {}) + + +def test_validate_links_rejects_peer_port_over_count(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=[{"local": "eth0", "peer": "sw01", "peer_port": 25}])] + hw_index = {"sw01": item(hostname="sw01", kind="switch", ports=24)} + with pytest.raises(gen_rack.SchemaError): + gen_rack.validate_links(items, hw_index) + + +def test_validate_links_accepts_peer_without_ports(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=[{"local": "eth0", "peer": "rtr01", "peer_port": 99}])] + hw_index = {"rtr01": item(hostname="rtr01", kind="server")} + gen_rack.validate_links(items, hw_index) # no ports -> range check skipped + + +def test_validate_links_rejects_missing_local(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=[{"peer": "sw01", "peer_port": 1}])] + hw_index = {"sw01": item(hostname="sw01", kind="switch", ports=24)} + with pytest.raises(gen_rack.SchemaError): + gen_rack.validate_links(items, hw_index) + + +def test_validate_links_rejects_malformed_entry(): + items = [item(hostname="mf01", rack_u=2, u_height=1, rack_face="front", + links=["sw01"])] + with pytest.raises(gen_rack.SchemaError): + gen_rack.validate_links(items, {}) + + +def test_generate_returns_1_on_bad_link_peer(tmp_path): + hw = tmp_path / "hardware" + out = tmp_path / "out" + hw.mkdir() + _write_item( + hw, "mf00", + "---\nhostname: mf00\nkind: server\nstatus: in-use\n" + "rack: rack01\nrack_u: 1\nu_height: 1\nrack_face: front\n" + "links:\n - { local: eth0, peer: ghost, peer_port: 1 }\n---\n", + ) + rc = gen_rack.generate(hw, out) + assert rc == 1 + assert not (out / "rack01.md").exists()