Version the common and derived rail contract

tools/validate_contracts.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""Validate Railiance rail and workload contracts without reading credentials."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+import yaml
+from jsonschema import Draft202012Validator, RefResolver
+
+
+READINESS_STATES = {
+    "declared",
+    "installed",
+    "verified",
+    "production-approved",
+    "deprecated",
+}
+
+
+def load(path: Path) -> Any:
+    with path.open(encoding="utf-8") as handle:
+        if path.suffix in {".yaml", ".yml"}:
+            return yaml.safe_load(handle)
+        return json.load(handle)
+
+
+def validate_schema(instance_path: Path, schema_path: Path) -> list[str]:
+    instance = load(instance_path)
+    schema = load(schema_path)
+    store: dict[str, Any] = {}
+    for candidate in schema_path.parent.glob("*.json"):
+        candidate_schema = load(candidate)
+        store[candidate.resolve().as_uri()] = candidate_schema
+        if schema_id := candidate_schema.get("$id"):
+            store[schema_id] = candidate_schema
+    resolver = RefResolver(
+        base_uri=schema_path.parent.resolve().as_uri() + "/",
+        referrer=schema,
+        store=store,
+    )
+    validator = Draft202012Validator(schema, resolver=resolver)
+    return [
+        f"{instance_path}: {'/'.join(str(part) for part in error.path) or '<root>'}: {error.message}"
+        for error in sorted(validator.iter_errors(instance), key=lambda item: list(item.path))
+    ]
+
+
+def validate_rail(path: Path) -> list[str]:
+    data = load(path)
+    errors: list[str] = []
+    required = {
+        "rail_id",
+        "ownership_repo",
+        "contract_version",
+        "composition_kind",
+        "execution_architecture",
+        "required_substrate_capabilities",
+        "readiness_state",
+    }
+    for field in sorted(required - set(data or {})):
+        errors.append(f"{path}: missing required field {field}")
+    if not isinstance(data, dict):
+        return [f"{path}: declaration must be an object"]
+    if data.get("readiness_state") not in READINESS_STATES:
+        errors.append(f"{path}: invalid readiness_state {data.get('readiness_state')!r}")
+    kind = data.get("composition_kind")
+    if kind not in {"base", "derived"}:
+        errors.append(f"{path}: composition_kind must be base or derived")
+    if kind == "derived":
+        for field in (
+            "base_rail",
+            "base_rail_contract",
+            "inherited_semantics",
+            "overridden_semantics",
+            "compatibility_constraints",
+        ):
+            if field not in data:
+                errors.append(f"{path}: derived rail missing {field}")
+        if data.get("base_rail") == data.get("rail_id"):
+            errors.append(f"{path}: derived rail cannot inherit from itself")
+    return errors
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--rail", action="append", default=[], type=Path)
+    parser.add_argument("--workload", action="append", default=[], type=Path)
+    parser.add_argument(
+        "--schema-dir",
+        type=Path,
+        default=Path(__file__).resolve().parents[1] / "schemas",
+    )
+    args = parser.parse_args()
+
+    errors: list[str] = []
+    for path in args.rail:
+        errors.extend(validate_rail(path))
+    for path in args.workload:
+        errors.extend(
+            validate_schema(path, args.schema_dir / "common-workload.schema.json")
+        )
+    if errors:
+        print("\n".join(errors), file=sys.stderr)
+        return 1
+    print(
+        f"validated {len(args.rail)} rail declaration(s) and "
+        f"{len(args.workload)} workload declaration(s)"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())