tegwick
f8d1c542d5
Implement RESOURCE-WP-0005: entity register, V0.1 terms parameters, entity association on inventory and planning records, transfer-price and credit-state arithmetic, monthly settlement, and entity views on the portfolio report. Live close emits nothing until delivered cost is known. Handoffs are FIN-WP-0006 and RAILIANCE-WP-0017.
101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Semantic validation for managed-infrastructure portfolio records."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from entities import association_ok
|
|
|
|
RESOURCE_ID = re.compile(r"^resource:[a-z0-9][a-z0-9:_-]+$")
|
|
STATUSES = {
|
|
"proposed", "ordered", "commissioning", "active", "suspended",
|
|
"retiring", "retired", "rejected",
|
|
}
|
|
TRANSITIONS = {
|
|
"proposed": {"ordered", "rejected"},
|
|
"ordered": {"commissioning", "rejected"},
|
|
"commissioning": {"active", "rejected"},
|
|
"active": {"suspended", "retiring"},
|
|
"suspended": {"active", "retiring"},
|
|
"retiring": {"retired", "active"},
|
|
"retired": set(),
|
|
"rejected": {"proposed"},
|
|
}
|
|
|
|
|
|
def _day(value: str | None) -> date | None:
|
|
return date.fromisoformat(value) if value else None
|
|
|
|
|
|
def validate_transition(current: str, target: str) -> None:
|
|
if current not in STATUSES or target not in STATUSES:
|
|
raise ValueError("unknown lifecycle status")
|
|
if target not in TRANSITIONS[current]:
|
|
raise ValueError(f"invalid lifecycle transition {current} -> {target}")
|
|
|
|
|
|
def validate_record(record: dict) -> None:
|
|
if record.get("schema_version") != "0.2":
|
|
raise ValueError("portfolio records must use schema_version 0.2")
|
|
if not RESOURCE_ID.fullmatch(record.get("id", "")):
|
|
raise ValueError("invalid resource id")
|
|
if record.get("status") not in STATUSES:
|
|
raise ValueError("unknown lifecycle status")
|
|
if record.get("record_scope") not in {"inventory", "example"}:
|
|
raise ValueError("record_scope must be inventory or example")
|
|
|
|
association_ok(record)
|
|
|
|
allocation = record["ownership"]["allocation"]
|
|
if allocation["mode"] == "unattributed":
|
|
if allocation["cost_attribution_key"] is not None:
|
|
raise ValueError("unattributed resources cannot have an attribution key")
|
|
elif not allocation["cost_attribution_key"]:
|
|
raise ValueError("dedicated/shared resources require an attribution key")
|
|
if allocation["mode"] == "shared" and (
|
|
not allocation["driver"] or not allocation["method_version"]
|
|
):
|
|
raise ValueError("shared resources require a driver and method version")
|
|
|
|
dimensions = [(item["metric"], item["kind"]) for item in record["capacity"]]
|
|
if len(dimensions) != len(set(dimensions)):
|
|
raise ValueError("capacity metric/kind pairs must be unique")
|
|
if any(rel["resource_id"] == record["id"] for rel in record["relationships"]):
|
|
raise ValueError("a resource cannot relate to itself")
|
|
|
|
lifecycle = record["lifecycle"]
|
|
proposed = _day(lifecycle["proposed_on"])
|
|
ordered = _day(lifecycle["ordered_on"])
|
|
commissioned = _day(lifecycle["commissioned_on"])
|
|
retired = _day(lifecycle["retired_on"])
|
|
dated = [value for value in (proposed, ordered, commissioned, retired) if value]
|
|
if dated != sorted(dated):
|
|
raise ValueError("lifecycle dates are out of order")
|
|
if record["status"] == "retired" and retired is None:
|
|
raise ValueError("retired resources require retired_on")
|
|
# Discovery must preserve unknown commercial and commissioning dates as
|
|
# null instead of manufacturing precision. Date ordering is enforced when
|
|
# the authoritative repositories or provider evidence supply the values.
|
|
if record["record_scope"] == "example" and not any(
|
|
item["kind"] == "example" for item in record["evidence"]
|
|
):
|
|
raise ValueError("examples require explicit example evidence")
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2:
|
|
print(f"usage: {sys.argv[0]} RECORD.json ...", file=sys.stderr)
|
|
return 2
|
|
for name in sys.argv[1:]:
|
|
validate_record(json.loads(Path(name).read_text()))
|
|
print(f"portfolio records valid: {len(sys.argv) - 1}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|