4864b7852d
Rename RAILIANCE-WP-0017..0021 to RMASTER-WP-* so railiance-master IDs no longer collide with railiance-platform's RAILIANCE-WP series. Hub UUIDs are unchanged.
596 lines
21 KiB
Python
Executable file
596 lines
21 KiB
Python
Executable file
#!/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/<family>.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 hosted 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())
|