mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-18 00:47:23 +02:00
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate conditional-gate policy and its CPU-coverage mappings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
REQUIRED_GATE_FIELDS = {
|
|
"id", "job", "marker", "runner", "prerequisites", "expected_cost", "coverage_paths"
|
|
}
|
|
|
|
|
|
def validate(policy_path: Path, quality_path: Path, workflow_path: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
policy = json.loads(policy_path.read_text())
|
|
quality = json.loads(quality_path.read_text())
|
|
workflow = workflow_path.read_text()
|
|
|
|
if policy.get("schema_version") != 1:
|
|
errors.append("conditional policy schema_version must be 1")
|
|
for key in ("owner", "cadence", "evidence_retention_days", "maximum_evidence_age_days"):
|
|
if not policy.get(key):
|
|
errors.append(f"conditional policy is missing {key}")
|
|
|
|
gates = policy.get("gates")
|
|
if not isinstance(gates, list) or not gates:
|
|
return errors + ["conditional policy gates must be a non-empty list"]
|
|
|
|
by_id: dict[str, dict] = {}
|
|
for index, gate in enumerate(gates):
|
|
missing = REQUIRED_GATE_FIELDS - set(gate)
|
|
if missing:
|
|
errors.append(f"gate {index} is missing fields: {sorted(missing)}")
|
|
continue
|
|
gate_id = gate["id"]
|
|
if gate_id in by_id:
|
|
errors.append(f"duplicate conditional gate id: {gate_id}")
|
|
by_id[gate_id] = gate
|
|
if not gate["coverage_paths"]:
|
|
errors.append(f"gate {gate_id} has no coverage paths")
|
|
if f"{gate['job']}:" not in workflow:
|
|
errors.append(f"workflow job {gate['job']!r} for {gate_id} was not found")
|
|
for source_path in gate["coverage_paths"]:
|
|
if not Path(source_path).is_file():
|
|
errors.append(f"gate {gate_id} maps missing source path: {source_path}")
|
|
|
|
exclusions = quality.get("mature_cpu_scope", {}).get("exclusions", [])
|
|
for exclusion in exclusions:
|
|
gate_id = exclusion.get("conditional_gate")
|
|
source_path = exclusion.get("path")
|
|
if gate_id not in by_id:
|
|
errors.append(f"CPU exclusion {source_path} references unknown gate {gate_id}")
|
|
continue
|
|
if source_path not in by_id[gate_id]["coverage_paths"]:
|
|
errors.append(f"CPU exclusion {source_path} is not mapped by gate {gate_id}")
|
|
|
|
required_workflow_tokens = (
|
|
"workflow_dispatch:", "schedule:", "release:", "permissions:", "contents: read",
|
|
"scripts/run_conditional_gate.py", "scripts/conditional_gate_summary.py",
|
|
)
|
|
for token in required_workflow_tokens:
|
|
if token not in workflow:
|
|
errors.append(f"conditional workflow is missing {token!r}")
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--policy", type=Path, default=Path("ci/conditional-test-policy.json"))
|
|
parser.add_argument("--quality", type=Path, default=Path("ci/test-quality-policy.json"))
|
|
parser.add_argument(
|
|
"--workflow", type=Path, default=Path(".github/workflows/conditional-tests.yml")
|
|
)
|
|
args = parser.parse_args()
|
|
errors = validate(args.policy, args.quality, args.workflow)
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}")
|
|
return 1
|
|
print("conditional test policy: valid")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|