mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
446 lines
16 KiB
Python
446 lines
16 KiB
Python
"""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())
|