#!/usr/bin/env python3 """Validate capability requirements and provisions against the live ITC-CAP catalog. resource-control does not copy the canon. It reads `capabilities.yaml` from info-tech-canon and checks that every capability id, profile, quality dimension, maturity level, resource class, and predicate this repository uses is one the canon actually declares. A drift in either repository fails here rather than silently producing a record the canon would reject. """ from __future__ import annotations import json import sys from pathlib import Path import basis as basis_module CANON_PATHS = ( Path("/home/worsch/info-tech-canon/infospace/models/capability/capabilities.yaml"), Path("../info-tech-canon/infospace/models/capability/capabilities.yaml"), ) PREDICATES = frozenset({"not_in", "in", "equals", "lte", "gte"}) def _parse_yaml(text: str) -> dict: try: import yaml except ModuleNotFoundError: # pragma: no cover - yaml is present in this env raise RuntimeError("PyYAML is required to read the canon catalog") return yaml.safe_load(text) def load_canon(paths=CANON_PATHS) -> dict: for path in paths: if path.exists(): catalog = _parse_yaml(path.read_text()) capabilities = { cap["id"]: cap for domain in catalog["domains"] for cap in domain.get("capabilities", []) } return { "version": catalog["canon"]["version"], "canon_version": catalog["canon"]["canon_version"], "capabilities": capabilities, "resource_classes": {rc["id"]: rc for rc in catalog["resource_classes"]}, "maturity_levels": {m["id"] if isinstance(m, dict) else m for m in catalog["maturity_levels"]}, "source": str(path), } raise FileNotFoundError("ITC-CAP catalog not found; is info-tech-canon checked out?") def validate_requirement(requirement: dict, canon: dict) -> None: cap_id = requirement["capability"] capability = canon["capabilities"].get(cap_id) if capability is None: raise ValueError(f"unknown capability {cap_id}") profile = requirement.get("profile") if profile and profile not in (capability.get("profiles") or []): raise ValueError(f"{cap_id} does not declare profile {profile!r}") maturity = requirement.get("minimum_maturity") if maturity and maturity not in canon["maturity_levels"]: raise ValueError(f"unknown maturity level {maturity!r}") dimensions = set(capability.get("quality_dimensions") or []) # CAP-R9: a target key or constraint dimension must be declared on the # capability. This is what stops a consumer inventing quality vocabulary. for key in (requirement.get("targets") or {}): if key not in dimensions: raise ValueError(f"{cap_id} does not declare quality dimension {key!r}") for constraint in requirement.get("constraints") or []: if constraint["dimension"] not in dimensions: raise ValueError(f"{cap_id} does not declare quality dimension {constraint['dimension']!r}") if constraint["predicate"] not in PREDICATES: raise ValueError(f"predicate {constraint['predicate']!r} is not in the closed set") if not constraint.get("of"): raise ValueError("a constraint must name what it applies to") def validate_provision(provision: dict, canon: dict) -> None: cap_id = provision["capability"] capability = canon["capabilities"].get(cap_id) if capability is None: raise ValueError(f"unknown capability {cap_id}") if provision["maturity"] not in canon["maturity_levels"]: raise ValueError(f"unknown maturity level {provision['maturity']!r}") profile = provision.get("profile") if profile and profile not in (capability.get("profiles") or []): raise ValueError(f"{cap_id} does not declare profile {profile!r}") hooks = set(capability.get("evidence_hooks") or []) for item in provision.get("evidence") or []: if item["hook"] not in hooks: raise ValueError(f"{cap_id} does not declare evidence hook {item['hook']!r}") basis_module.rank(item["basis"]) seen = set() for row in provision.get("consumes") or []: klass = row["class"] if klass not in canon["resource_classes"]: raise ValueError(f"unknown resource class {klass!r}") if klass in seen: raise ValueError(f"duplicate consumption row for class {klass}") seen.add(klass) declared = canon["resource_classes"][klass] unit = row["quantity"]["unit"] if unit != declared["native_unit"]: raise ValueError( f"class {klass} native unit is {declared['native_unit']!r}, got {unit!r}" ) # Unknown is recorded as unknown, never as zero. basis_module.validate_value({ "basis": row["basis"], "value": row["quantity"]["value"], "gap": row.get("gap"), "derived_from": row.get("derived_from"), }) if row.get("supply") and row["supply"] not in {"internal", "external"}: raise ValueError(f"supply must be internal or external, got {row['supply']!r}") def evidence_coverage(provision: dict, canon: dict) -> dict: """Which of the capability's declared evidence hooks this provision satisfies.""" capability = canon["capabilities"][provision["capability"]] declared = list(capability.get("evidence_hooks") or []) supplied = {item["hook"] for item in provision.get("evidence") or []} return { "declared": declared, "supplied": sorted(supplied), "missing": sorted(set(declared) - supplied), "complete": not (set(declared) - supplied), } def consumption_profile(provision: dict) -> dict: values = [ { "name": f"{row['class']}:{row.get('name', '')}", "basis": row["basis"], "value": row["quantity"]["value"], "gap": row.get("gap"), "derived_from": row.get("derived_from"), "proxy_for": row.get("proxy_for"), } for row in provision.get("consumes") or [] ] return basis_module.profile(values) def review(record: dict, canon: dict) -> dict: for requirement in record.get("requires") or []: validate_requirement(requirement, canon) for provision in record.get("provisions") or []: validate_provision(provision, canon) # Does each requirement actually have a provision that meets it? provisions = {p["capability"]: p for p in record.get("provisions") or []} met = [] for requirement in record.get("requires") or []: provision = provisions.get(requirement["capability"]) wanted = requirement.get("minimum_maturity") if provision is None: met.append({"capability": requirement["capability"], "status": "unprovided"}) continue levels = sorted(canon["maturity_levels"]) satisfied = levels.index(provision["maturity"]) >= levels.index(wanted) if wanted else True met.append({ "capability": requirement["capability"], "required": wanted, "provided": provision["maturity"], "status": "met" if satisfied else "below_requirement", }) return { "record_id": record["record_id"], "canon": {k: canon[k] for k in ("version", "canon_version", "source")}, "requirements": met, "provisions": [ { "capability": provision["capability"], "maturity": provision["maturity"], "evidence": evidence_coverage(provision, canon), "consumption": consumption_profile(provision), } for provision in record.get("provisions") or [] ], "alternatives_grade": ( basis_module.decision_grade(record["modelled_alternatives"]["values"]) if record.get("modelled_alternatives") else None ), } def main() -> int: paths = sys.argv[1:] or sorted(str(p) for p in Path("data/capability").glob("*.json")) canon = load_canon() reports = [review(json.loads(Path(p).read_text()), canon) for p in paths] print(json.dumps(reports, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())