home-network-k3s-impl/conformance/lib/coverage.py

#!/usr/bin/env python3
"""Requirement-coverage report generator (R-TEST-04).

Maps EVERY R-* requirement in the authoritative registry to the Chainsaw test(s) that
verify it, and flags requirements with no verifying test as visible, blocking gaps.

Single source of truth for coverage is the per-requirement label on each Chainsaw Test:

    metadata:
      labels:
        conformance.home.k3s/R-NET-01: "true"

Reads:
  * requirements/registry.yaml   -- the mirror of every spec R-* ID (see that file).
  * tests/**/chainsaw-test.yaml   -- discovers labels + the requirements annotation.

Writes (under --out, default _artifacts/coverage/):
  * coverage.json  -- machine-readable {requirement: [tests...]}, gaps, safety gaps.
  * coverage.md    -- human-readable table for PRs / CI logs.

Exit status / gating (R-TEST-03 gates are *scoped*, not "all 51 always"):
  0  always in plain report mode.
  --strict         HYGIENE gate (always-on, WS-TEST owns it): fails on tag/annotation
                   drift or tags absent from the registry. Does NOT fail merely because
                   a future-phase requirement has no test yet — those are visible gaps,
                   not harness defects.
  --require IDs    SCOPED completeness gate (merge gate): fails if any listed requirement
                   is uncovered, or a listed safety-critical requirement lacks an L3 test.
                   CI passes the requirement IDs the change touches.
  --require-all    FULL completeness gate (phase-exit / final): fails if ANY registry
                   requirement is uncovered or any safety-critical one lacks an L3 test.
  --allow-gap IDs  exempt specific IDs from the completeness gates (documented, out-of-
                   scope gaps such as R-ENV-02/R-GITOPS-01 owned by another workstream).

Pure standard library (no PyYAML) so the harness has no pip dependency: the registry and
the test label/annotation lines we care about are parsed with line-oriented regexes, and
the file formats are constrained (documented in registry.yaml / common.sh) to keep that
parse deterministic.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from pathlib import Path

LABEL_PREFIX = "conformance.home.k3s"
RID_RE = re.compile(r"R-[A-Z0-9]+-[0-9]+")
LABEL_LINE_RE = re.compile(
    r"^\s*" + re.escape(LABEL_PREFIX) + r"/(R-[A-Z0-9]+-[0-9]+)\s*:\s*[\"']?true", re.I
)


# --------------------------------------------------------------------------- registry
def parse_registry(path: Path) -> list[dict]:
    """Parse the constrained registry.yaml (`- id:` blocks, single-line scalars)."""
    reqs: list[dict] = []
    cur: dict | None = None
    for raw in path.read_text().splitlines():
        line = raw.split("#", 1)[0] if raw.lstrip().startswith("#") else raw
        m = re.match(r"^\s*-\s*id:\s*(\S+)", line)
        if m:
            if cur:
                reqs.append(cur)
            cur = {"id": m.group(1).strip(), "doc": "", "safety_critical": False,
                   "summary": ""}
            continue
        if cur is None:
            continue
        m = re.match(r"^\s+doc:\s*(.+?)\s*$", line)
        if m:
            cur["doc"] = m.group(1).strip().strip('"')
            continue
        m = re.match(r"^\s+safety_critical:\s*(true|false)\s*$", line)
        if m:
            cur["safety_critical"] = (m.group(1) == "true")
            continue
        m = re.match(r"^\s+summary:\s*(.+?)\s*$", line)
        if m:
            cur["summary"] = m.group(1).strip().strip('"')
    if cur:
        reqs.append(cur)
    return reqs


# ------------------------------------------------------------------------------ tests
def parse_test(path: Path) -> dict:
    """Extract name, verified R-IDs (from labels), annotation list, and layer."""
    text = path.read_text()
    labels: set[str] = set()
    for line in text.splitlines():
        m = LABEL_LINE_RE.match(line)
        if m:
            labels.add(m.group(1).upper())

    ann_reqs: set[str] = set()
    m = re.search(r"^\s*" + re.escape(LABEL_PREFIX) + r"/requirements:\s*(.+)$",
                  text, re.M)
    if m:
        ann_reqs = {r.upper() for r in RID_RE.findall(m.group(1))}

    layer = ""
    m = re.search(r"^\s*" + re.escape(LABEL_PREFIX) + r"/layer:\s*[\"']?([LlAaMm0-9]+)",
                  text, re.M)
    if m:
        layer = m.group(1).upper()

    name = ""
    m = re.search(r"^\s*name:\s*(\S+)", text, re.M)
    if m:
        name = m.group(1)

    # Layer inferred from path as a fallback / cross-check.
    rel = str(path)
    if not layer:
        if "l0-static" in rel:
            layer = "L0"
        elif "l1-rendering" in rel:
            layer = "L1"
        elif "l2-behavioral" in rel:
            layer = "L2"
        elif "l3-fault-injection" in rel:
            layer = "L3"
        elif "/meta/" in rel:
            layer = "META"

    return {
        "name": name,
        "path": rel,
        "labels": sorted(labels),
        "annotation_requirements": sorted(ann_reqs),
        "layer": layer,
    }


def discover_tests(tests_dir: Path) -> list[dict]:
    return [parse_test(p) for p in sorted(tests_dir.rglob("chainsaw-test.yaml"))]


# ----------------------------------------------------------------------------- report
def build(registry: list[dict], tests: list[dict]) -> dict:
    by_req: dict[str, list[dict]] = {r["id"]: [] for r in registry}
    unknown: list[tuple[str, str]] = []       # (rid, test path) not in registry
    mismatches: list[dict] = []               # label set != annotation set

    for t in tests:
        lbl = set(t["labels"])
        ann = set(t["annotation_requirements"])
        if lbl != ann:
            mismatches.append({"test": t["path"], "labels": sorted(lbl),
                               "annotation": sorted(ann)})
        for rid in lbl:
            if rid in by_req:
                by_req[rid].append(t)
            else:
                unknown.append((rid, t["path"]))

    reg_index = {r["id"]: r for r in registry}
    covered, uncovered, safety_gaps = [], [], []
    for rid, req in reg_index.items():
        verifiers = by_req[rid]
        if verifiers:
            covered.append(rid)
        else:
            uncovered.append(rid)
        if req["safety_critical"]:
            has_l3 = any(v["layer"] == "L3" for v in verifiers)
            if not has_l3:
                safety_gaps.append(rid)

    return {
        "summary": {
            "total_requirements": len(registry),
            "covered": len(covered),
            "uncovered": len(uncovered),
            "coverage_pct": round(100.0 * len(covered) / max(1, len(registry)), 1),
            "safety_critical_gaps": len(safety_gaps),
            "annotation_label_mismatches": len(mismatches),
            "unknown_requirement_tags": len(unknown),
            "total_tests": len(tests),
        },
        "by_requirement": {
            r["id"]: {
                "doc": r["doc"],
                "safety_critical": r["safety_critical"],
                "summary": r["summary"],
                "verified_by": [
                    {"name": v["name"], "path": v["path"], "layer": v["layer"]}
                    for v in by_req[r["id"]]
                ],
            }
            for r in registry
        },
        "uncovered": sorted(uncovered),
        "safety_critical_gaps": sorted(safety_gaps),
        "annotation_label_mismatches": mismatches,
        "unknown_requirement_tags": [{"requirement": u[0], "test": u[1]} for u in unknown],
    }


def render_md(rep: dict) -> str:
    s = rep["summary"]
    out: list[str] = []
    out.append("# Requirement coverage report\n")
    out.append(
        f"- Requirements: **{s['total_requirements']}** | "
        f"Covered: **{s['covered']}** ({s['coverage_pct']}%) | "
        f"Uncovered: **{s['uncovered']}**\n"
        f"- Safety-critical requirements missing an L3 fault-injection test: "
        f"**{s['safety_critical_gaps']}**\n"
        f"- Scaffolded tests: **{s['total_tests']}** | "
        f"Annotation/label mismatches: **{s['annotation_label_mismatches']}** | "
        f"Unknown tags: **{s['unknown_requirement_tags']}**\n"
    )
    out.append("\n> Phase 0: tests are scaffolds. A requirement is **satisfied** only "
               "when its tagged test passes green against a cluster (R-TEST-03) — this "
               "report tracks *scaffold* coverage, i.e. that a verifying test exists.\n")

    out.append("\n## Coverage by requirement\n")
    out.append("| Requirement | Safety | Verified by (layer) | Status |")
    out.append("|---|---|---|---|")
    for rid, info in rep["by_requirement"].items():
        vs = info["verified_by"]
        if vs:
            cell = "<br>".join(f"`{v['name']}` ({v['layer']})" for v in vs)
            status = "covered"
        else:
            cell = "—"
            status = "**GAP**"
        sc = "yes" if info["safety_critical"] else ""
        out.append(f"| {rid} | {sc} | {cell} | {status} |")

    if rep["uncovered"]:
        out.append("\n## Uncovered requirements (gaps)\n")
        for rid in rep["uncovered"]:
            out.append(f"- {rid} — {rep['by_requirement'][rid]['summary']}")
    if rep["safety_critical_gaps"]:
        out.append("\n## Safety-critical requirements without an L3 fault-injection test "
                   "(R-TEST-05)\n")
        for rid in rep["safety_critical_gaps"]:
            out.append(f"- {rid}")
    if rep["annotation_label_mismatches"]:
        out.append("\n## Tests whose labels disagree with their `requirements` annotation\n")
        for m in rep["annotation_label_mismatches"]:
            out.append(f"- `{m['test']}` labels={m['labels']} annotation={m['annotation']}")
    if rep["unknown_requirement_tags"]:
        out.append("\n## Tags not present in the registry (typo or spec drift)\n")
        for u in rep["unknown_requirement_tags"]:
            out.append(f"- {u['requirement']} in `{u['test']}`")
    return "\n".join(out) + "\n"


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    here = Path(__file__).resolve().parent.parent  # conformance/
    ap.add_argument("--registry", type=Path,
                    default=here / "requirements" / "registry.yaml")
    ap.add_argument("--tests", type=Path, default=here / "tests")
    ap.add_argument("--out", type=Path, default=here / "_artifacts" / "coverage")
    ap.add_argument("--strict", action="store_true",
                    help="hygiene gate: fail on tag/annotation drift or unknown tags")
    ap.add_argument("--require", default="",
                    help="comma/space list of R-IDs that MUST be covered (merge gate)")
    ap.add_argument("--require-all", action="store_true",
                    help="every registry requirement MUST be covered (phase-exit gate)")
    ap.add_argument("--allow-gap", default="",
                    help="comma/space list of R-IDs exempt from completeness gates")
    args = ap.parse_args()

    registry = parse_registry(args.registry)
    tests = discover_tests(args.tests)
    rep = build(registry, tests)

    args.out.mkdir(parents=True, exist_ok=True)
    (args.out / "coverage.json").write_text(json.dumps(rep, indent=2) + "\n")
    (args.out / "coverage.md").write_text(render_md(rep))

    s = rep["summary"]
    print(f"[coverage] {s['covered']}/{s['total_requirements']} requirements have a "
          f"scaffolded test ({s['coverage_pct']}%); {s['uncovered']} gap(s), "
          f"{s['safety_critical_gaps']} safety-critical gap(s).")
    print(f"[coverage] wrote {args.out/'coverage.md'} and {args.out/'coverage.json'}")

    def split(v):
        return {t.upper() for t in re.split(r"[,\s]+", v) if t}

    exempt = split(args.allow_gap)
    uncovered = set(rep["uncovered"])
    safety_gaps = set(rep["safety_critical_gaps"])
    rc = 0

    # HYGIENE (always-on): things WS-TEST controls and must keep green.
    if args.strict:
        if rep["annotation_label_mismatches"] or rep["unknown_requirement_tags"]:
            print("[coverage] STRICT(hygiene): FAILED — label/annotation drift or "
                  "tags not in the registry.", file=sys.stderr)
            rc = 1

    # SCOPED completeness (merge gate): only the requirements the change touches.
    required = split(args.require)
    if args.require_all:
        required |= {r["id"] for r in registry}
    required -= exempt
    if required:
        miss = sorted(required & uncovered)
        sgap = sorted(required & safety_gaps)
        if miss:
            print(f"[coverage] REQUIRE: FAILED — required requirement(s) uncovered: "
                  f"{miss}", file=sys.stderr)
            rc = 1
        if sgap:
            print(f"[coverage] REQUIRE: FAILED — required safety-critical requirement(s) "
                  f"lack an L3 fault-injection test: {sgap}", file=sys.stderr)
            rc = 1
    return rc


if __name__ == "__main__":
    raise SystemExit(main())