SlabUploader/tools/check_coverage.py

#!/usr/bin/env python3
"""Enforce the coverage ledger described in .audit/coverage/README.md.

Run from the repo root:

    python tools/check_coverage.py            # forward checks; report gaps
    python tools/check_coverage.py --strict   # also require full reverse coverage
    python tools/check_coverage.py --ledger-dir <path>

Forward checks always run. `--strict` adds the reverse rule and fails on any `gap`.
Exit 0 = clean for the selected mode. Exit 1 = problems.
"""
from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
REQUIREMENTS = ROOT / "docs" / "REQUIREMENTS.md"
DEFAULT_LEDGER = ROOT / ".audit" / "coverage"

REQUIREMENT_ROW = re.compile(
    r"^\|\s*([A-Z]+)-(\d+)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*$"
)
SEPARATOR = re.compile(r"^-+$")

DISPOSITIONS = {"ported", "duplicate", "new", "superseded", "deferred", "dropped", "gap"}
NEEDS_ID = {"ported", "new"}
FORBIDS_ID = {"dropped", "gap"}
PLACEHOLDER = {"", "\u2014", "-"}


def defined_ids() -> set[str]:
    ids: set[str] = set()
    if not REQUIREMENTS.exists():
        return ids
    for line in REQUIREMENTS.read_text(encoding="utf-8").splitlines():
        match = REQUIREMENT_ROW.match(line)
        if match:
            ids.add(f"{match.group(1)}-{match.group(2)}")
    return ids


def cells_of(line: str) -> list[str] | None:
    """Return the six cells of a ledger data row, or None for non-rows."""
    stripped = line.strip()
    if not stripped.startswith("|"):
        return None
    cells = [cell.strip() for cell in stripped.split("|")][1:-1]
    if not cells:
        return None
    if cells[0].lower() == "source":
        return None
    if all(SEPARATOR.match(cell) for cell in cells if cell):
        return None
    return cells


def parse_ids(field: str) -> list[str]:
    parts = [part.strip() for part in field.split(",")]
    return [part for part in parts if part not in PLACEHOLDER]


def main() -> int:
    parser = argparse.ArgumentParser(description="Check the coverage ledger.")
    parser.add_argument("--strict", action="store_true", help="require full reverse coverage")
    parser.add_argument("--ledger-dir", type=Path, default=DEFAULT_LEDGER)
    args = parser.parse_args()

    ids = defined_ids()
    errors: list[str] = []
    gaps: list[str] = []
    claimed: set[str] = set()
    seen_atoms: dict[tuple[str, str], str] = {}
    counts: dict[str, int] = {}

    ledger_dir: Path = args.ledger_dir
    if not ledger_dir.exists():
        print(f"FAIL check_coverage: ledger directory missing: {ledger_dir}")
        return 1

    files = [path for path in sorted(ledger_dir.glob("*.md")) if path.name != "README.md"]
    if not files:
        print(f"FAIL check_coverage: no ledger files in {ledger_dir}")
        return 1

    for path in files:
        for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            cells = cells_of(line)
            if cells is None:
                continue
            loc = f"{path.name}:{number}"
            if len(cells) != 6:
                errors.append(f"{loc} expected 6 columns, got {len(cells)}")
                continue
            source, atom, requirement, disposition, poer, rationale = cells
            if not source or not atom or not requirement:
                errors.append(f"{loc} empty Source, Atom, or Requirement")
                continue
            if disposition not in DISPOSITIONS:
                errors.append(f"{loc} unknown disposition '{disposition}'")
                continue

            counts[disposition] = counts.get(disposition, 0) + 1
            key = (source, atom)
            if key in seen_atoms:
                errors.append(f"{loc} duplicate atom {source} {atom} (first at {seen_atoms[key]})")
            else:
                seen_atoms[key] = loc

            cited = parse_ids(poer)
            if disposition in NEEDS_ID and not cited:
                errors.append(f"{loc} {source} {atom}: {disposition} needs a PoR ID")
            if disposition in FORBIDS_ID and cited:
                errors.append(f"{loc} {source} {atom}: {disposition} must not cite a PoR ID")
            if disposition != "ported" and rationale in PLACEHOLDER:
                errors.append(f"{loc} {source} {atom}: {disposition} needs a rationale")
            for rid in cited:
                if rid not in ids:
                    errors.append(f"{loc} {source} {atom}: unknown PoR ID {rid}")
                else:
                    claimed.add(rid)

            if disposition == "gap":
                gaps.append(f"{source} {atom}: {rationale}")

    if args.strict:
        for rid in sorted(ids):
            if rid not in claimed:
                errors.append(f"unclaimed requirement ID (no source atom and not marked new): {rid}")
        for gap in gaps:
            errors.append(f"open gap: {gap}")

    total_rows = sum(counts.values())
    summary = ", ".join(f"{key}={counts[key]}" for key in sorted(counts))
    claimed_pct = (len(claimed) / len(ids) * 100) if ids else 0.0

    if errors:
        print(f"FAIL check_coverage: {len(errors)} problem(s)")
        for error in errors:
            print(f"  - {error}")
        return 1

    print(
        f"PASS check_coverage: {total_rows} atoms ({summary}); "
        f"claimed {len(claimed)}/{len(ids)} IDs ({claimed_pct:.0f}%); "
        f"{len(gaps)} gap(s)"
    )
    for gap in gaps:
        print(f"  gap: {gap}")
    return 0


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