AISA v2.0 / Technical documentation / validate_product.py
validate_product.py
Python · 225 lines · 8,821 bytes · wiki path 10 Architecture/20 Configuration/contracts/validate_product.py · download the raw file · cited from Stage S1 — Normalization
Same folder: product.schema.json
#!/usr/bin/env python3
"""
Validate a whitelabel product specification.
Two layers, because a JSON Schema can only check that a number is *present*:
1. structural - product.schema.json (types, enums, required fields)
2. semantic - the rules below, which RECOMPUTE from the document body
rather than trusting what the document claims about itself
Exit 0 = the specification may go to H1. Exit 1 = it may not.
python validate_product.py product.json [--schema product.schema.json]
Reference: aisa-next/10 Architecture/20 Configuration/Stage Normalization.md (the WS-n sections)
"""
from __future__ import annotations
import json, sys, argparse
from pathlib import Path
BLOCKERS: list[str] = []
WARNINGS: list[str] = []
def blocker(msg: str) -> None:
BLOCKERS.append(msg)
def warn(msg: str) -> None:
WARNINGS.append(msg)
# ---------------------------------------------------------------- structural
def structural(doc: dict, schema_path: Path) -> None:
try:
import jsonschema
except ImportError:
warn("jsonschema not installed - structural layer skipped (pip install jsonschema)")
return
schema = json.loads(schema_path.read_text(encoding="utf-8"))
validator = jsonschema.Draft202012Validator(schema)
for err in sorted(validator.iter_errors(doc), key=lambda e: list(e.path)):
where = "/".join(str(p) for p in err.path) or "(root)"
blocker(f"schema: {where}: {err.message}")
# ------------------------------------------------------- semantic: counting
def recount(doc: dict) -> dict[str, int]:
"""Recompute every reconcilable count FROM THE BODY. Never read `in_model`."""
covers = doc.get("covers", [])
counts = {
# distinct cover CODES, never row counts - one document cover
# legitimately becomes N rows, one per object
"covers": len({c["key"] for c in covers if "key" in c}),
"ld": len(doc.get("ld", [])),
"factors": len(doc.get("factors", [])),
}
tariff = doc.get("tariff", {})
lm = tariff.get("layer_model")
if lm:
product = 1
for axis in lm.get("axes", []):
product *= int(axis.get("cardinality", 1))
counts["tariff_layers"] = product
cells = 0
for cr in tariff.get("cover_rates", []):
grid = cr.get("grid") or {}
for row in grid.get("rates", []) or []:
cells += len(row)
if cells:
counts["rate_cells"] = cells
return counts
def check_reconciliation(doc: dict) -> None:
claimed = doc.get("reconciliation", {})
actual = recount(doc)
for key, entry in claimed.items():
src = entry.get("claimed_in_source")
stated = entry.get("in_model")
real = actual.get(key)
if real is None:
warn(f"reconciliation.{key}: nothing in the body to recount against")
continue
if stated != real:
blocker(
f"reconciliation.{key}: the document claims in_model={stated} "
f"but the body actually contains {real}"
)
if src != real:
blocker(
f"reconciliation.{key}: source claims {src}, body contains {real} "
f"- a mismatch is a stop, not a warning"
)
for key in ("covers", "factors"):
if key not in claimed:
blocker(f"reconciliation.{key} is missing and is mandatory")
# -------------------------------------------------------- semantic: tariff
def check_layer_model(doc: dict) -> None:
tariff = doc.get("tariff", {})
rates = tariff.get("cover_rates", [])
lm = tariff.get("layer_model")
grids = [c for c in rates if (c.get("grid") or {}).get("rates")]
# "the source repeats its rate tables under more than one qualifier"
repeats = len(grids) > 1 or len(rates) > len({c.get("cover") for c in rates})
if repeats and not lm:
blocker(
"tariff.layer_model is absent while the tariff repeats its rate tables. "
"Name the axes and their cardinality, or the layer count cannot be checked at all"
)
if lm:
for axis in lm.get("axes", []):
if not axis.get("carried_by"):
blocker(f"tariff.layer_model axis '{axis.get('name')}' does not say what carries it in the document")
# ----------------------------------------------------- semantic: provenance
def check_sources(doc: dict) -> None:
"""Every element that declares a source must actually carry one."""
def walk(node, path):
if isinstance(node, dict):
if "source" in node and not str(node.get("source") or "").strip():
blocker(f"{path}: empty source - an element without provenance is a guess")
for k, v in node.items():
walk(v, f"{path}.{k}" if path else k)
elif isinstance(node, list):
for i, v in enumerate(node):
walk(v, f"{path}[{i}]")
walk(doc, "")
# ------------------------------------------- semantic: cross-reference sanity
def check_references(doc: dict) -> None:
cover_keys = {c["key"] for c in doc.get("covers", []) if "key" in c}
factor_codes = {f["code"] for f in doc.get("factors", []) if "code" in f}
object_codes = {o["code"] for o in doc.get("objects", []) if "code" in o}
for cr in doc.get("tariff", {}).get("cover_rates", []):
if cr.get("cover") not in cover_keys:
blocker(f"tariff.cover_rates references cover '{cr.get('cover')}' which is not in covers[]")
base = cr.get("base")
if base and base not in factor_codes:
blocker(f"tariff.cover_rates[{cr.get('cover')}].base references factor '{base}' which is not in factors[]")
grid = cr.get("grid") or {}
for role in ("row_factor", "column_factor"):
code = grid.get(role)
if code and code not in factor_codes:
blocker(f"tariff grid {role} '{code}' is not in factors[]")
for c in doc.get("covers", []):
obj = c.get("object")
if obj and object_codes and obj not in object_codes:
blocker(f"cover '{c.get('key')}' references object '{obj}' which is not in objects[]")
for off in doc.get("offers", []):
for oc in off.get("covers", []):
if oc.get("cover") not in cover_keys:
blocker(f"offer '{off.get('code')}' references cover '{oc.get('cover')}' which is not in covers[]")
for ld in doc.get("ld", []):
scope = ld.get("scope", "")
if scope.startswith("cover:"):
key = scope.split(":", 1)[1].split("@")[0]
if key not in cover_keys:
blocker(f"ld '{ld.get('code')}' is scoped to cover '{key}' which is not in covers[]")
# ------------------------------------------------- semantic: stage coherence
def check_factor_coherence(doc: dict) -> None:
for f in doc.get("factors", []):
code = f.get("code")
if f.get("rating_axis") and f.get("answered_by") == "backend_operation":
warn(f"factor '{code}': a rating axis supplied by a backend operation needs a PR_OPERATIONS row, not a factor row")
if f.get("datatype") == "LIST" and not f.get("values"):
blocker(f"factor '{code}': datatype LIST with no values[]")
defaults = [v for v in f.get("values", []) if v.get("default")]
if len(defaults) > 1:
blocker(f"factor '{code}': {len(defaults)} default values - exactly one is allowed")
# ------------------------------------------------------- semantic: H1 gating
def check_h1(doc: dict) -> None:
for u in doc.get("unresolved", []):
if u.get("blocking"):
blocker(f"unresolved (blocking): {u.get('what')} - {u.get('why')}")
for c in doc.get("contradictions", []):
if not c.get("resolved_by_user"):
blocker(f"contradiction on '{c.get('subject')}' is unanswered - contradictions are never resolved by rule")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("product")
ap.add_argument("--schema", default=str(Path(__file__).with_name("product.schema.json")))
args = ap.parse_args()
doc = json.loads(Path(args.product).read_text(encoding="utf-8"))
structural(doc, Path(args.schema))
check_sources(doc)
check_reconciliation(doc)
check_layer_model(doc)
check_references(doc)
check_factor_coherence(doc)
check_h1(doc)
for w in WARNINGS:
print(f" warn {w}")
for b in BLOCKERS:
print(f" BLOCK {b}")
if BLOCKERS:
print(f"\nFAIL - {len(BLOCKERS)} blocker(s). This specification does not pass H1.")
return 1
print(f"\nPASS - {len(WARNINGS)} warning(s). Ready for H1.")
return 0
if __name__ == "__main__":
sys.exit(main())