160 lines
6.7 KiB
Python
160 lines
6.7 KiB
Python
"""Read-only RC plan/ledger consistency checks; not artifact qualification."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CONFIG = "config/release/rc-program.json"
|
|
LEDGER = "docs/quality/RC_GAP_MATRIX.md"
|
|
ROADMAP = "docs/product/RC_EXECUTION_ROADMAP.md"
|
|
GAP_ID = re.compile(r"RC-[A-Z]+-[0-9]{2}")
|
|
PACKET_ID = re.compile(r"P[0-9]{2}[A-Z]?")
|
|
CLASSIFICATIONS = {
|
|
"COMPLETE", "NEEDS POLISH", "FUNCTIONALLY INCOMPLETE", "UX INCOMPLETE",
|
|
"OPERATIONALLY INCOMPLETE", "SECURITY BLOCKER", "RC BLOCKER", "POST-RC",
|
|
}
|
|
STATUSES = {"OPEN", "IN PROGRESS", "RESOLVED", "VERIFIED", "DEFERRED"}
|
|
|
|
|
|
class InvalidPlan(ValueError):
|
|
"""The plan cannot safely be interpreted."""
|
|
|
|
|
|
def _object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise InvalidPlan(f"duplicate JSON field: {key}")
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _read(root: Path, relative: str) -> str:
|
|
path = (root / relative).resolve()
|
|
if not path.is_relative_to(root.resolve()):
|
|
raise InvalidPlan(f"repository path escapes root: {relative}")
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def _keys(value: Any, expected: set[str], label: str) -> None:
|
|
if not isinstance(value, dict) or set(value) != expected:
|
|
raise InvalidPlan(f"invalid {label} fields")
|
|
|
|
|
|
def _ids(value: Any, pattern: re.Pattern[str], label: str) -> list[str]:
|
|
if not isinstance(value, list) or any(
|
|
not isinstance(item, str) or pattern.fullmatch(item) is None for item in value
|
|
):
|
|
raise InvalidPlan(f"invalid {label} IDs")
|
|
if len(set(value)) != len(value):
|
|
raise InvalidPlan(f"duplicate {label} IDs")
|
|
return value
|
|
|
|
|
|
def validate(root: Path = ROOT) -> dict[str, Any]:
|
|
plan = json.loads(_read(root, CONFIG), object_pairs_hook=_object)
|
|
_keys(plan, {"schema_version", "target_version", "baseline_commit", "ledger", "roadmap",
|
|
"packets", "deferred_gaps"}, "configuration")
|
|
if type(plan["schema_version"]) is not int or plan["schema_version"] != 1:
|
|
raise InvalidPlan("unsupported schema_version")
|
|
if not isinstance(plan["target_version"], str) or not re.fullmatch(
|
|
r"[0-9]+\.[0-9]+\.[0-9]+-rc\.[1-9][0-9]*", plan["target_version"]
|
|
):
|
|
raise InvalidPlan("invalid target_version")
|
|
if not isinstance(plan["baseline_commit"], str) or not re.fullmatch(
|
|
r"[0-9a-f]{40}", plan["baseline_commit"]
|
|
):
|
|
raise InvalidPlan("invalid baseline_commit")
|
|
if plan["ledger"] != LEDGER or plan["roadmap"] != ROADMAP:
|
|
raise InvalidPlan("ledger and roadmap must use the fixed repository paths")
|
|
if not _read(root, ROADMAP).strip():
|
|
raise InvalidPlan("roadmap is empty")
|
|
rows: dict[str, tuple[str, str]] = {}
|
|
for line in _read(root, LEDGER).splitlines():
|
|
if not line.startswith("| RC-"):
|
|
continue
|
|
cells = [cell.strip() for cell in re.split(r"(?<!\\)\|", line)[1:-1]]
|
|
if not line.endswith("|") or len(cells) != 7 or not GAP_ID.fullmatch(cells[0]):
|
|
raise InvalidPlan("malformed RC ledger row")
|
|
gap, _, _, classification, _, _, status = cells
|
|
if gap in rows:
|
|
raise InvalidPlan(f"duplicate ledger ID: {gap}")
|
|
if classification not in CLASSIFICATIONS:
|
|
raise InvalidPlan(f"unknown classification: {gap}")
|
|
if status.startswith("RESOLVED — ") and status.removeprefix("RESOLVED — ").strip():
|
|
status = "RESOLVED"
|
|
if status not in STATUSES:
|
|
raise InvalidPlan(f"unknown status: {gap}")
|
|
if (status == "DEFERRED") != (classification == "POST-RC"):
|
|
raise InvalidPlan(f"invalid deferral: {gap}")
|
|
rows[gap] = classification, status
|
|
if not rows:
|
|
raise InvalidPlan("ledger has no RC issue rows")
|
|
packets = plan["packets"]
|
|
if not isinstance(packets, list) or not packets:
|
|
raise InvalidPlan("packets must be a nonempty list")
|
|
dependencies: dict[str, list[str]] = {}
|
|
owners: dict[str, str] = {}
|
|
for packet in packets:
|
|
_keys(packet, {"id", "depends_on", "gaps"}, "packet")
|
|
packet_id = packet["id"]
|
|
_ids([packet_id], PACKET_ID, "packet")
|
|
if packet_id in dependencies:
|
|
raise InvalidPlan(f"duplicate packet ID: {packet_id}")
|
|
dependencies[packet_id] = _ids(packet["depends_on"], PACKET_ID, "dependency")
|
|
gaps = _ids(packet["gaps"], GAP_ID, "gap")
|
|
if not gaps:
|
|
raise InvalidPlan(f"packet has no gaps: {packet_id}")
|
|
for gap in gaps:
|
|
if gap in owners:
|
|
raise InvalidPlan(f"duplicate gap ownership: {gap}")
|
|
owners[gap] = packet_id
|
|
deferred = set(_ids(plan["deferred_gaps"], GAP_ID, "deferred gap"))
|
|
if deferred & owners.keys():
|
|
raise InvalidPlan("deferred gaps also assigned to packets")
|
|
if set(rows) != owners.keys() | deferred:
|
|
raise InvalidPlan("ledger coverage mismatch: unknown or unmapped gaps")
|
|
if deferred != {gap for gap, (kind, _) in rows.items() if kind == "POST-RC"}:
|
|
raise InvalidPlan("only POST-RC rows may be deferred")
|
|
order: list[str] = []
|
|
for packet_id, refs in dependencies.items():
|
|
if set(refs) - dependencies.keys():
|
|
raise InvalidPlan(f"unknown dependencies: {packet_id}")
|
|
remaining = dict(dependencies)
|
|
while remaining:
|
|
available = sorted(key for key, refs in remaining.items() if set(refs) <= set(order))
|
|
if not available:
|
|
raise InvalidPlan("packet dependency cycle")
|
|
order.extend(available)
|
|
for key in available:
|
|
del remaining[key]
|
|
outstanding = sorted(gap for gap in owners if rows[gap][1] != "VERIFIED")
|
|
return {
|
|
"packet_count": len(packets), "required_gap_count": len(owners),
|
|
"verified_count": len(owners) - len(outstanding), "outstanding_ids": outstanding,
|
|
"topological_order": order, "all_required_rows_verified": not outstanding,
|
|
"scope": "ledger consistency only; artifact qualification is separate",
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--require-verified", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
result = validate()
|
|
except (ValueError, OSError) as exc:
|
|
print(json.dumps({"error": str(exc), "all_required_rows_verified": False}))
|
|
return 2
|
|
print(json.dumps(result, sort_keys=True))
|
|
return int(args.require_verified and not result["all_required_rows_verified"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|