test: enforce model and CLI boundary contracts

scripts/check_coverage_thresholds.py

@@ -1,16 +1,19 @@
-"""Enforce separate line and branch floors from a coverage.py JSON report."""
+"""Enforce global, critical-file, and changed-line floors from coverage.py JSON."""
 
 from __future__ import annotations
 
 import argparse
 import json
 import math
+import re
+import subprocess
 from pathlib import Path
 from typing import Any
 
 
 def validate_coverage(
     report: dict[str, Any], *, min_line: float, min_branch: float,
+    file_floors: dict[str, float] | None = None,
 ) -> list[str]:
     """Return human-readable failures for coverage totals below their floors."""
     totals = report.get("totals")
@@ -34,14 +37,111 @@ def validate_coverage(
             failures.append(
                 f"{label} coverage {value:.2f}% is below the {minimum:.2f}% floor",
             )
+
+    files = report.get("files")
+    for path, minimum in (file_floors or {}).items():
+        if not isinstance(files, dict) or not isinstance(files.get(path), dict):
+            failures.append(f"coverage report is missing critical file {path}")
+            continue
+        summary = files[path].get("summary")
+        value = summary.get("percent_statements_covered") if isinstance(summary, dict) else None
+        if (
+            isinstance(value, bool)
+            or not isinstance(value, (int, float))
+            or not math.isfinite(value)
+        ):
+            failures.append(f"coverage report is missing numeric line coverage for {path}")
+        elif value < minimum:
+            failures.append(
+                f"critical file {path} coverage {value:.2f}% is below the {minimum:.2f}% floor",
+            )
     return failures
 
 
+_HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@")
+
+
+def parse_changed_lines(diff: str) -> dict[str, set[int]]:
+    """Return added/modified line numbers by path from a zero-context git diff."""
+    changed: dict[str, set[int]] = {}
+    path: str | None = None
+    for line in diff.splitlines():
+        if line.startswith("+++ "):
+            marker = line[4:]
+            path = marker[2:] if marker.startswith("b/") else None
+            continue
+        match = _HUNK.match(line)
+        if path is None or match is None:
+            continue
+        start = int(match.group(1))
+        count = int(match.group(2) or "1")
+        changed.setdefault(path, set()).update(range(start, start + count))
+    return changed
+
+
+def changed_line_coverage(
+    report: dict[str, Any], changed: dict[str, set[int]],
+) -> tuple[int, int, float]:
+    """Return covered, executable, and percentage for changed measured lines."""
+    covered = 0
+    executable = 0
+    files = report.get("files", {})
+    if not isinstance(files, dict):
+        return 0, 0, 100.0
+    for path, lines in changed.items():
+        entry = files.get(path)
+        if not isinstance(entry, dict):
+            continue
+        executed = set(entry.get("executed_lines", []))
+        missing = set(entry.get("missing_lines", []))
+        measured = lines & (executed | missing)
+        executable += len(measured)
+        covered += len(measured & executed)
+    percentage = covered / executable * 100 if executable else 100.0
+    return covered, executable, percentage
+
+
+def validate_changed_coverage(
+    report: dict[str, Any], changed: dict[str, set[int]], *, minimum: float,
+) -> list[str]:
+    """Return a failure when executable changed lines miss their coverage floor."""
+    covered, executable, percentage = changed_line_coverage(report, changed)
+    if percentage < minimum:
+        return [
+            f"changed-line coverage {percentage:.2f}% ({covered}/{executable}) "
+            f"is below the {minimum:.2f}% floor",
+        ]
+    return []
+
+
+def _file_floor(value: str) -> tuple[str, float]:
+    try:
+        path, minimum = value.rsplit("=", 1)
+        if not path:
+            raise ValueError
+        return path, float(minimum)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("expected PATH=PERCENT") from exc
+
+
 def _parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("report", type=Path, help="coverage.py JSON report")
     parser.add_argument("--min-line", type=float, required=True)
     parser.add_argument("--min-branch", type=float, required=True)
+    parser.add_argument(
+        "--min-file",
+        action="append",
+        default=[],
+        type=_file_floor,
+        metavar="PATH=PERCENT",
+        help="minimum statement coverage for a critical file (repeatable)",
+    )
+    parser.add_argument("--min-changed", type=float)
+    parser.add_argument(
+        "--base-ref",
+        help="git base commit/ref used to calculate changed executable lines",
+    )
     return parser
 
 
@@ -52,7 +152,34 @@ def main() -> int:
         report,
         min_line=args.min_line,
         min_branch=args.min_branch,
+        file_floors=dict(args.min_file),
     )
+    changed_result: tuple[int, int, float] | None = None
+    if args.min_changed is not None:
+        if not args.base_ref:
+            failures.append("changed-line coverage requires --base-ref")
+        else:
+            try:
+                subprocess.run(
+                    ["git", "rev-parse", "--verify", f"{args.base_ref}^{{commit}}"],
+                    check=True,
+                    capture_output=True,
+                    text=True,
+                )
+                diff = subprocess.run(
+                    ["git", "diff", "--unified=0", args.base_ref, "--", "*.py"],
+                    check=True,
+                    capture_output=True,
+                    text=True,
+                ).stdout
+            except subprocess.CalledProcessError:
+                failures.append(f"cannot calculate changed-line coverage from {args.base_ref!r}")
+            else:
+                changed = parse_changed_lines(diff)
+                changed_result = changed_line_coverage(report, changed)
+                failures.extend(
+                    validate_changed_coverage(report, changed, minimum=args.min_changed),
+                )
     if failures:
         for failure in failures:
             print(f"coverage gate failed: {failure}")
@@ -64,6 +191,11 @@ def main() -> int:
         f"line={totals['percent_statements_covered']:.2f}% "
         f"branch={totals['percent_branches_covered']:.2f}%",
     )
+    if changed_result is not None:
+        covered, executable, percentage = changed_result
+        print(
+            f"changed-line gate passed: {percentage:.2f}% ({covered}/{executable})",
+        )
     return 0