6krrt/scripts/verify_commit.py

#!/usr/bin/env python3
"""
verify_commit.py — check a commit for validity, cleanliness, and test health.

Design A checks::

    commit   — the given SHA is a valid, non-empty commit.
    clean    — working tree is clean (or only untracked).
    tests    — pytest passes on files touched by the commit.
    lint     — ruff lint on touched files (optional, off by default).

Tests and lint operate on the committed tree at the given SHA
(``git archive``), not on the working tree.

Output is one line per check in the format ``PASS|FAIL|SKIP <check>: <detail>``.
Total output is at most 40 lines; on FAIL the detail section is at most 20 lines.

Usage::

    python3 scripts/verify_commit.py [options] SHA

Flags::

    --repo DIR          Git repository root (default: current directory).
    --base REF          Base ref to diff against (default: SHA^).
    --full              Run the full pytest suite instead of selective.
    --require-clean     Fail if tracked files are modified.
    --no-lint           Skip lint checks entirely.
    --python PATH       Python interpreter (default: sys.executable).
    --ruff-cmd PATH     Ruff command (default: uvx ruff@0.16.9).
"""

import argparse
import io
import os
import re
import shlex
import subprocess
import sys
import tarfile
import tempfile

# ---------------------------------------------------------------------------
# Argument parsing
# ---------------------------------------------------------------------------

def parse_args(argv: "list[str] | None" = None) -> argparse.Namespace:
    """Parse command-line arguments."""
    p = argparse.ArgumentParser(description="Verify a commit")
    p.add_argument("--repo", default=".",
                   help="Git repository root (default: .)")
    p.add_argument("--base", default=None,
                   help="Base ref to diff against (default: SHA^)")
    p.add_argument("--full", action="store_true",
                   help="Run the full pytest suite instead of selective")
    p.add_argument("--require-clean", action="store_true",
                   help="Fail if tracked files are modified")
    p.add_argument("--no-lint", action="store_true",
                   help="Skip lint checks")
    p.add_argument("--python", default=sys.executable,
                   help="Python interpreter (default: sys.executable)")
    p.add_argument("--ruff-cmd", default="uvx ruff@0.16.9",
                   help="Ruff command (default: uvx ruff@0.16.9)")
    p.add_argument("sha", help="Commit SHA to verify")
    return p.parse_args(argv)


# ---------------------------------------------------------------------------
# Pure function: select_tests
# ---------------------------------------------------------------------------

def select_tests(
    changed: "list[str]",
    test_files: "list[str]",
    read: "callable[[str], str]",
) -> "list[str]":
    """Select test files affected by changed source files.

    Rules (applied to each path in *changed*):

    1. A changed test file (``tests/test_*.py``) selects itself.
    2. A changed source module (``src/<mod>.py``) selects every test file
       that references ``import <mod>``, ``from src import <mod>``, or
       ``from src.<mod> import …``.

    The *read* callable receives an absolute or relative path and returns
    the file content as a string.
    """
    selected: list[str] = []
    seen: set[str] = set()

    for path in changed:
        # --- Direct test file match -------------------------------------------
        if path.startswith("tests/") and path.endswith(".py"):
            if path not in seen:
                selected.append(path)
                seen.add(path)
            continue

        # --- Source module: src/<mod>.py ------------------------------------
        if path.startswith("src/") and path.endswith(".py"):
            mod = path[4:-3]  # "src/foo.py" -> "foo"
            if not mod:
                continue  # guard against "src/.py"
            _select_tests_for_module(mod, test_files, read, selected, seen)

    return selected


def _select_tests_for_module(
    mod: str,
    test_files: "list[str]",
    read: "callable[[str], str]",
    selected: "list[str]",
    seen: "set[str]",
) -> None:
    """Append test files that import *mod* to *selected*."""
    import_bare = re.compile(rf"\bimport\s+{re.escape(mod)}\b")
    from_src = re.compile(rf"\bfrom\s+src\s+import\s+{re.escape(mod)}\b")
    from_src_dot = re.compile(rf"\bfrom\s+src\.{re.escape(mod)}\b")

    for tf in test_files:
        if tf in seen:
            continue
        try:
            content = read(tf)
        except OSError:
            continue
        if import_bare.search(content) or from_src.search(content) or from_src_dot.search(content):
            selected.append(tf)
            seen.add(tf)


# ---------------------------------------------------------------------------
# Git helpers
# ---------------------------------------------------------------------------

def _git(*args: str, repo: str = ".", capture: bool = True) -> "tuple[int, str, str]":
    """Run a git command and return ``(returncode, stdout, stderr)``."""
    cmd: list[str] = ["git", "-C", repo] + list(args)
    try:
        r = subprocess.run(
            cmd,
            capture_output=capture,
            text=True,
            timeout=30,
            check=False,
        )
        return r.returncode, r.stdout or "", r.stderr or ""
    except subprocess.TimeoutExpired:
        return -1, "", "timeout"
    except FileNotFoundError:
        return -2, "", "git not found"


