gps-denied-onboard/scripts/gen_ac_traceability.py

"""Generate .planning/AC-TRACEABILITY.md from pytest collection.

Bidirectional AC<->test traceability per Phase 2 / AC-06.

Forward orphan  : declared AC with no test         -> visible + (with --check) exit 1
Backward orphan : test references unknown AC ID    -> visible + (with --check) exit 1
Deferred-hw AC  : `validation_method: deferred-hardware` in the AC entry
                  -> excluded from orphan check, rendered as "DEFERRED (hardware)"

Usage:
    python scripts/gen_ac_traceability.py             # write matrix, exit 0
    python scripts/gen_ac_traceability.py --check     # CI: exit 1 on drift
"""
from __future__ import annotations

import argparse
import json
import logging
import re
import subprocess
import sys
import tempfile
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("gen_ac_traceability")

ROOT = Path(__file__).resolve().parent.parent
AC_DOC = ROOT / "_docs" / "00_problem" / "acceptance_criteria.md"
OUT_MD = ROOT / ".planning" / "AC-TRACEABILITY.md"

_AC_HEADER_RE = re.compile(r"^\s*-\s*\*\*(AC-(?:\d+\.\d+[a-z]?|NEW-\d+))\*\*")
_DEFERRED_RE = re.compile(r"deferred-hardware", re.IGNORECASE)


def collect_acs_from_doc() -> dict[str, dict]:
    """Parse _docs/00_problem/acceptance_criteria.md.

    Returns: {ac_id: {"deferred": bool}}.
    An AC is `deferred` if the literal token `deferred-hardware` (case-insensitive)
    appears anywhere in the block between this AC's header and the next AC's header
    (or EOF).
    """
    if not AC_DOC.exists():
        log.error("AC doc not found: %s", AC_DOC)
        sys.exit(2)

    acs: dict[str, dict] = {}
    current_id: str | None = None
    for line in AC_DOC.read_text(encoding="utf-8").splitlines():
        m = _AC_HEADER_RE.match(line)
        if m:
            current_id = m.group(1)
            acs[current_id] = {"deferred": False}
            continue
        if current_id is not None and _DEFERRED_RE.search(line):
            acs[current_id]["deferred"] = True
    return acs


def collect_acs_from_tests() -> dict[str, list[str]]:
    """Run `pytest --collect-only --ac-dump=<tmp>` and parse the JSON.

    Returns: {ac_id: [test_nodeid, ...]}.
    """
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp:
        tmp_path = Path(tmp.name)
    try:
        result = subprocess.run(
            ["pytest", "--collect-only", "-q", f"--ac-dump={tmp_path}", "tests/"],
            cwd=ROOT,
            check=False,             # collection may report 0 errors but exit non-zero on warnings -- handle below
            capture_output=True,
            text=True,
        )
        # Collection errors are a fatal traceability problem; surface stderr.
        if result.returncode != 0 and not tmp_path.exists():
            log.error("pytest --collect-only failed:\n%s", result.stderr or result.stdout)
            sys.exit(2)
        if not tmp_path.exists() or tmp_path.stat().st_size == 0:
            return {}
        return json.loads(tmp_path.read_text(encoding="utf-8"))
    finally:
        tmp_path.unlink(missing_ok=True)


def render_md(doc_acs: dict[str, dict], test_map: dict[str, list[str]]) -> str:
    """Render the AC-TRACEABILITY.md content."""
    declared = sorted(doc_acs)
    tested = set(test_map)
    declared_set = set(declared)

    lines: list[str] = [
        "# AC Traceability Matrix",
        "",
        "> Auto-generated by `scripts/gen_ac_traceability.py`. Do not edit by hand.",
        "> Run `python scripts/gen_ac_traceability.py` to regenerate after AC doc or test edits.",
        "",
        f"**ACs declared in acceptance_criteria.md:** {len(declared)}",
        f"**ACs covered by at least one test:** {len(declared_set & tested)}",
        f"**ACs deferred to hardware:** {sum(1 for a in declared if doc_acs[a]['deferred'])}",
        "",
        "## AC -> Test mapping",
        "",
        "| AC ID | Test count | Tests | Status |",
        "|-------|-----------|-------|--------|",
    ]
    for ac_id in declared:
        tests = test_map.get(ac_id, [])
        if doc_acs[ac_id]["deferred"]:
            status = "DEFERRED (hardware)"
        elif tests:
            status = "OK"
        else:
            status = "**ORPHAN -- no test**"
        tests_str = "<br>".join(f"`{t}`" for t in tests) if tests else "_none_"
        lines.append(f"| {ac_id} | {len(tests)} | {tests_str} | {status} |")

    unknown = sorted(tested - declared_set)
    if unknown:
        lines += ["", "## Tests reference unknown AC IDs (backward orphans)", ""]
        for ac_id in unknown:
            for nid in test_map[ac_id]:
                lines.append(f"- `{ac_id}` <- `{nid}`")

    return "\n".join(lines) + "\n"


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument(
        "--check", "--strict", action="store_true",
        help="Exit 1 if any non-deferred AC has 0 tests, or any test references an AC ID not in the doc.",
    )
    args = p.parse_args()

    doc_acs = collect_acs_from_doc()
    test_map = collect_acs_from_tests()
    out = render_md(doc_acs, test_map)
    OUT_MD.parent.mkdir(parents=True, exist_ok=True)
    OUT_MD.write_text(out, encoding="utf-8")
    log.info("wrote %s (%d AC, %d tagged, %d deferred)",
             OUT_MD.relative_to(ROOT),
             len(doc_acs),
             sum(1 for a in doc_acs if test_map.get(a)),
             sum(1 for a in doc_acs if doc_acs[a]["deferred"]))

    if args.check:
        bad_orphans = sorted(
            a for a, meta in doc_acs.items() if not meta["deferred"] and not test_map.get(a)
        )
        unknown = sorted(set(test_map) - set(doc_acs))
        if bad_orphans or unknown:
            log.error("AC traceability drift detected:")
            for a in bad_orphans:
                log.error("  ORPHAN AC (no test):       %s", a)
            for a in unknown:
                log.error("  UNKNOWN AC ID in test:     %s", a)
            return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())