#!/usr/bin/env python3 """Validate conditional-gate policy and its CPU-coverage mappings.""" from __future__ import annotations import argparse import json import re from pathlib import Path REQUIRED_GATE_FIELDS = { "id", "job", "marker", "runner", "prerequisites", "expected_cost", "coverage_paths" } SOFTWARE_ONLY_GATES = ( "model-download-runtime", "external-evaluation", "network-services", "operator-ui", ) SHA = re.compile(r"^[0-9a-f]{40}$") ISSUE_URL = "https://github.com/elder-plinius/OBLITERATUS/issues/" 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 _load_json_object(path: Path, label: str, errors: list[str]) -> dict: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: errors.append(f"cannot read {label}: {exc}") return {} if not isinstance(value, dict): errors.append(f"{label} root must be an object") return {} return value def _valid_stale_exception(reason: str | None, issue: str | None) -> bool: return ( isinstance(reason, str) and bool(reason.strip()) and isinstance(issue, str) and issue.startswith(ISSUE_URL) ) def validate_evidence( policy_path: Path, evidence_dir: Path, *, candidate_sha: str, required_gates: list[str] | None = None, stale_exception_reason: str | None = None, stale_exception_issue: str | None = None, ) -> list[str]: """Validate selected software-only conditional evidence against a candidate SHA.""" errors: list[str] = [] policy = _load_json_object(policy_path, "conditional policy", errors) if errors: return errors if SHA.fullmatch(candidate_sha) is None: errors.append("candidate SHA must be a 40-character lowercase hex commit") gates = policy.get("gates") policy_gate_ids = { gate.get("id") for gate in gates if isinstance(gates, list) and isinstance(gate, dict) } if isinstance(gates, list) else set() requested = required_gates or list(SOFTWARE_ONLY_GATES) for gate_id in requested: if gate_id not in SOFTWARE_ONLY_GATES: errors.append(f"hardware or credential gate is not software-only: {gate_id}") if gate_id not in policy_gate_ids: errors.append(f"conditional policy does not define gate {gate_id}") exception = _valid_stale_exception(stale_exception_reason, stale_exception_issue) if (stale_exception_reason or stale_exception_issue) and not exception: errors.append( "stale evidence exception requires a non-empty reason and an OBLITERATUS issue URL", ) for gate_id in requested: evidence = _load_json_object( evidence_dir / f"{gate_id}.json", f"conditional evidence {gate_id}", errors, ) if not evidence: continue if evidence.get("gate") != gate_id: errors.append(f"conditional evidence {gate_id} records gate {evidence.get('gate')!r}") if evidence.get("status") != "passed": errors.append(f"conditional evidence {gate_id} did not pass: {evidence.get('status')!r}") evidence_sha = evidence.get("git_sha") if evidence_sha != candidate_sha and not exception: errors.append( f"conditional evidence {gate_id} git_sha {evidence_sha!r} " f"does not match candidate {candidate_sha}", ) 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") ) parser.add_argument( "--evidence-dir", type=Path, help="validate software-only conditional evidence files in this directory", ) parser.add_argument( "--candidate-sha", help="40-character candidate commit SHA required for evidence freshness validation", ) parser.add_argument( "--require-gate", action="append", default=[], help="software-only gate that must have current passed evidence (repeatable)", ) parser.add_argument( "--stale-evidence-reason", default="", help="maintainer reason for accepting older software conditional evidence", ) parser.add_argument( "--stale-evidence-issue", default="", help="OBLITERATUS issue URL approving older software conditional evidence", ) args = parser.parse_args() errors = validate(args.policy, args.quality, args.workflow) if args.evidence_dir is not None: if not args.candidate_sha: errors.append("evidence freshness validation requires --candidate-sha") else: errors.extend( validate_evidence( args.policy, args.evidence_dir, candidate_sha=args.candidate_sha, required_gates=args.require_gate or None, stale_exception_reason=args.stale_evidence_reason or None, stale_exception_issue=args.stale_evidence_issue or None, ), ) 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())