# ---------------------------------------------------------------------------
# Archive helpers
# ---------------------------------------------------------------------------

def _extract_tree(repo: str, sha: str) -> "tempfile.TemporaryDirectory | None":
    """Extract the committed tree at *sha* into a temporary directory.

    Returns a ``TemporaryDirectory`` (access its ``.name``) or ``None``
    on failure. Caller is responsible for cleanup.
    """
    try:
        r = subprocess.run(
            ["git", "-C", repo, "archive", "--format=tar", sha],
            capture_output=True,
            timeout=30,
            check=False,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return None
    if r.returncode != 0 or not r.stdout:
        return None

    tmp = tempfile.TemporaryDirectory(prefix="verify_commit_")
    with tarfile.open(fileobj=io.BytesIO(r.stdout), mode="r:") as tar:
        tar.extractall(path=tmp.name)
    return tmp


# ---------------------------------------------------------------------------
# Per-check functions
# ---------------------------------------------------------------------------

def check_commit(repo: str, sha: str) -> "tuple[str, str]":
    """Return ``(status, detail)`` for the *commit* check, resolving *sha*."""
    rc, out, _err = _git("rev-parse", "--verify", f"{sha}^{{commit}}", repo=repo)
    if rc != 0:
        return "FAIL", f"error: Cannot resolve commit '{sha}'"
    resolved = out.strip()
    if not resolved:
        return "FAIL", f"error: Empty commit at '{sha}'"
    return "PASS", resolved[:12]


def check_clean(repo: str, require_clean: bool) -> "tuple[str, str]":
    """Return ``(status, detail)`` for the *clean* check."""
    rc, out, _err = _git("status", "--porcelain", repo=repo)
    if rc != 0:
        return "FAIL", "git status failed"

    lines = [l for l in out.splitlines() if l.strip()]
    if not lines:
        return "PASS", "Clean working tree"

    if not require_clean:
        n = len(lines)
        return "PASS", f"{n} dirty file(s) (--require-clean not set)"

    modified = [l for l in lines if not l.startswith("??")]
    untracked = [l for l in lines if l.startswith("??")]

    if not modified:
        return "PASS", f"Only untracked file(s) ({len(untracked)})"

    names = [m[3:] for m in modified[:5]]
    suffix = f" … and {len(modified) - 5} more" if len(modified) > 5 else ""
    return "FAIL", f"Modified tracked file(s): {', '.join(names)}{suffix}"


def check_known_failures() -> "tuple[str, str]":
    """Return ``(status, detail)`` for known-failures registration.

    The file may contain only comment lines (``#``) — that is equivalent
    to "no known failures" and returns PASS.
    """
    script_dir = os.path.dirname(os.path.abspath(__file__))
    kf_path = os.path.join(script_dir, "verify_known_failures.txt")
    if not os.path.exists(kf_path):
        return "SKIP", "verify_known_failures.txt missing"
    with open(kf_path, encoding="utf-8") as f:
        raw = f.read()
    entries = [
        l for l in raw.splitlines()
        if l.strip() and not l.strip().startswith("#")
    ]
    if not entries:
        return "PASS", "No known failures registered"
    return "SKIP", f"{len(entries)} known failure(s) registered"


# ---------------------------------------------------------------------------
# Lint helpers
# ---------------------------------------------------------------------------

def _strip_coords(line: str) -> str:
    """Strip line/col coords from a ruff line (e.g. 'file.py:10:5: F401 ...' -> 'file.py: F401 ...')."""
    parts = line.split(":", 3)
    if len(parts) == 4:
        return f"{parts[0]}: {parts[3].strip()}"
    return line.strip()


def new_findings(before: list[str], after: list[str]) -> list[str]:
    """Return findings present in 'after' but not in 'before', ignoring line/col shifts."""
    before_set = { _strip_coords(l) for l in before if l.strip() }
    after_set = { _strip_coords(l) for l in after if l.strip() }

    diff = after_set - before_set
    return sorted(diff)


# ---------------------------------------------------------------------------
# Lint
# ---------------------------------------------------------------------------

def check_lint(
    changed: "list[str]",
    ruff_cmd: str,
    repo: str,
    base: str,
    sha: str,
) -> "tuple[str, list[str]]":
    """Return ``(status, detail_lines)`` for lint.

    Lints only touched Python files. Reports FAIL if new findings are introduced
    compared to the base ref, ignoring line/column shifts. Both baseline and
    current findings are read from the committed tree via ``git show``.
    """
    if not changed:
        return "PASS", ["Nothing to lint"]

    py_files = [f for f in changed if f.endswith(".py")]
    if not py_files:
        return "PASS", ["No Python files to lint"]

    all_new_findings: list[str] = []

    # Finding pattern: <path>:<line>:<col>: <CODE> <message>
    # Example: src/foo.py:10:5: F401 `os` imported but unused
    finding_pattern = re.compile(r"^[^:\n]+:\d+:\d+: [A-Z\d]+ .+$")

    def filter_findings(lines: list[str]) -> list[str]:
        return [l for l in lines if l.strip() and finding_pattern.match(l.strip())]

    for path in py_files:
        # 1. Baseline findings: git show <base>:<path> | ruff check --stdin-filename <path> -
        before_findings = []
        try:
            rc_base, out_base, _err_base = _git("show", f"{base}:{path}", repo=repo)
            if rc_base == 0 and out_base:
                lint_cmd = shlex.split(ruff_cmd) + ["check", "--output-format=concise", "--stdin-filename", path, "-"]
                r_before = subprocess.run(
                    lint_cmd,
                    input=out_base,
                    capture_output=True,
                    text=True,
                    timeout=15,
                    check=False
                )
                if r_before.returncode == 127:
                    return "SKIP", ["ruff command not found (127), skipping lint"]
                before_findings = filter_findings((r_before.stdout or "").splitlines())
        except (FileNotFoundError, subprocess.TimeoutExpired):
            pass

        # 2. Current findings: git show <sha>:<path> | ruff check --stdin-filename <path> -
        rc_curr, out_curr, _err_curr = _git("show", f"{sha}:{path}", repo=repo)
        if rc_curr != 0 or not out_curr:
            # File not present at SHA (e.g. deleted); no current findings
            current_findings: list[str] = []
        else:
            lint_cmd = shlex.split(ruff_cmd) + ["check", "--output-format=concise", "--stdin-filename", path, "-"]
            try:
                r_curr = subprocess.run(
                    lint_cmd,
                    input=out_curr,
                    capture_output=True,
                    text=True,
                    timeout=15,
                    check=False
                )
                if r_curr.returncode == 127:
                    return "SKIP", ["ruff command not found (127), skipping lint"]
                current_findings = filter_findings((r_curr.stdout or "").splitlines())
            except FileNotFoundError:
                return "SKIP", ["ruff command not found, skipping lint"]
            except subprocess.TimeoutExpired:
                return "SKIP", ["lint timed out, skipping"]

        # 3. Compare
        new = new_findings(before_findings, current_findings)
        if new:
            all_new_findings.extend([f"New in {path}: {f}" for f in new])

    if not all_new_findings:
        return "PASS", ["No new lint findings"]

    return "FAIL", all_new_findings


# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------

def get_changed_files(repo: str, base: str, sha: str) -> "list[str]":
    """Return list of file paths changed between *base* and *sha*.

    Falls back to ``ls-tree`` when *base* does not exist -- useful for a
    repo with a single (root) commit.
    """
    rc, out, _err = _git("diff", "--name-only", base, sha, repo=repo)
    if rc == 0 and out.strip():
        return [l for l in out.splitlines() if l.strip()]
    # base ref may not exist (root commit); list all tracked files at sha
    rc, out, _err = _git("ls-tree", "-r", "--name-only", sha, repo=repo)
    if rc == 0 and out.strip():
        return [l for l in out.splitlines() if l.strip()]
    return []


def get_test_files(repo: str, sha: str) -> "list[str]":
    """Return list of tracked test file paths at *sha*."""
    rc, out, _err = _git("ls-tree", "-r", "--name-only", sha, "tests/", repo=repo)
    if rc != 0:
        return []
    return [l for l in out.splitlines() if l.strip() and l.endswith(".py")]


def _make_file_reader(repo: str) -> "callable[[str], str]":
    """Return a *read* callable suitable for ``select_tests``."""

    def _read(path: str) -> str:
        full = os.path.join(repo, path)
        with open(full, encoding="utf-8") as f:
            return f.read()

    return _read


def run_tests_for_sha(
    repo: str,
    sha: str,
    base: str,
    python: str,
    full: bool,
) -> "tuple[str, list[str]]":
    """Run pytest on the committed tree at *sha*.

    Extracts the tree to a temporary directory via ``git archive``, selects
    tests based on changed files between *base* and *sha*, and runs pytest
    inside the extracted tree.
    """
    tmp_obj = _extract_tree(repo, sha)
    if tmp_obj is None:
        return "SKIP", ["Could not extract archive for", sha[:12]]

    tmp = tmp_obj.name
    changed = get_changed_files(repo, base, sha)
    all_tests = get_test_files(repo, sha)
    if not all_tests:
        return "SKIP", ["No tracked test files"]

    if full:
        selected = all_tests
    else:
        read = _make_file_reader(tmp)
        selected = select_tests(changed, all_tests, read)
        if not selected:
            return "SKIP", ["No tests selected by changed files"]

    return _run_pytest(selected, tmp, python)


def _run_pytest(
    test_files: "list[str]",
    repo: str,
    python: str,
) -> "tuple[str, list[str]]":
    """Run pytest on *test_files* and return ``(status, detail_lines)``."""
    test_paths = [os.path.join(repo, t) for t in test_files]
    cmd: list[str] = [
        python, "-m", "pytest",
        "-q", "--tb=short", "--no-header",
    ] + test_paths
    env = dict(os.environ)
    existing = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = (
        f"{repo}:{os.path.join(repo, 'src')}"
        f"{':' + existing if existing else ''}"
    )
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=180,
                           env=env, check=False)
    except subprocess.TimeoutExpired:
        return "FAIL", ["Tests timed out (180s)"]

    output = (r.stdout or "") + (r.stderr or "")
    lines = [l for l in output.splitlines() if l.strip()]

    if r.returncode == 0:
        summary = lines[-1] if lines else "All tests passed"
        return "PASS", [summary]
    
    summary_lines = [l for l in lines if l.startswith(("FAILED ", "ERROR "))]
    count_line = next((l for l in reversed(lines) if "failed" in l.lower() and "passed" in l.lower()), "")
    if count_line and count_line not in summary_lines:
        summary_lines.append(count_line)
    
    if not summary_lines:
        return "FAIL", lines[-20:]

    # Cap at 20 total lines.
    if len(summary_lines) > 20:
        if count_line and count_line == summary_lines[-1]:
            n_truncated = len(summary_lines) - 20
            summary_lines = summary_lines[:18] + [f"... and {n_truncated} more"]
        else:
            n_truncated = len(summary_lines) - 19
            summary_lines = summary_lines[:19] + [f"... and {n_truncated} more"]

    return "FAIL", summary_lines


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main(argv: "list[str] | None" = None) -> int:
    """Entry point. Returns 0 on success, 1 on any FAIL, 2 on usage error."""
    args = parse_args(argv)
    repo = os.path.abspath(args.repo)
    sha = args.sha

    # Resolve SHA — the *commit* check handles this, and we use it early
    # to fail fast on unresolvable SHAs (exit 2, usage error).
    rc, out, _err = _git("rev-parse", "--verify", f"{sha}^{{commit}}", repo=repo)
    if rc != 0:
        detail = _err[:120].strip() if _err.strip() else "unknown revision"
        print(f"FAIL commit: error: Cannot resolve commit '{sha}': {detail}")
        parser = argparse.ArgumentParser(description="Verify a commit")
        parser.print_usage()
        return 2
    resolved = out.strip()
    if not resolved:
        print(f"FAIL commit: error: Empty commit at '{sha}'")
        argparse.ArgumentParser(description="Verify a commit").print_usage()
        return 2

    base = args.base if args.base is not None else f"{resolved}^"

    results: list[tuple[str, str, str | list[str]]] = []

    # 1. known_failures ----------------------------------------------------
    status, detail = check_known_failures()
    results.append(("known_failures", status, detail))

    # 2. commit -----------------------------------------------------------
    status, detail = check_commit(repo, sha)
    results.append(("commit", status, detail))

    # 3. clean -------------------------------------------------------------
    status, detail = check_clean(repo, args.require_clean)
    results.append(("clean", status, detail))

    # 4. lint (unless --no-lint) -------------------------------------------
    if not args.no_lint:
        changed = get_changed_files(repo, base, resolved)
        status, detail_lines = check_lint(changed, args.ruff_cmd, repo, base, resolved)
        results.append(("lint", status, detail_lines))

    # 5. tests -------------------------------------------------------------
    status, detail_lines = run_tests_for_sha(repo, resolved, base, args.python, args.full)
    results.append(("tests", status, detail_lines))

    # ── Render output ─────────────────────────────────────────────────────
    lines_out: list[str] = []
    for check, status, detail in results:
        if isinstance(detail, list):
            if len(detail) <= 1:
                d = detail[0] if detail else ""
                lines_out.append(f"{status} {check}: {d}")
            else:
                lines_out.append(f"{status} {check}:")
                for d in detail[:20]:
                    lines_out.append(f"  {d}")
        else:
            lines_out.append(f"{status} {check}: {detail}")

    # Enforce 40-line output budget
    for line in lines_out[:40]:
        print(line)

    any_fail = any(s == "FAIL" for _, s, _ in results)
    return 1 if any_fail else 0


if __name__ == "__main__":
    sys.exit(main())