feat(rack): validate network links against peer files and ports

This commit is contained in:
sjat 2026-06-24 15:02:34 +02:00
parent 734a6522c1
commit ed5bda83e0
2 changed files with 134 additions and 0 deletions

View file

@ -120,6 +120,61 @@ def check_overlaps(items: list[dict]) -> None:
occupied[key] = name 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]: def _pdu_index(items: list[dict]) -> dict[str, dict]:
"""Map hostname -> frontmatter for every kind:pdu item.""" """Map hostname -> frontmatter for every kind:pdu item."""
return { return {
@ -383,6 +438,7 @@ def render_page(rack: str, items: list[dict]) -> str:
def generate(hardware_dir: Path, output_dir: Path) -> int: def generate(hardware_dir: Path, output_dir: Path) -> int:
items = load_rack_items(hardware_dir) items = load_rack_items(hardware_dir)
hw_index = load_hardware_index(hardware_dir)
errors: list[str] = [] errors: list[str] = []
for fm in items: for fm in items:
@ -400,6 +456,7 @@ def generate(hardware_dir: Path, output_dir: Path) -> int:
try: try:
check_overlaps(ritems) check_overlaps(ritems)
validate_power(ritems) validate_power(ritems)
validate_links(ritems, hw_index)
except SchemaError as e: except SchemaError as e:
errors.append(f"{rack}: {e}") errors.append(f"{rack}: {e}")

View file

@ -342,3 +342,80 @@ def test_generate_includes_power_section(tmp_path):
assert "## Power" in page assert "## Power" in page
assert "```mermaid" in page assert "```mermaid" in page
assert "outlet 1" 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()