feat(rack): validate power feeds against PDU outlets

This commit is contained in:
sjat 2026-06-24 14:35:19 +02:00
parent a45d6d0266
commit ed4e7c751a
2 changed files with 136 additions and 0 deletions

View file

@ -120,6 +120,58 @@ def check_overlaps(items: list[dict]) -> None:
occupied[key] = name occupied[key] = name
def _pdu_index(items: list[dict]) -> dict[str, dict]:
"""Map hostname -> frontmatter for every kind:pdu item."""
return {
fm.get("hostname"): fm
for fm in items
if fm.get("kind") == "pdu"
}
def validate_power(items: list[dict]) -> None:
"""Validate PDU outlet declarations and `power` feeds within one rack.
Rule 3: every power[].pdu resolves to a kind:pdu file, and outlet is
within that PDU's `outlets` count.
"""
pdus = _pdu_index(items)
for name, fm in pdus.items():
outlets = fm.get("outlets")
if not isinstance(outlets, int) or outlets < 1:
raise SchemaError(
f"{name}: kind:pdu must declare a positive integer 'outlets'"
)
for fm in items:
feeds = fm.get("power")
if feeds is None:
continue
name = fm.get("hostname", "?")
if not isinstance(feeds, list):
raise SchemaError(f"{name}: power must be a list")
for feed in feeds:
if not isinstance(feed, dict):
raise SchemaError(f"{name}: power entry must be a mapping")
pdu = feed.get("pdu")
outlet = feed.get("outlet")
if not isinstance(pdu, str) or not pdu:
raise SchemaError(f"{name}: power entry needs a non-empty 'pdu'")
if not isinstance(outlet, int):
raise SchemaError(
f"{name}: power entry for {pdu} needs an integer 'outlet'"
)
target = pdus.get(pdu)
if target is None:
raise SchemaError(
f"{name}: power pdu={pdu!r} is not a known kind:pdu file"
)
count = target["outlets"]
if outlet < 1 or outlet > count:
raise SchemaError(
f"{name}: outlet {outlet} out of range 1..{count} on {pdu}"
)
def _esc(s: object) -> str: def _esc(s: object) -> str:
return str(s).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;") return str(s).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
@ -295,6 +347,7 @@ def generate(hardware_dir: Path, output_dir: Path) -> int:
for rack, ritems in racks.items(): for rack, ritems in racks.items():
try: try:
check_overlaps(ritems) check_overlaps(ritems)
validate_power(ritems)
except SchemaError as e: except SchemaError as e:
errors.append(f"{rack}: {e}") errors.append(f"{rack}: {e}")

View file

@ -192,3 +192,86 @@ def test_generate_returns_1_on_overlap(tmp_path):
assert rc == 1 assert rc == 1
assert not (out / "rack01.md").exists() assert not (out / "rack01.md").exists()
def test_validate_power_accepts_valid_feed():
items = [
item(hostname="pdu01", kind="pdu", rack_face="left", outlets=8),
item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=[{"pdu": "pdu01", "outlet": 1}]),
]
gen_rack.validate_power(items)
def test_validate_power_rejects_unknown_pdu():
items = [item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=[{"pdu": "ghost", "outlet": 1}])]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_validate_power_rejects_non_pdu_target():
items = [
item(hostname="sw01", kind="switch", rack_u=1, u_height=1,
rack_face="front"),
item(hostname="mf00", rack_u=2, u_height=1, rack_face="front",
power=[{"pdu": "sw01", "outlet": 1}]),
]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_validate_power_rejects_outlet_over_count():
items = [
item(hostname="pdu01", kind="pdu", rack_face="left", outlets=8),
item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=[{"pdu": "pdu01", "outlet": 9}]),
]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_validate_power_rejects_outlet_zero():
items = [
item(hostname="pdu01", kind="pdu", rack_face="left", outlets=8),
item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=[{"pdu": "pdu01", "outlet": 0}]),
]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_validate_power_rejects_malformed_entry():
items = [
item(hostname="pdu01", kind="pdu", rack_face="left", outlets=8),
item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=["pdu01"]),
]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_validate_power_rejects_pdu_without_outlets():
items = [
item(hostname="pdu01", kind="pdu", rack_face="left"),
item(hostname="mf00", rack_u=1, u_height=1, rack_face="front",
power=[{"pdu": "pdu01", "outlet": 1}]),
]
with pytest.raises(gen_rack.SchemaError):
gen_rack.validate_power(items)
def test_generate_returns_1_on_bad_power_ref(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"
"power:\n - { pdu: ghost, outlet: 1 }\n---\n",
)
rc = gen_rack.generate(hw, out)
assert rc == 1
assert not (out / "rack01.md").exists()