#!/usr/bin/env python3 """Validate rail.yaml, rapp.yaml, and reef.yaml family declarations. RMASTER-WP-0021-T05. Discovers rail-*, rapp-*, and reef-* repos under a root (default: the parent of this repository), loads each declarations/.yaml, and checks, in this order: 1. the file conforms to its family schema 2. sibling-field constraints JSON Schema cannot express 3. declared member repos, named rails, and named reefs resolve on disk 4. reef bound_rapps matches the projection of rapp.bound_reefs 5. a deployable name belongs to at most one rapp 6. if --inventory is given, every live deployable belongs to exactly one rapp This repo does not query a cluster. Live coverage consumes an inventory file produced by an implementation repo (RMASTER-WP-0021-T06). Usage: tools/validate-family-declarations.py tools/validate-family-declarations.py --root /path/to/siblings tools/validate-family-declarations.py --repo /path/to/rapp-openbao tools/validate-family-declarations.py --inventory inventory.json tools/validate-family-declarations.py --self-test Exit 0 if there are no errors (warnings are allowed). Exit 1 on any error. Depends on PyYAML and jsonschema, the same pair used to author the schemas. """ from __future__ import annotations import argparse import json import re import sys from collections import defaultdict from dataclasses import dataclass, field from pathlib import Path from typing import Any, Iterable import yaml from jsonschema import Draft202012Validator REPO_ROOT = Path(__file__).resolve().parent.parent SCHEMA_DIR = REPO_ROOT / "schemas" TESTDATA = Path(__file__).resolve().parent / "testdata" / "family-declarations" FAMILY_PREFIXES = ("rail-", "rapp-", "reef-") DECL_BY_PREFIX = { "rail-": ("rail.yaml", "rail.schema.json"), "rapp-": ("rapp.yaml", "rapp.schema.json"), "reef-": ("reef.yaml", "reef.schema.json"), } FLOATING_PIN = re.compile(r"^(latest|[\^~*]|.*\*|.*x$)", re.IGNORECASE) SLUG = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$") @dataclass class Finding: path: str field: str message: str level: str = "error" def __str__(self) -> str: loc = self.field if self.field else "(file)" return f"{self.path}: {self.level}: {loc}: {self.message}" @dataclass class Loaded: repo_dir: Path rel: str family: str doc: dict[str, Any] @dataclass class Report: findings: list[Finding] = field(default_factory=list) files: int = 0 def add(self, path: str, field: str, message: str, level: str = "error") -> None: self.findings.append(Finding(path, field, message, level)) @property def errors(self) -> list[Finding]: return [f for f in self.findings if f.level == "error"] @property def warnings(self) -> list[Finding]: return [f for f in self.findings if f.level == "warning"] def load_yaml(path: Path) -> Any: with path.open() as fh: return yaml.safe_load(fh) def load_schemas() -> dict[str, Draft202012Validator]: validators: dict[str, Draft202012Validator] = {} for prefix, (_decl, schema_name) in DECL_BY_PREFIX.items(): family = prefix.rstrip("-") schema = load_yaml(SCHEMA_DIR / schema_name) Draft202012Validator.check_schema(schema) validators[family] = Draft202012Validator(schema) return validators def discover_repos(root: Path, extra: Iterable[Path] = ()) -> list[Path]: found: dict[str, Path] = {} if root.is_dir(): for child in sorted(root.iterdir()): if child.is_dir() and child.name.startswith(FAMILY_PREFIXES): found[child.name] = child for repo in extra: repo = repo.resolve() if repo.is_dir(): found[repo.name] = repo return [found[name] for name in sorted(found)] def family_of(repo_dir: Path) -> str | None: for prefix in FAMILY_PREFIXES: if repo_dir.name.startswith(prefix): return prefix.rstrip("-") return None def json_path(error: Any) -> str: parts = [str(p) for p in error.absolute_path] return ".".join(parts) if parts else "(root)" def schema_check(loaded: Loaded, validator: Draft202012Validator, report: Report) -> None: for error in sorted(validator.iter_errors(loaded.doc), key=lambda e: list(e.absolute_path)): report.add(loaded.rel, json_path(error), error.message) def as_list(value: Any) -> list[Any]: return value if isinstance(value, list) else [] def check_rail(loaded: Loaded, report: Report) -> None: doc = loaded.doc if not isinstance(doc, dict): return if doc.get("rail_id") and doc.get("repo") and doc["rail_id"] != doc["repo"]: report.add(loaded.rel, "rail_id", f"{doc['rail_id']!r} must equal repo {doc['repo']!r}") def check_rapp(loaded: Loaded, report: Report) -> None: doc = loaded.doc if not isinstance(doc, dict): return primary = doc.get("primary_rail") supported = as_list(doc.get("supported_rails")) if primary and supported and primary not in supported: report.add( loaded.rel, "primary_rail", f"{primary!r} is not in supported_rails {supported}", ) repo = doc.get("repo") owner = doc.get("ownership_repo") if repo and owner and repo == owner: report.add( loaded.rel, "ownership_repo", "must not be the rapp repo itself", ) for component in as_list((doc.get("composition") or {}).get("upstream_components")): if not isinstance(component, dict): continue version = str(component.get("version") or "") name = component.get("name") or "(unnamed)" if not version or FLOATING_PIN.match(version): report.add( loaded.rel, f"composition.upstream_components.{name}.version", f"{version!r} is not an exact pin", ) def check_reef(loaded: Loaded, report: Report) -> None: doc = loaded.doc if not isinstance(doc, dict): return if doc.get("reef_id") and doc.get("repo") and doc["reef_id"] != doc["repo"]: report.add(loaded.rel, "reef_id", f"{doc['reef_id']!r} must equal repo {doc['repo']!r}") primary = doc.get("primary_rail") hosted = as_list(doc.get("hosted_rails")) if primary and primary not in hosted: report.add( loaded.rel, "primary_rail", f"{primary!r} is not in hosted_rails {hosted}", ) def present_slugs(repos: list[Path]) -> set[str]: return {repo.name for repo in repos} def resolve_family_slug(slug: Any, present: set[str], root: Path) -> bool: if not isinstance(slug, str) or not SLUG.match(slug): return False if slug in present: return True return (root / slug).is_dir() def check_resolution(loaded: Loaded, present: set[str], root: Path, report: Report) -> None: doc = loaded.doc if not isinstance(doc, dict): return family = loaded.family def need(field: str, slug: Any) -> None: if not isinstance(slug, str): return if not slug.startswith(FAMILY_PREFIXES): return if not resolve_family_slug(slug, present, root): report.add(loaded.rel, field, f"{slug} does not resolve under {root}") if family == "rapp": need("primary_rail", doc.get("primary_rail")) for i, rail in enumerate(as_list(doc.get("supported_rails"))): need(f"supported_rails[{i}]", rail) for i, reef in enumerate(as_list(doc.get("bound_reefs"))): need(f"bound_reefs[{i}]", reef) members = as_list((doc.get("composition") or {}).get("member_repos")) for i, member in enumerate(members): if not isinstance(member, dict): continue slug = member.get("repo") if not isinstance(slug, str): continue if slug == doc.get("repo"): continue if slug.startswith(FAMILY_PREFIXES) and not resolve_family_slug(slug, present, root): report.add( loaded.rel, f"composition.member_repos[{i}].repo", f"{slug} does not resolve under {root}", ) elif not slug.startswith(FAMILY_PREFIXES) and not (root / slug).is_dir(): report.add( loaded.rel, f"composition.member_repos[{i}].repo", f"{slug} is not present under {root}", level="warning", ) elif family == "rail": need("base_rail", doc.get("base_rail")) elif family == "reef": need("primary_rail", doc.get("primary_rail")) for i, rail in enumerate(as_list(doc.get("hosted_rails"))): need(f"hosted_rails[{i}]", rail) def check_bound_rapps(loaded_docs: list[Loaded], report: Report) -> None: derived: dict[str, set[str]] = defaultdict(set) rapp_ids: set[str] = set() for loaded in loaded_docs: if loaded.family != "rapp" or not isinstance(loaded.doc, dict): continue rapp_id = loaded.doc.get("rapp_id") if isinstance(rapp_id, str): rapp_ids.add(rapp_id) for reef in as_list(loaded.doc.get("bound_reefs")): if isinstance(reef, str) and isinstance(rapp_id, str): derived[reef].add(rapp_id) for loaded in loaded_docs: if loaded.family != "reef" or not isinstance(loaded.doc, dict): continue reef_id = loaded.doc.get("reef_id") if not isinstance(reef_id, str): continue expected = sorted(derived.get(reef_id, set())) declared = loaded.doc.get("bound_rapps") if declared is None: if expected: report.add( loaded.rel, "bound_rapps", f"omitted; derived projection is {expected}", level="warning", ) continue if not isinstance(declared, list): continue actual = sorted(str(item) for item in declared) if actual != expected: report.add( loaded.rel, "bound_rapps", f"hand-listed {actual} != derived {expected}", ) unknown = [item for item in actual if item not in rapp_ids and item.startswith("rapp-")] for item in unknown: report.add( loaded.rel, "bound_rapps", f"{item} is listed but no rapp declaration was loaded", ) def check_deployable_uniqueness(loaded_docs: list[Loaded], report: Report) -> None: owners = declared_deployables(loaded_docs) for deployable, claimed in owners.items(): rapps = sorted({rapp for _path, rapp in claimed}) if len(rapps) > 1: paths = ", ".join(f"{path} ({rapp})" for path, rapp in claimed) report.add( claimed[0][0], f"composition.member_repos.deployables.{deployable}", f"claimed by more than one rapp: {rapps} via {paths}", ) def declared_deployables(loaded_docs: list[Loaded]) -> dict[str, list[tuple[str, str]]]: owners: dict[str, list[tuple[str, str]]] = defaultdict(list) for loaded in loaded_docs: if loaded.family != "rapp" or not isinstance(loaded.doc, dict): continue rapp_id = str(loaded.doc.get("rapp_id") or loaded.repo_dir.name) members = as_list((loaded.doc.get("composition") or {}).get("member_repos")) for member in members: if not isinstance(member, dict): continue for deployable in as_list(member.get("deployables")): if isinstance(deployable, str): owners[deployable].append((loaded.rel, rapp_id)) return owners def load_inventory(path: Path) -> tuple[dict[str, Any] | None, Finding | None]: try: raw = path.read_text() data = json.loads(raw) except FileNotFoundError: return None, Finding(str(path), "(file)", "inventory file not found") except json.JSONDecodeError as exc: return None, Finding(str(path), "(file)", f"inventory is not JSON: {exc}") if not isinstance(data, dict): return None, Finding(str(path), "(root)", "inventory must be a mapping") deployables = data.get("deployables") if not isinstance(deployables, list): return None, Finding(str(path), "deployables", "must be a list") return data, None def check_inventory_coverage( loaded_docs: list[Loaded], inventory: dict[str, Any], inventory_path: str, report: Report, ) -> None: owners = declared_deployables(loaded_docs) live_names: list[str] = [] for i, item in enumerate(as_list(inventory.get("deployables"))): if isinstance(item, str): name = item elif isinstance(item, dict) and isinstance(item.get("name"), str): name = item["name"] else: report.add(inventory_path, f"deployables[{i}]", "each entry needs a name") continue live_names.append(name) claimed = owners.get(name, []) if not claimed: report.add( inventory_path, f"deployables.{name}", "live deployable is not claimed by any rapp (wave-2 worklist item)", ) elif len({rapp for _path, rapp in claimed}) > 1: rapps = sorted({rapp for _path, rapp in claimed}) report.add( inventory_path, f"deployables.{name}", f"live deployable is claimed by more than one rapp: {rapps}", ) live_set = set(live_names) for name, claimed in owners.items(): if name not in live_set: report.add( claimed[0][0], f"composition.member_repos.deployables.{name}", "declared deployable is not in the live inventory", level="warning", ) def check_undeclared( repos: list[Path], loaded_docs: list[Loaded], root: Path, report: Report ) -> None: declared = {loaded.repo_dir.resolve() for loaded in loaded_docs} for repo in repos: if repo.resolve() in declared: continue family = family_of(repo) if family is None: continue decl_name, _schema = DECL_BY_PREFIX[f"{family}-"] path = repo / "declarations" / decl_name try: rel = str(path.relative_to(root)) except ValueError: rel = str(path) report.add( rel, "(file)", f"{repo.name} claims the {family}- prefix but has no declarations/{decl_name}", ) def load_repo(repo_dir: Path, root: Path) -> tuple[Loaded | None, Finding | None]: family = family_of(repo_dir) if family is None: return None, Finding(str(repo_dir), "(file)", "not a rail-*, rapp-*, or reef-* repo") decl_name, _schema = DECL_BY_PREFIX[f"{family}-"] path = repo_dir / "declarations" / decl_name if not path.is_file(): return None, None try: rel = str(path.relative_to(root)) except ValueError: rel = str(path) try: doc = load_yaml(path) except yaml.YAMLError as exc: return None, Finding(rel, "(file)", f"YAML parse error: {exc}") if not isinstance(doc, dict): return None, Finding(rel, "(root)", "declaration must be a mapping") return Loaded(repo_dir=repo_dir, rel=rel, family=family, doc=doc), None def validate( root: Path, repos: list[Path], validators: dict[str, Draft202012Validator], inventory_path: Path | None = None, ) -> Report: report = Report() loaded_docs: list[Loaded] = [] present = present_slugs(repos) for repo in repos: loaded, finding = load_repo(repo, root) if finding is not None: report.findings.append(finding) continue if loaded is None: continue report.files += 1 loaded_docs.append(loaded) schema_check(loaded, validators[loaded.family], report) if loaded.family == "rail": check_rail(loaded, report) elif loaded.family == "rapp": check_rapp(loaded, report) elif loaded.family == "reef": check_reef(loaded, report) check_resolution(loaded, present, root, report) check_bound_rapps(loaded_docs, report) check_deployable_uniqueness(loaded_docs, report) check_undeclared(repos, loaded_docs, root, report) if inventory_path is not None: inventory, finding = load_inventory(inventory_path) if finding is not None: report.findings.append(finding) elif inventory is not None: check_inventory_coverage(loaded_docs, inventory, str(inventory_path), report) return report def print_report(report: Report) -> None: if not report.findings: print(f"{report.files} declaration(s) ok") return for finding in report.findings: print(finding) print( f"{report.files} declaration(s), " f"{len(report.errors)} error(s), " f"{len(report.warnings)} warning(s)" ) def self_test() -> int: validators = load_schemas() failures: list[str] = [] good_root = TESTDATA / "good" good = validate( good_root, discover_repos(good_root), validators, inventory_path=TESTDATA / "good" / "inventory.json", ) if good.errors: failures.append("good fixture produced errors:\n " + "\n ".join(str(f) for f in good.errors)) bad_root = TESTDATA / "bad-stale-bound" bad = validate( bad_root, discover_repos(bad_root), validators, inventory_path=TESTDATA / "bad-stale-bound" / "inventory.json", ) messages = "\n".join(str(f) for f in bad.errors) if not any("hand-listed" in f.message for f in bad.errors): failures.append(f"bad-stale-bound did not flag bound_rapps drift:\n{messages}") if not any("rapp-orphan" in f.path or "rapp-orphan" in f.message for f in bad.errors): failures.append(f"bad-stale-bound did not flag undeclared rapp-orphan:\n{messages}") if not any("wave-2 worklist" in f.message for f in bad.errors): failures.append(f"bad-stale-bound did not flag uncovered live deployable:\n{messages}") if not bad.errors: failures.append("bad-stale-bound produced no errors") live_rails = [ Path("/home/worsch/rail-kubernetes"), Path("/home/worsch/rail-knative"), ] if all(path.is_dir() for path in live_rails): live_root = Path("/home/worsch") live = validate(live_root, live_rails, validators) rail_errors = [f for f in live.errors if "/rail-" in f.path or f.path.startswith("rail-")] if rail_errors: failures.append( "live rails no longer conform:\n " + "\n ".join(str(f) for f in rail_errors) ) if failures: print("self-test FAILED") for item in failures: print(item) return 1 print("self-test ok") print(f" good: {good.files} file(s), {len(good.errors)} error(s)") print(f" bad-stale-bound: {bad.files} file(s), {len(bad.errors)} error(s) (expected)") return 0 def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0]) parser.add_argument( "--root", type=Path, default=REPO_ROOT.parent, help="directory that holds rail-*/rapp-*/reef-* siblings (default: parent of this repo)", ) parser.add_argument( "--repo", type=Path, action="append", default=[], help="extra family repo to include (repeatable)", ) parser.add_argument( "--inventory", type=Path, help="JSON inventory of live deployables produced by an implementation repo", ) parser.add_argument( "--self-test", action="store_true", help="run fixture checks and exit", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv if argv is not None else sys.argv[1:]) if args.self_test: return self_test() try: validators = load_schemas() except Exception as exc: # noqa: BLE001 — surface schema load failures cleanly print(f"failed to load schemas from {SCHEMA_DIR}: {exc}", file=sys.stderr) return 2 root = args.root.resolve() repos = discover_repos(root, args.repo) if not repos: print(f"no rail-*, rapp-*, or reef-* repos under {root}", file=sys.stderr) return 2 report = validate(root, repos, validators, inventory_path=args.inventory) print_report(report) return 1 if report.errors else 0 if __name__ == "__main__": sys.exit(main())