86 lines
2.7 KiB
Python
Executable file
86 lines
2.7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Check reef binding files against authoritative sibling declarations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
def load_yaml(path: Path) -> dict:
|
|
value = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path}: expected a YAML mapping")
|
|
return value
|
|
|
|
|
|
def ids(items: object, key: str) -> set[str]:
|
|
if not isinstance(items, list):
|
|
return set()
|
|
return {
|
|
item[key]
|
|
for item in items
|
|
if isinstance(item, dict) and isinstance(item.get(key), str)
|
|
}
|
|
|
|
|
|
def expected_rapps(parent: Path, reef_id: str) -> set[str]:
|
|
result: set[str] = set()
|
|
for declaration_path in sorted(parent.glob("rapp-*/declarations/rapp.yaml")):
|
|
declaration = load_yaml(declaration_path)
|
|
if reef_id in declaration.get("bound_reefs", []):
|
|
result.add(str(declaration["rapp_id"]))
|
|
return result
|
|
|
|
|
|
def expected_rails(parent: Path, declared: list[str]) -> set[str]:
|
|
result: set[str] = set()
|
|
for rail_id in declared:
|
|
declaration_path = parent / rail_id / "declarations" / "rail.yaml"
|
|
declaration = load_yaml(declaration_path)
|
|
result.add(str(declaration["rail_id"]))
|
|
return result
|
|
|
|
|
|
def assess(repo: Path, parent: Path) -> list[str]:
|
|
reef = load_yaml(repo / "declarations" / "reef.yaml")
|
|
reef_id = str(reef["reef_id"])
|
|
rapp_bindings = load_yaml(repo / "bindings" / "rapps.yaml")
|
|
rail_bindings = load_yaml(repo / "bindings" / "rails.yaml")
|
|
|
|
actual_rapps = ids(rapp_bindings.get("bound_rapps"), "rapp_id")
|
|
derived_rapps = expected_rapps(parent, reef_id)
|
|
actual_rails = ids(rail_bindings.get("hosted_rails"), "rail_id")
|
|
derived_rails = expected_rails(parent, list(reef.get("hosted_rails", [])))
|
|
|
|
failures: list[str] = []
|
|
if actual_rapps != derived_rapps:
|
|
failures.append(
|
|
f"rapp projection differs: binding={sorted(actual_rapps)}, "
|
|
f"derived={sorted(derived_rapps)}"
|
|
)
|
|
if actual_rails != derived_rails:
|
|
failures.append(
|
|
f"rail projection differs: binding={sorted(actual_rails)}, "
|
|
f"declared={sorted(derived_rails)}"
|
|
)
|
|
return failures
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--repo", type=Path, default=Path(__file__).parents[1])
|
|
parser.add_argument("--parent", type=Path)
|
|
args = parser.parse_args()
|
|
repo = args.repo.resolve()
|
|
parent = (args.parent or repo.parent).resolve()
|
|
failures = assess(repo, parent)
|
|
print(json.dumps({"pass": not failures, "failures": failures}))
|
|
return bool(failures)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|