"""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 dataclasses import dataclass from pathlib import Path from typing import Any @dataclass(frozen=True) class ChangedModule: """One live production module changed between the base and head commits.""" status: str base_path: str | None head_path: str @dataclass(frozen=True) class ModuleCoverage: """Unrounded line and branch percentages for one measured module.""" line: float branch: float | None 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") if not isinstance(totals, dict): return ["coverage report is missing the totals object"] failures: list[str] = [] metrics = ( ("line", "percent_statements_covered", min_line), ("branch", "percent_branches_covered", min_branch), ) for label, key, minimum in metrics: value = totals.get(key) if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) ): failures.append(f"coverage report is missing numeric {key}") elif value < minimum: 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 parse_changed_modules( raw: bytes, *, roots: tuple[str, ...] = ("obliteratus",), ) -> list[ChangedModule]: """Parse ``git diff --name-status -M -z`` for live Python modules.""" tokens = raw.decode("utf-8", "surrogateescape").split("\0") if tokens and tokens[-1] == "": tokens.pop() normalized_roots = tuple(root.rstrip("/") for root in roots if root.rstrip("/")) def included(path: str) -> bool: return path.endswith(".py") and any( path.startswith(f"{root}/") for root in normalized_roots ) changes: list[ChangedModule] = [] index = 0 while index < len(tokens): status = tokens[index] index += 1 if not status: raise ValueError("changed-module diff contains an empty status") kind = status[0] if kind in {"R", "C"}: if index + 1 >= len(tokens): raise ValueError(f"changed-module diff is truncated after {status!r}") old_path, new_path = tokens[index], tokens[index + 1] index += 2 if kind == "R" and included(old_path) and not included(new_path): changes.append(ChangedModule("D", old_path, old_path)) continue if not included(new_path): continue base_path = old_path if kind == "R" and included(old_path) else None changes.append(ChangedModule(status, base_path, new_path)) continue if index >= len(tokens): raise ValueError(f"changed-module diff is truncated after {status!r}") path = tokens[index] index += 1 if not included(path): continue if kind == "D": changes.append(ChangedModule(status, path, path)) continue if kind not in {"A", "M", "T"}: raise ValueError(f"unsupported changed-module status {status!r}") changes.append( ChangedModule(status, None if kind == "A" else path, path), ) return changes def module_coverage( report: dict[str, Any], path: str, ) -> tuple[ModuleCoverage | None, str | None]: """Return exact per-module line/branch coverage or a validation failure.""" files = report.get("files") entry = files.get(path) if isinstance(files, dict) else None if not isinstance(entry, dict): return None, f"coverage report is missing touched module {path}" summary = entry.get("summary") if not isinstance(summary, dict): return None, f"coverage report is missing summary for touched module {path}" values: dict[str, int] = {} for key in ("covered_lines", "num_statements", "covered_branches", "num_branches"): value = summary.get(key) if isinstance(value, bool) or not isinstance(value, int) or value < 0: return None, f"coverage report has invalid {key} for touched module {path}" values[key] = value if values["num_statements"] == 0: return None, f"coverage report measures no statements for touched module {path}" if values["covered_lines"] > values["num_statements"]: return None, f"coverage report overcounts covered lines for touched module {path}" if values["covered_branches"] > values["num_branches"]: return None, f"coverage report overcounts covered branches for touched module {path}" line = values["covered_lines"] / values["num_statements"] * 100 branch = ( values["covered_branches"] / values["num_branches"] * 100 if values["num_branches"] else None ) return ModuleCoverage(line=line, branch=branch), None def validate_touched_module_regression( head_report: dict[str, Any], base_report: dict[str, Any], changes: list[ChangedModule], *, line_tolerance: float = 0.0, branch_tolerance: float = 0.0, new_module_min_line: float = 80.0, new_module_min_branch: float = 75.0, ) -> list[str]: """Reject line or branch regressions in each touched production module.""" def regressed(head: float, base: float, tolerance: float) -> bool: return base - head > tolerance + 1e-9 failures: list[str] = [] for change in changes: if change.status.startswith("D"): failures.append( f"deleted production module {change.base_path} requires an explicit " "reviewed coverage-policy exception", ) continue head, head_failure = module_coverage(head_report, change.head_path) if head_failure: failures.append(head_failure) continue assert head is not None if change.base_path is None: if head.line < new_module_min_line: failures.append( f"new module {change.head_path} line coverage {head.line:.2f}% " f"is below the {new_module_min_line:.2f}% floor", ) if head.branch is not None and head.branch < new_module_min_branch: failures.append( f"new module {change.head_path} branch coverage {head.branch:.2f}% " f"is below the {new_module_min_branch:.2f}% floor", ) continue base, base_failure = module_coverage(base_report, change.base_path) if base_failure: failures.append(f"base {base_failure}") continue assert base is not None if regressed(head.line, base.line, line_tolerance): failures.append( f"touched module {change.head_path} line coverage regressed " f"from {base.line:.2f}% to {head.line:.2f}%", ) if base.branch is None and head.branch is not None: if head.branch < new_module_min_branch: failures.append( f"touched module {change.head_path} added branches at " f"{head.branch:.2f}% coverage, below the " f"{new_module_min_branch:.2f}% floor", ) elif base.branch is not None and head.branch is not None: if regressed(head.branch, base.branch, branch_tolerance): failures.append( f"touched module {change.head_path} branch coverage regressed " f"from {base.branch:.2f}% to {head.branch:.2f}%", ) return failures 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", ) parser.add_argument( "--base-report", type=Path, help="coverage.py JSON report generated from the base commit", ) parser.add_argument( "--touched-module-no-regression", action="store_true", help="compare each touched production module against --base-report", ) parser.add_argument( "--module-root", action="append", default=[], help="production module root for touched-module comparison (default: obliteratus)", ) parser.add_argument("--module-line-tolerance", type=float, default=0.0) parser.add_argument("--module-branch-tolerance", type=float, default=0.0) parser.add_argument("--new-module-min-line", type=float, default=80.0) parser.add_argument("--new-module-min-branch", type=float, default=75.0) return parser def main() -> int: args = _parser().parse_args() report = json.loads(args.report.read_text(encoding="utf-8")) failures = validate_coverage( 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), ) touched_count: int | None = None if args.touched_module_no_regression: if not args.base_ref or args.base_report is None: failures.append( "touched-module comparison requires --base-ref and --base-report", ) elif args.module_line_tolerance < 0 or args.module_branch_tolerance < 0: failures.append("touched-module tolerances must be non-negative") else: try: base_report = json.loads(args.base_report.read_text(encoding="utf-8")) pathspecs = [ f":(glob){root.rstrip('/')}/**/*.py" for root in (args.module_root or ["obliteratus"]) ] raw_changes = subprocess.run( [ "git", "diff", "--name-status", "-M", "-z", args.base_ref, "HEAD", "--", *pathspecs, ], check=True, capture_output=True, ).stdout module_changes = parse_changed_modules( raw_changes, roots=tuple(args.module_root or ["obliteratus"]), ) except (OSError, json.JSONDecodeError, subprocess.CalledProcessError, ValueError) as exc: failures.append(f"cannot compare touched-module coverage: {exc}") else: touched_count = len(module_changes) failures.extend( validate_touched_module_regression( report, base_report, module_changes, line_tolerance=args.module_line_tolerance, branch_tolerance=args.module_branch_tolerance, new_module_min_line=args.new_module_min_line, new_module_min_branch=args.new_module_min_branch, ), ) if failures: for failure in failures: print(f"coverage gate failed: {failure}") return 1 totals = report["totals"] print( "coverage gate passed: " 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})", ) if touched_count is not None: print(f"touched-module coverage gate passed: {touched_count} module(s)") return 0 if __name__ == "__main__": raise SystemExit(main())