OBLITERATUS/scripts/check_coverage_thresholds.py

"""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")
    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 _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


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),
                )
    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})",
        )
    return 0


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