129 lines
4.3 KiB
Python
129 lines
4.3 KiB
Python
"""Reject Ruff findings introduced since a stable release tag."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
|
|
def _run(command: list[str], *, cwd: Path, accepted: tuple[int, ...] = (0,)) -> subprocess.CompletedProcess[str]:
|
|
result = subprocess.run(command, cwd=cwd, text=True, capture_output=True, check=False)
|
|
if result.returncode not in accepted:
|
|
raise RuntimeError(
|
|
f"command failed ({result.returncode}): {' '.join(command)}\n{result.stderr}"
|
|
)
|
|
return result
|
|
|
|
|
|
def changed_python_paths(repo: Path, base: str) -> list[str]:
|
|
result = _run(
|
|
[
|
|
"git",
|
|
"diff",
|
|
"--name-only",
|
|
"-z",
|
|
"--diff-filter=ACMR",
|
|
f"{base}...HEAD",
|
|
"--",
|
|
"*.py",
|
|
],
|
|
cwd=repo,
|
|
)
|
|
return sorted(path for path in result.stdout.split("\0") if path)
|
|
|
|
|
|
def _source_line(root: Path, relative_path: str, row: int) -> str:
|
|
lines = (root / relative_path).read_text(encoding="utf-8").splitlines()
|
|
return lines[row - 1].strip() if 0 < row <= len(lines) else ""
|
|
|
|
|
|
def normalized_findings(
|
|
diagnostics: list[dict[str, object]], root: Path
|
|
) -> Counter[tuple[str, str, str, str]]:
|
|
findings: Counter[tuple[str, str, str, str]] = Counter()
|
|
for diagnostic in diagnostics:
|
|
filename = Path(str(diagnostic["filename"])).resolve()
|
|
relative_path = filename.relative_to(root.resolve()).as_posix()
|
|
location = diagnostic["location"]
|
|
if not isinstance(location, dict):
|
|
raise TypeError("Ruff diagnostic location must be an object")
|
|
row = int(location["row"])
|
|
finding = (
|
|
relative_path,
|
|
str(diagnostic["code"]),
|
|
str(diagnostic["message"]),
|
|
_source_line(root, relative_path, row),
|
|
)
|
|
findings[finding] += 1
|
|
return findings
|
|
|
|
|
|
def _ruff_diagnostics(root: Path, paths: list[str]) -> list[dict[str, object]]:
|
|
if not paths:
|
|
return []
|
|
ruff = shutil.which("ruff")
|
|
if ruff is None:
|
|
raise RuntimeError("ruff executable is not available")
|
|
result = _run(
|
|
[ruff, "check", *paths, "--output-format=json"],
|
|
cwd=root,
|
|
accepted=(0, 1),
|
|
)
|
|
payload = json.loads(result.stdout)
|
|
if not isinstance(payload, list):
|
|
raise TypeError("Ruff JSON output must be a list")
|
|
return payload
|
|
|
|
|
|
def introduced_findings(repo: Path, base: str) -> tuple[list[str], Counter[tuple[str, str, str, str]]]:
|
|
paths = changed_python_paths(repo, base)
|
|
if not paths:
|
|
return paths, Counter()
|
|
|
|
current = normalized_findings(_ruff_diagnostics(repo, paths), repo)
|
|
with tempfile.TemporaryDirectory(prefix="portfoliohub-ruff-base-") as parent:
|
|
baseline_root = Path(parent) / "tree"
|
|
_run(["git", "worktree", "add", "--detach", str(baseline_root), base], cwd=repo)
|
|
try:
|
|
baseline_paths = [path for path in paths if (baseline_root / path).is_file()]
|
|
baseline = normalized_findings(
|
|
_ruff_diagnostics(baseline_root, baseline_paths), baseline_root
|
|
)
|
|
finally:
|
|
_run(
|
|
["git", "worktree", "remove", "--force", str(baseline_root)],
|
|
cwd=repo,
|
|
)
|
|
return paths, current - baseline
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base", required=True, help="stable comparison tag")
|
|
args = parser.parse_args()
|
|
|
|
repo = Path(
|
|
_run(["git", "rev-parse", "--show-toplevel"], cwd=Path.cwd()).stdout.strip()
|
|
)
|
|
ruff_version = _run([shutil.which("ruff") or "ruff", "--version"], cwd=repo).stdout.strip()
|
|
paths, introduced = introduced_findings(repo, args.base)
|
|
print(f"base={args.base} head=HEAD selector=ACMR-python files={len(paths)} tool={ruff_version}")
|
|
for path in paths:
|
|
print(f"selected={path}")
|
|
if introduced:
|
|
for (path, code, message, source), count in sorted(introduced.items()):
|
|
print(f"NEW {count}x {path}: {code} {message} | {source}")
|
|
print(f"introduced_findings={sum(introduced.values())}")
|
|
return 1
|
|
print("introduced_findings=0")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|