diff --git a/.github/workflows/mutation-testing.yml b/.github/workflows/mutation-testing.yml deleted file mode 100644 index 51425e19..00000000 --- a/.github/workflows/mutation-testing.yml +++ /dev/null @@ -1,26 +0,0 @@ -name: Mutation testing - -# Thin caller of the shared mutation-testing reusable workflow; caller -# guide: leynos/shared-actions docs/mutation-mutmut-workflow.md. -on: - schedule: - - cron: "50 12 * * *" - workflow_dispatch: - -permissions: {} - -concurrency: - group: mutation-testing-${{ github.ref }} - cancel-in-progress: false - -jobs: - mutation: - permissions: - contents: read - id-token: write # OIDC workflow-source resolution - uses: leynos/shared-actions/.github/workflows/mutation-mutmut.yml@e9f155f1613acdc3dae4ea99da3a2ce8adff7597 - with: - # Mutable Python source lives in hooks/, not the src/ default. - paths: "hooks/" - # Flat layout: changed-file paths map to module globs unaltered. - module-prefix-strip: "" diff --git a/.gitignore b/.gitignore index dd374c94..a2c9a132 100644 --- a/.gitignore +++ b/.gitignore @@ -201,10 +201,6 @@ cython_debug/ # memtrace local knowledge-graph database .memdb/ - -# mutmut mutation-testing working copy and results -mutants/ - # Untracked cache of the estate-wide en-GB-oxendict dictionary. The generated # typos.toml remains tracked and is guarded by a drift test. .typos-oxendict-base.json diff --git a/Makefile b/Makefile index de83e997..c27356d7 100644 --- a/Makefile +++ b/Makefile @@ -27,17 +27,15 @@ PYTHON_SCRIPTS := $(sort $(wildcard hooks/*.py scripts/*.py tests/*.py)) PYTEST := uv run --group dev python -m pytest TYPOS_VERSION ?= 1.48.0 TYPOS := uv tool run typos@$(TYPOS_VERSION) -HOOK_TESTS := $(sort $(wildcard hooks/test_*.py)) REPO_TESTS := $(sort $(wildcard tests/test_*.py)) ENTRYPOINT_TESTS := $(filter tests/test_rust_entrypoints.py,$(REPO_TESTS)) -TEST_TARGETS := $(HOOK_TESTS) $(REPO_TESTS) +TEST_TARGETS := $(REPO_TESTS) # Test targets: -# - test-hooks: post-turn hook behaviour and git-state decisions. # - test-entrypoints: rust-entrypoint process tests using cuprum and cmd-mox. # - test: full pytest suite for all repository tests. # - ci: complete CI/CD gate sequence used by GitHub Actions. -.PHONY: all clean check-fmt fmt lint typecheck syntax-check shell-syntax-check check-home-phase-boundary spelling test-hooks test-entrypoints test ci +.PHONY: all clean check-fmt fmt lint typecheck syntax-check shell-syntax-check check-home-phase-boundary spelling test-entrypoints test ci all: ci @@ -77,8 +75,5 @@ spelling: test: @$(PYTEST) $(TEST_TARGETS) -v -test-hooks: - @$(PYTEST) $(HOOK_TESTS) -v - test-entrypoints: @$(PYTEST) $(ENTRYPOINT_TESTS) -v diff --git a/README.md b/README.md index 4eae0507..fa8a08ba 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,10 @@ RUST_ENTRYPOINT_PHASE=system bash rust-entrypoint RUST_ENTRYPOINT_PHASE=home bash rust-entrypoint ``` +The post-turn quality stop hook is no longer provided here. See + for installation, +configuration, and support. + ## Shared spelling dictionary `data/typos-oxendict-base.toml` is the shared en-GB-oxendict dictionary for the diff --git a/docs/developers-guide.md b/docs/developers-guide.md index d736ec9a..cea3ba41 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -224,6 +224,9 @@ distinction visible when adding new bootstrap behaviour. - Reuses the managed helper checkout path when `HELPER_TOOLS_REPO_DIR` is exported. - Fetches the requested helper branch before copying hook files. +- Copies repository hook files into `~/.claude/hooks`; it no longer registers + any hook in Claude Code settings. The post-turn quality stop hook moved to + its own project: https://github.com/leynos/post-turn-quality-stop-hook. ### `install-skills` @@ -387,9 +390,6 @@ The later Dakar audit on 15 July 2026 added the `polymer` stem from four correct - `make check-home-phase-boundary` - Rejects APT, `sudo`, and linker mutation patterns in home-phase scripts. - Scans non-comment lines only. -- `make test-hooks` - - Runs the hook-only pytest subset (`HOOK_TESTS`) via - `uv run python -m pytest`. - `make test-entrypoints` - Runs the entrypoint-only pytest subset (`ENTRYPOINT_TESTS`) via `uv run python -m pytest`. diff --git a/docs/users-guide.md b/docs/users-guide.md index 34b1047d..7a60e729 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -49,6 +49,12 @@ shared libraries required by the user tools. See the [migration guide](migration-guide.md) when moving from the previous single-phase `rust-entrypoint` bootstrap to the system/home phase split. +### Post-turn quality stop hook removed + +This repository no longer provides the post-turn quality stop hook. It has +moved to its own project. For installation, configuration, and support see +. + ## CodeScene skills Use [`codescene-cli`](../skills/codescene-cli/SKILL.md) to run local CodeScene diff --git a/hooks/post-turn-quality-stop-hook.py b/hooks/post-turn-quality-stop-hook.py deleted file mode 100644 index d94669ec..00000000 --- a/hooks/post-turn-quality-stop-hook.py +++ /dev/null @@ -1,1188 +0,0 @@ -#!/usr/bin/env python3 -""" -Claude Code Stop-hook quality gate. - -At "turn end" (Claude Code Stop hook): -1) Ensure refs/remotes/origin/main exists (git fetch only if missing by default) -2) Compute changed files vs origin/main using merge-base(origin/main, HEAD) -3) If changes exist: - - If any Python/TypeScript files changed: run `make check-fmt lint typecheck` (only targets that exist) - - If any Rust files changed: run `make check-fmt lint` (only targets that exist) - - If any Markdown files changed: run `make markdownlint` (only targets that exist) -4) If any invoked command fails, BLOCK the stop with a detailed reason. - -Behaviour knobs (env vars): -- POST_TURN_ALWAYS_FETCH=1 -> always `git fetch origin main` (otherwise only if origin/main missing) -- POST_TURN_BASE_REF=... -> override base ref (default: origin/main) -- POST_TURN_MAX_OUTPUT_CHARS -> truncate per-command output (default: 12000) -- POST_TURN_COMPUSH=1 -> after successful checks, BLOCK if uncommitted/untracked changes - remain, or if local commits are ahead of upstream, - and remind the agent to commit and/or push - -Claude Code contract: -- Reads JSON hook input from stdin (but works even if stdin isn't JSON) -- On failure: prints JSON {"decision":"block","reason":"..."} to stdout and exits 0 -- On success: prints nothing and exits 0 - -Examples --------- -Run the hook manually with a default environment: - - POST_TURN_ALWAYS_FETCH=1 python3 ~/.claude/hooks/post-turn-quality-stop-hook.py < /dev/null -""" - -from __future__ import annotations - -import json -import os -import re -import shutil -import subprocess -import sys -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - - -PY_TS_EXTS = {".py", ".pyi", ".ts", ".tsx", ".mts", ".cts"} -RUST_EXTS = {".rs"} -MD_EXTS = {".md", ".mdx", ".markdown"} - -CATS_TO_TARGETS: dict[str, list[str]] = { - "python_ts": ["check-fmt", "lint", "typecheck"], - "rust": ["check-fmt", "lint"], - "markdown": ["markdownlint"], -} -CODE_CATS = {"python_ts", "rust"} -MD_CATS = {"markdown"} -TRUTHY_VALUES = {"1", "true", "yes"} - - -def default_categories() -> dict[str, bool]: - """Return a default category mapping. - - Returns - ------- - dict[str, bool] - Default mapping of category names to enabled flags. - """ - return {"python_ts": False, "rust": False, "markdown": False} - - -@dataclass -class HookState: - """Execution state for the stop hook. - - Attributes - ---------- - ok - Whether the hook checks succeeded. - base_ref - Base ref used for comparison. - base_commit - Resolved merge-base commit. - changed_files - Files changed relative to the base commit. - categories - Detected change categories. - make_targets_requested - Make targets requested based on change categories. - make_targets_run - Make targets executed. - make_targets_skipped - Requested targets that were not present in the Makefile. - commands - Executed commands and their outputs. - fetched - Whether a fetch was performed. - error - Error message when blocking. - """ - ok: bool = True - base_ref: str = "origin/main" - base_commit: str | None = None - changed_files: list[str] = field(default_factory=list) - categories: dict[str, bool] = field(default_factory=default_categories) - make_targets_requested: list[str] = field(default_factory=list) - make_targets_run: list[str] = field(default_factory=list) - make_targets_skipped: list[str] = field(default_factory=list) - commands: list[dict[str, Any]] = field(default_factory=list) - fetched: bool = False - error: str | None = None - - -@dataclass -class RunStopChecksPreparation: - """Prepared state for ``run_stop_checks``. - - Attributes - ---------- - ok - Whether preparation succeeded and execution should continue. - exit_code - Exit code to return immediately when preparation did not succeed. - state - Hook state populated during preparation. - repo - Resolved repository root when preparation succeeded. - """ - - ok: bool - exit_code: int - state: HookState - repo: Path | None = None - - -def run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess[str]: - """Run a subprocess command in the given working directory. - - Parameters - ---------- - cmd - Command and arguments to run. - cwd - Working directory for the subprocess. - - Returns - ------- - subprocess.CompletedProcess[str] - Completed process with captured output. - """ - try: - return subprocess.run( # noqa: S603 # valid: command and args are controlled (no shell, no user-supplied command strings). - cmd, cwd=str(cwd), text=True, capture_output=True, check=False - ) - except FileNotFoundError as exc: - if Path(exc.filename or "") != cwd: - raise - return subprocess.CompletedProcess( - args=cmd, returncode=1, stdout="", stderr=str(exc) - ) - except NotADirectoryError as exc: - return subprocess.CompletedProcess( - args=cmd, returncode=1, stdout="", stderr=str(exc) - ) - - -def truncate(text: str, max_chars: int) -> str: - """Truncate text to a maximum length. - - Parameters - ---------- - text - Text to truncate. - max_chars - Maximum number of characters to keep. - - Returns - ------- - str - Truncated text with a placeholder if needed. - """ - if max_chars <= 0: - return "" - if len(text) <= max_chars: - return text - marker = "\n... (output truncated) ...\n" - if max_chars <= len(marker): - return text[:max_chars] - remaining = max_chars - len(marker) - head = remaining // 2 - tail = remaining - head - return text[:head] + marker + text[-tail:] - - -def repo_root(start_cwd: Path) -> tuple[Path | None, str | None]: - """Resolve the git repository root for a starting directory. - - Parameters - ---------- - start_cwd - Directory to resolve from. - - Returns - ------- - tuple[Path | None, str | None] - Repository root path and error message, if any. - """ - p = run(["git", "rev-parse", "--show-toplevel"], start_cwd) - if p.returncode != 0: - err = (p.stderr.strip() or p.stdout.strip() or "not a git repository") - return None, err - root = p.stdout.strip() - if not root: - return None, "git rev-parse --show-toplevel returned empty output" - return Path(root), None - - -def ensure_origin_remote(repo: Path) -> tuple[bool, str | None]: - """Ensure the origin remote is configured. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - tuple[bool, str | None] - ok and error message, if any. - """ - remotes = run(["git", "remote"], repo) - if remotes.returncode != 0: - return False, f"git remote failed: {remotes.stderr.strip() or remotes.stdout.strip()}" - if "origin" not in remotes.stdout.split(): - return False, "git remote 'origin' not found" - return True, None - - -def fetch_origin_main(repo: Path) -> tuple[bool, str | None]: - """Fetch origin/main. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - tuple[bool, str | None] - ok and error message, if any. - """ - fetch = run(["git", "fetch", "--quiet", "origin", "main"], repo) - if fetch.returncode != 0: - return False, f"git fetch origin main failed: {fetch.stderr.strip() or fetch.stdout.strip()}" - return True, None - - -def ref_exists(repo: Path, ref: str) -> tuple[bool, str | None]: - """Check whether a ref exists. - - Parameters - ---------- - repo - Repository root path. - ref - Fully qualified ref name. - - Returns - ------- - tuple[bool, str | None] - True if the ref exists, otherwise False and an error if the check failed. - """ - verify = run(["git", "show-ref", "--verify", "--quiet", ref], repo) - if verify.returncode == 0: - return True, None - if verify.returncode == 1: - return False, None - return False, verify.stderr.strip() or verify.stdout.strip() or "git show-ref failed" - - -def verify_ref(repo: Path, ref: str) -> tuple[bool, str | None]: - """Verify that a ref can be resolved. - - Parameters - ---------- - repo - Repository root path. - ref - Ref name to verify with rev-parse. - - Returns - ------- - tuple[bool, str | None] - ok and error message, if any. - """ - rp = run(["git", "rev-parse", "--verify", "--quiet", ref], repo) - if rp.returncode != 0: - return False, f"Cannot resolve {ref}" - return True, None - - -def ensure_origin_main(repo: Path, *, always_fetch: bool) -> tuple[bool, str | None, bool]: - """Ensure origin/main is present and resolvable. - - Parameters - ---------- - repo - Repository root path. - always_fetch - If True, always fetch origin/main. - - Returns - ------- - tuple[bool, str | None, bool] - ok, error message (if any), fetched. - """ - ok, err = ensure_origin_remote(repo) - if not ok: - return False, err, False - - ok, err, fetched = ensure_origin_main_ref(repo, always_fetch=always_fetch) - if not ok: - return False, err, fetched - - ok, err = verify_ref(repo, "origin/main") - if not ok: - return False, err, fetched - return True, None, fetched - - -def ensure_origin_main_ref(repo: Path, *, always_fetch: bool) -> tuple[bool, str | None, bool]: - """Ensure refs/remotes/origin/main exists, fetching if needed. - - Parameters - ---------- - repo - Repository root path. - always_fetch - If True, always fetch origin/main. - - Returns - ------- - tuple[bool, str | None, bool] - ok, error message (if any), fetched. - """ - if always_fetch: - ok, err = fetch_origin_main(repo) - if not ok: - return False, err, True - return True, None, True - - exists, err = ref_exists(repo, "refs/remotes/origin/main") - if err: - return False, err, False - if exists: - return True, None, False - - ok, err = fetch_origin_main(repo) - if not ok: - return False, err, True - - exists, err = ref_exists(repo, "refs/remotes/origin/main") - if err: - return False, err, True - if not exists: - return False, "origin/main still missing after fetch", True - - return True, None, True - - -def ensure_base_ref( - repo: Path, - base_ref: str, - *, - always_fetch: bool, -) -> tuple[bool, str | None, bool]: - """Ensure a base ref is available and resolvable. - - Parameters - ---------- - repo - Repository root path. - base_ref - Base git ref used to compute the merge-base. - always_fetch - If True, always fetch origin/main when base_ref is origin/main. - - Returns - ------- - tuple[bool, str | None, bool] - ok, error message (if any), fetched. - """ - if base_ref == "origin/main": - return ensure_origin_main(repo, always_fetch=always_fetch) - - ok, err = verify_ref(repo, base_ref) - if not ok: - return False, err or f"Cannot resolve base ref '{base_ref}'", False - return True, None, False - - -def merge_base(repo: Path, base_ref: str) -> tuple[str | None, str | None]: - """Compute the merge-base of base_ref and HEAD. - - Parameters - ---------- - repo - Repository root path. - base_ref - Base ref to compare against HEAD. - - Returns - ------- - tuple[str | None, str | None] - Merge-base commit hash and error message, if any. - """ - p = run(["git", "merge-base", base_ref, "HEAD"], repo) - if p.returncode != 0: - return None, f"git merge-base {base_ref} HEAD failed: {p.stderr.strip() or p.stdout.strip()}" - base = p.stdout.strip() - if not base: - return None, "git merge-base returned empty output" - return base, None - - -def changed_files(repo: Path, base_commit: str) -> tuple[list[str] | None, str | None]: - """List files changed relative to a base commit. - - Parameters - ---------- - repo - Repository root path. - base_commit - Base commit hash for diffing. - - Returns - ------- - tuple[list[str] | None, str | None] - Sorted list of changed files and error message, if any. - """ - changed: set[str] = set() - - # Tracked changes (unstaged and staged) relative to base_commit - for args in ( - ["git", "diff", "--name-only", base_commit], - ["git", "diff", "--cached", "--name-only", base_commit], - ): - p = run(args, repo) - if p.returncode != 0: - return None, f"{' '.join(args)} failed: {p.stderr.strip() or p.stdout.strip()}" - for line in p.stdout.splitlines(): - line = line.strip() - if line: - changed.add(line) - - # Untracked (but not ignored) - u = run(["git", "ls-files", "--others", "--exclude-standard"], repo) - if u.returncode != 0: - return None, f"git ls-files failed: {u.stderr.strip() or u.stdout.strip()}" - for line in u.stdout.splitlines(): - line = line.strip() - if line: - changed.add(line) - - return sorted(changed), None - - -def has_uncommitted_changes(repo: Path) -> tuple[bool | None, str | None]: - """Check whether the working tree has uncommitted or untracked changes. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - tuple[bool | None, str | None] - True if dirty, False if clean, None on error; and an error message. - """ - for args in ( - ["git", "diff", "--quiet"], - ["git", "diff", "--cached", "--quiet"], - ): - p = run(args, repo) - if p.returncode == 1: - return True, None - if p.returncode != 0: - return None, f"{' '.join(args)} failed: {p.stderr.strip() or p.stdout.strip()}" - - u = run(["git", "ls-files", "--others", "--exclude-standard"], repo) - if u.returncode != 0: - return None, f"git ls-files failed: {u.stderr.strip() or u.stdout.strip()}" - if u.stdout.strip(): - return True, None - - return False, None - - -def get_upstream_ref(repo: Path) -> tuple[str | None, str | None]: - """Get the upstream tracking ref for the current branch. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - tuple[str | None, str | None] - Upstream ref name (e.g. ``origin/main``) and error message, if any. - """ - p = run(["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], repo) - if p.returncode != 0: - return None, p.stderr.strip() or p.stdout.strip() or "no upstream configured" - ref = p.stdout.strip() - if not ref: - return None, "git rev-parse returned empty upstream" - return ref, None - - -def has_unpushed_commits(repo: Path, upstream: str) -> tuple[bool | None, str | None]: - """Check whether ``HEAD`` is ahead of the given upstream ref. - - Parameters - ---------- - repo - Repository root path. - upstream - Upstream tracking ref to compare against. - - Returns - ------- - tuple[bool | None, str | None] - True if local commits are ahead, False if not, None on error; and an error message. - """ - p = run(["git", "rev-list", "--count", f"{upstream}..HEAD"], repo) - if p.returncode != 0: - return None, ( - f"git rev-list --count {upstream}..HEAD failed: " - f"{p.stderr.strip() or p.stdout.strip()}" - ) - - ahead = p.stdout.strip() - if not ahead: - return None, "git rev-list --count returned empty output" - - try: - return int(ahead) > 0, None - except ValueError: - return None, f"git rev-list --count returned non-integer output: {ahead}" - - -def detect_categories(files: list[str]) -> dict[str, bool]: - """Detect change categories from a file list. - - Parameters - ---------- - files - List of file paths. - - Returns - ------- - dict[str, bool] - Mapping of category names to detection flags. - """ - cats = default_categories() - for f in files: - ext = Path(f).suffix.lower() - if ext in PY_TS_EXTS: - cats["python_ts"] = True - if ext in RUST_EXTS: - cats["rust"] = True - if ext in MD_EXTS: - cats["markdown"] = True - return cats - - -def parse_make_targets(make_stdout: str) -> set[str]: - """Parse make -qp output for target names. - - Parameters - ---------- - make_stdout - Stdout from make -qp. - - Returns - ------- - set[str] - Parsed make target names. - """ - targets: set[str] = set() - rule_re = re.compile(r"^([^\s:#=]+(?:\s+[^\s:#=]+)*)\s*::?\s*.*$") - for line in make_stdout.splitlines(): - if not line: - continue - if line.startswith(("#", "\t", " ")): - continue - m = rule_re.match(line) - if not m: - continue - lhs = m.group(1) - for t in lhs.split(): - if "%" in t: - continue - targets.add(t) - return targets - - -def is_missing_makefile(output: str) -> bool: - """Check output for a missing Makefile condition. - - Parameters - ---------- - output - Combined output from make. - - Returns - ------- - bool - True if the output indicates no Makefile was found. - """ - lowered = output.lower() - return "no makefile found" in lowered - - -def get_make_targets(repo: Path) -> tuple[set[str] | None, str | None]: - """Collect available make targets from a repository. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - tuple[set[str] | None, str | None] - Target set and error message, if any. - """ - try: - p = run(["make", "-qp", "--no-print-directory"], repo) - except FileNotFoundError: - return None, "make not found on PATH" - - # make -q can return 0 or 1 without being an error; 2 means failure - if p.returncode == 2: - combined = f"{p.stderr.strip()}\n{p.stdout.strip()}".strip() - if is_missing_makefile(combined): - return set(), None - return None, combined or "make -qp failed" - - return parse_make_targets(p.stdout), None - - -def dedup_preserve_order(items: list[str]) -> list[str]: - """Deduplicate items while preserving order. - - Parameters - ---------- - items - Items to deduplicate. - - Returns - ------- - list[str] - Deduplicated items in original order. - """ - out: list[str] = [] - seen: set[str] = set() - for x in items: - if x in seen: - continue - seen.add(x) - out.append(x) - return out - - -def run_make(repo: Path, kind: str, targets: list[str], max_out: int) -> dict[str, Any]: - """Run make targets and capture output. - - Parameters - ---------- - repo - Repository root path. - kind - Label describing the target group. - targets - Make targets to run. - max_out - Maximum number of output characters to capture. - - Returns - ------- - dict[str, Any] - Execution metadata and captured output. - """ - if not targets: - return {"kind": kind, "cmd": "", "exit_code": 0, "stdout": "", "stderr": ""} - - try: - p = run(["make", "--no-print-directory", *targets], repo) - except FileNotFoundError as exc: - return { - "kind": kind, - "cmd": "make " + " ".join(targets), - "exit_code": 127, - "stdout": "", - "stderr": f"make not found on PATH: {exc}", - } - return { - "kind": kind, - "cmd": "make " + " ".join(targets), - "exit_code": int(p.returncode), - "stdout": truncate(p.stdout, max_out), - "stderr": truncate(p.stderr, max_out), - } - - -def format_reason(state: HookState) -> str: - """Format a blocking reason for hook output. - - Parameters - ---------- - state - Hook execution state. - - Returns - ------- - str - Human-readable reason string. - """ - lines: list[str] = [] - lines.append("Post-turn checks failed.") - - if state.error: - lines.append("") - lines.append(f"Error: {state.error}") - - base_ref = state.base_ref or "?" - base_commit = state.base_commit or "?" - lines.append("") - lines.append(f"Diff base: {base_ref} ({base_commit})") - - changed = state.changed_files - lines.append("") - lines.append(f"Changed files vs {base_ref}: {len(changed)}") - for f in changed[:60]: - lines.append(f"- {f}") - if len(changed) > 60: - lines.append(f"- … (+{len(changed) - 60} more)") - - cats = state.categories - detected: list[str] = [] - if cats.get("python_ts"): - detected.append("Python/TypeScript") - if cats.get("rust"): - detected.append("Rust") - if cats.get("markdown"): - detected.append("Markdown") - if detected: - lines.append("") - lines.append("Detected change types: " + ", ".join(detected)) - - if state.make_targets_requested: - lines.append("") - lines.append("Requested make targets: " + " ".join(state.make_targets_requested)) - if state.make_targets_run: - lines.append("Targets run: " + " ".join(state.make_targets_run)) - if state.make_targets_skipped: - lines.append("Targets skipped (missing): " + " ".join(state.make_targets_skipped)) - - failures = [c for c in state.commands if int(c.get("exit_code", 0)) != 0] - for c in failures: - cmd = c.get("cmd", "") - code = c.get("exit_code", "?") - combined = "\n".join([x for x in [c.get("stdout", ""), c.get("stderr", "")] if x]).strip() - - lines.append("") - lines.append(f"Command failed (exit {code}): {cmd}") - lines.append("```") - lines.append(combined or "(no output captured)") - lines.append("```") - - lines.append("") - lines.append("Fix the failures above. The checks will re-run at the end of the next turn.") - return "\n".join(lines) - - -def block_and_print(state: HookState) -> int: - """Emit a blocking response and return a stop code. - - Parameters - ---------- - state - Hook execution state. - - Returns - ------- - int - Exit code for the hook. - """ - payload = {"decision": "block", "reason": format_reason(state)} - print(json.dumps(payload)) - return 0 - - -def targets_for_categories( - categories: dict[str, bool], - *, - include: set[str] | None = None, -) -> list[str]: - """Expand enabled categories into make targets. - - Parameters - ---------- - categories - Mapping of category flags. - include - Optional subset of categories to include. - - Returns - ------- - list[str] - Deduplicated target list. - """ - requested: list[str] = [] - for category, enabled in categories.items(): - if not enabled: - continue - if include is not None and category not in include: - continue - requested.extend(CATS_TO_TARGETS.get(category, [])) - return dedup_preserve_order(requested) - - -def parse_bool_env(value: str) -> bool: - """Parse a boolean environment value. - - Parameters - ---------- - value - Raw environment value. - - Returns - ------- - bool - True when the value is a recognized truthy token. - """ - return value.strip().lower() in TRUTHY_VALUES - - -def parse_max_output(value: str, default: int = 12000) -> int: - """Parse the max output character limit. - - Parameters - ---------- - value - Raw environment value. - default - Default value to use on parse failure. - - Returns - ------- - int - Parsed maximum output length. - """ - try: - return int(value) - except ValueError: - return default - - -def parse_env() -> tuple[str, bool, int, bool]: - """Parse environment configuration for the hook. - - Returns - ------- - tuple[str, bool, int, bool] - Base ref, always-fetch flag, max output length, and compush flag. - """ - base_ref = os.environ.get("POST_TURN_BASE_REF", "origin/main") - always_fetch = parse_bool_env(os.environ.get("POST_TURN_ALWAYS_FETCH", "")) - max_out = parse_max_output(os.environ.get("POST_TURN_MAX_OUTPUT_CHARS", "12000")) - compush = parse_bool_env(os.environ.get("POST_TURN_COMPUSH", "")) - return base_ref, always_fetch, max_out, compush - - -def parse_hook_input() -> dict[str, Any]: - """Parse hook input from stdin. - - Returns - ------- - dict[str, Any] - Parsed hook input as a dict (empty if missing or invalid). - """ - try: - hook_input = json.load(sys.stdin) - except (json.JSONDecodeError, ValueError): - return {} - match hook_input: - case dict() as data: - return data - case _: - return {} - - -def resolve_start_cwd(hook_input: dict[str, Any]) -> Path: - """Resolve the starting working directory for the hook. - - Parameters - ---------- - hook_input - Parsed hook input. - - Returns - ------- - Path - Working directory for git operations. - """ - match hook_input.get("cwd"): - case str() as cwd_value if cwd_value: - return Path(cwd_value) - case _: - pass - - match os.environ.get("CLAUDE_PROJECT_DIR"): - case str() as cwd_value if cwd_value: - return Path(cwd_value) - case _: - return Path(os.getcwd()) - - -def fail_state(state: HookState, message: str | None) -> int: - """Mark the state as failed and emit a block response. - - Parameters - ---------- - state - Hook execution state. - message - Error message to include in the response. - - Returns - ------- - int - Exit code for the hook. - """ - state.ok = False - state.error = message - return block_and_print(state) - - -def evaluate_changes(state: HookState, repo: Path, max_out: int) -> int: - """Select and execute checks based on detected changes. - - Parameters - ---------- - state - Hook execution state. - repo - Repository root path. - max_out - Maximum number of output characters to capture. - - Returns - ------- - int - Exit code for the hook. - """ - cats = detect_categories(state.changed_files) - state.categories = cats - - requested = targets_for_categories(cats) - state.make_targets_requested = requested - if not requested: - return 0 - - make_targets, make_err = get_make_targets(repo) - if make_targets is None: - return fail_state(state, f"Could not enumerate make targets: {make_err}") - - run_targets = [t for t in requested if t in make_targets] - skip_targets = [t for t in requested if t not in make_targets] - state.make_targets_run = run_targets - state.make_targets_skipped = skip_targets - - commands: list[dict[str, Any]] = [] - code_targets = [ - t for t in targets_for_categories(cats, include=CODE_CATS) if t in make_targets - ] - md_targets = [ - t for t in targets_for_categories(cats, include=MD_CATS) if t in make_targets - ] - - if code_targets: - commands.append(run_make(repo, "code", code_targets, max_out)) - if md_targets: - commands.append(run_make(repo, "markdown", md_targets, max_out)) - - state.commands = commands - - if not commands: - return 0 - - ok_all = all(int(c.get("exit_code", 0)) == 0 for c in commands) - if ok_all: - return 0 - - state.ok = False - return block_and_print(state) - - -def prepare_run_stop_checks( - start_cwd: Path, base_ref: str, *, always_fetch: bool -) -> RunStopChecksPreparation: - """Prepare repository state for ``run_stop_checks``. - - Parameters - ---------- - start_cwd - Working directory for git operations. - base_ref - Base git ref used for comparisons. - always_fetch - Whether to always fetch the base ref. - - Returns - ------- - RunStopChecksPreparation - Structured preparation result containing the populated hook state and - repository root when preparation succeeded. - """ - state = HookState(base_ref=base_ref) - - if shutil.which("git") is None: - return RunStopChecksPreparation( - ok=False, - exit_code=fail_state(state, "git not found on PATH"), - state=state, - ) - - repo, _err = repo_root(start_cwd) - if repo is None: - return RunStopChecksPreparation(ok=False, exit_code=0, state=state) - - ok, err, fetched = ensure_base_ref(repo, base_ref, always_fetch=always_fetch) - state.fetched = fetched - if not ok: - return RunStopChecksPreparation( - ok=False, - exit_code=fail_state(state, err), - state=state, - ) - - base_commit, err = merge_base(repo, base_ref) - if base_commit is None: - return RunStopChecksPreparation( - ok=False, - exit_code=fail_state(state, err), - state=state, - ) - state.base_commit = base_commit - - files, err = changed_files(repo, base_commit) - if files is None: - return RunStopChecksPreparation( - ok=False, - exit_code=fail_state(state, err), - state=state, - ) - - state.changed_files = files - return RunStopChecksPreparation(ok=True, exit_code=0, state=state, repo=repo) - - -def compush_check(repo: Path) -> int: - """Block the stop with commit/push reminders when local work is not published. - - Parameters - ---------- - repo - Repository root path. - - Returns - ------- - int - Exit code for the hook (always 0 per hook contract). - """ - upstream, _err = get_upstream_ref(repo) - upstream_label = upstream or "origin (upstream not configured)" - - dirty, err = has_uncommitted_changes(repo) - if err is not None: - return 0 - if dirty: - payload = { - "decision": "block", - "reason": f"Please commit and push to {upstream_label}", - } - print(json.dumps(payload)) - return 0 - - if upstream is None: - return 0 - - ahead, err = has_unpushed_commits(repo, upstream) - if err is not None or not ahead: - return 0 - - payload = { - "decision": "block", - "reason": f"Please push committed changes to {upstream_label}", - } - print(json.dumps(payload)) - return 0 - - -def run_stop_checks( - start_cwd: Path, - base_ref: str, - *, - always_fetch: bool, - max_out: int, - compush: bool = False, -) -> int: - """Run stop-hook checks for a given working directory. - - Parameters - ---------- - start_cwd - Working directory for git operations. - base_ref - Base git ref used for comparisons. - always_fetch - Whether to always fetch origin/main. - max_out - Maximum number of output characters to capture. - compush - Whether to remind the agent to commit and push when dirty. - - Returns - ------- - int - Exit code for the hook. - """ - preparation = prepare_run_stop_checks( - start_cwd, base_ref, always_fetch=always_fetch - ) - if not preparation.ok: - return preparation.exit_code - - state = preparation.state - repo = preparation.repo - assert repo is not None - - if state.changed_files: - rc = evaluate_changes(state, repo, max_out) - if rc != 0: - return rc - - if compush: - return compush_check(repo) - - return 0 - - -def main() -> int: - """Run the stop-hook checks. - - Returns - ------- - int - Exit code for the hook. - """ - hook_input = parse_hook_input() - start_cwd = resolve_start_cwd(hook_input) - base_ref, always_fetch, max_out, compush = parse_env() - return run_stop_checks( - start_cwd, - base_ref, - always_fetch=always_fetch, - max_out=max_out, - compush=compush, - ) - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/hooks/test_post_turn_quality_stop_hook.py b/hooks/test_post_turn_quality_stop_hook.py deleted file mode 100644 index 226f2299..00000000 --- a/hooks/test_post_turn_quality_stop_hook.py +++ /dev/null @@ -1,477 +0,0 @@ -"""Exercise the post-turn quality stop hook end to end and in pieces. - -This module covers the hook's expected git-state decisions, environment -parsing, subprocess error handling, make-target discovery, and compush -follow-up behaviour. The tests focus on observable inputs and outputs, -including successful checks, soft-failure paths that must stay silent, -and blocking/error conditions that should surface through the hook -contract. - -Run the suite from the repository root with `pytest` or the repository's -test target. No external fixtures or environment setup are required -beyond a Python environment with `pytest` installed because the tests -mock subprocess-heavy interactions. - -Example: - python3 -m pytest hooks/test_post_turn_quality_stop_hook.py -v -""" - -from __future__ import annotations - -import importlib.util -import json -import subprocess -import sys -from pathlib import Path -from types import ModuleType -from unittest.mock import patch - -import pytest - - -def _load_hook_module() -> ModuleType: - """Load the hook script as a module despite the dashes in the filename.""" - hook_path = Path(__file__).parent / "post-turn-quality-stop-hook.py" - # The module name must match the path-derived name mutmut assigns - # ("hooks."), so mutation runs can attribute trampoline - # hits recorded by this suite to the mutants it generates. - spec = importlib.util.spec_from_file_location( - "hooks.post-turn-quality-stop-hook", hook_path - ) - assert spec is not None, "expected import spec to be created for hook module" - assert spec.loader is not None, "expected import spec loader for hook module" - mod = importlib.util.module_from_spec(spec) - sys.modules[spec.name] = mod - spec.loader.exec_module(mod) - return mod - - -hook = _load_hook_module() - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - -def _completed( - returncode: int, stdout: str = "", stderr: str = "" -) -> subprocess.CompletedProcess[str]: - """Build a ``CompletedProcess`` stub with the given return code and output.""" - return subprocess.CompletedProcess( - args=["unit-test"], returncode=returncode, stdout=stdout, stderr=stderr - ) - - -REPO = Path("/fake/repo") - - -# --------------------------------------------------------------------------- -# has_uncommitted_changes -# --------------------------------------------------------------------------- - - -class TestHasUncommittedChanges: - """Tests for has_uncommitted_changes().""" - - def test_clean_working_tree(self) -> None: - """All three checks pass -> False.""" - with patch.object(hook, "run") as mock_run: - mock_run.side_effect = [ - _completed(0), # git diff --quiet - _completed(0), # git diff --cached --quiet - _completed(0, stdout=""), # git ls-files - ] - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is False, f"expected dirty to be False but was {dirty!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_unstaged_changes(self) -> None: - """git diff --quiet exits 1 -> True.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(1) - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is True, f"expected dirty to be True but was {dirty!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_staged_changes(self) -> None: - """git diff --cached --quiet exits 1 -> True.""" - with patch.object(hook, "run") as mock_run: - mock_run.side_effect = [ - _completed(0), # unstaged clean - _completed(1), # staged dirty - ] - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is True, f"expected dirty to be True but was {dirty!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_untracked_files(self) -> None: - """ls-files returns output -> True.""" - with patch.object(hook, "run") as mock_run: - mock_run.side_effect = [ - _completed(0), - _completed(0), - _completed(0, stdout="newfile.py\n"), - ] - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is True, f"expected dirty to be True but was {dirty!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_diff_error(self) -> None: - """Non-0/1 exit from diff -> None + error.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(128, stderr="fatal: bad") - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is None, f"expected dirty to be None on error but was {dirty!r}" - assert err is not None, "expected an error message from git diff failure" - assert "fatal: bad" in err, f"expected fatal error in message but got {err!r}" - - def test_ls_files_error(self) -> None: - """ls-files failure -> None + error.""" - with patch.object(hook, "run") as mock_run: - mock_run.side_effect = [ - _completed(0), - _completed(0), - _completed(128, stderr="fatal: oops"), - ] - dirty, err = hook.has_uncommitted_changes(REPO) - assert dirty is None, f"expected dirty to be None on error but was {dirty!r}" - assert err is not None, "expected an error message from git ls-files failure" - assert "git ls-files failed" in err, ( - f"expected ls-files failure in message but got {err!r}" - ) - - -# --------------------------------------------------------------------------- -# get_upstream_ref -# --------------------------------------------------------------------------- - - -class TestGetUpstreamRef: - """Tests for get_upstream_ref().""" - - def test_returns_upstream(self) -> None: - """Successful rev-parse --abbrev-ref returns the tracking ref.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="origin/main\n") - ref, err = hook.get_upstream_ref(REPO) - assert ref == "origin/main", f"expected upstream ref origin/main but got {ref!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_no_upstream(self) -> None: - """Non-zero rev-parse exit returns None.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(128, stderr="no upstream") - ref, err = hook.get_upstream_ref(REPO) - assert ref is None, f"expected no upstream ref but got {ref!r}" - assert "no upstream" in (err or ""), ( - f"expected no-upstream message but got {err!r}" - ) - - def test_empty_stdout(self) -> None: - """Empty stdout from rev-parse returns None.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="") - ref, err = hook.get_upstream_ref(REPO) - assert ref is None, f"expected no upstream ref but got {ref!r}" - assert err is not None, "expected an error when upstream stdout is empty" - - -# --------------------------------------------------------------------------- -# has_unpushed_commits -# --------------------------------------------------------------------------- - - -class TestHasUnpushedCommits: - """Tests for has_unpushed_commits().""" - - def test_ahead_of_upstream(self) -> None: - """Positive rev-list count returns True.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="2\n") - ahead, err = hook.has_unpushed_commits(REPO, "origin/main") - assert ahead is True, f"expected ahead to be True but was {ahead!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_not_ahead_of_upstream(self) -> None: - """Zero rev-list count returns False.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="0\n") - ahead, err = hook.has_unpushed_commits(REPO, "origin/main") - assert ahead is False, f"expected ahead to be False but was {ahead!r}" - assert err is None, f"expected no error but got {err!r}" - - def test_rev_list_error(self) -> None: - """Non-zero rev-list exit returns None.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(128, stderr="fatal: bad revision") - ahead, err = hook.has_unpushed_commits(REPO, "origin/main") - assert ahead is None, f"expected ahead to be None on error but was {ahead!r}" - assert err is not None, "expected an error message from rev-list failure" - assert "fatal: bad revision" in err, ( - f"expected bad revision error in message but got {err!r}" - ) - - def test_empty_output(self) -> None: - """Empty rev-list output returns None with an error message.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="") - ahead, err = hook.has_unpushed_commits(REPO, "origin/main") - assert ahead is None, f"expected ahead to be None for empty output but was {ahead!r}" - assert "empty output" in (err or ""), ( - f"expected empty output error but got {err!r}" - ) - - def test_non_integer_output(self) -> None: - """Non-integer rev-list output returns None.""" - with patch.object(hook, "run") as mock_run: - mock_run.return_value = _completed(0, stdout="two\n") - ahead, err = hook.has_unpushed_commits(REPO, "origin/main") - assert ahead is None, ( - f"expected ahead to be None for non-integer output but was {ahead!r}" - ) - assert "non-integer" in (err or ""), ( - f"expected non-integer error but got {err!r}" - ) - - -# --------------------------------------------------------------------------- -# compush_check -# --------------------------------------------------------------------------- - - -class TestCompushCheck: - """Tests for compush_check().""" - - def test_dirty_with_upstream(self, capsys: pytest.CaptureFixture[str]) -> None: - """Dirty tree + upstream -> block with push message.""" - with patch.object(hook, "has_uncommitted_changes", return_value=(True, None)), \ - patch.object(hook, "get_upstream_ref", return_value=("origin/feature", None)): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - out = json.loads(capsys.readouterr().out) - assert out["decision"] == "block", ( - f"expected block decision but got {out['decision']!r}" - ) - assert "Please commit and push to origin/feature" in out["reason"], ( - f"expected commit/push reminder but got {out['reason']!r}" - ) - - def test_dirty_no_upstream(self, capsys: pytest.CaptureFixture[str]) -> None: - """Dirty tree + no upstream -> block with fallback text.""" - with patch.object(hook, "has_uncommitted_changes", return_value=(True, None)), \ - patch.object(hook, "get_upstream_ref", return_value=(None, "no upstream")): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - out = json.loads(capsys.readouterr().out) - assert out["decision"] == "block", ( - f"expected block decision but got {out['decision']!r}" - ) - assert "origin (upstream not configured)" in out["reason"], ( - f"expected upstream fallback in reason but got {out['reason']!r}" - ) - - def test_clean_tree(self, capsys: pytest.CaptureFixture[str]) -> None: - """Clean tree -> no output, exit 0.""" - with patch.object(hook, "get_upstream_ref", return_value=("origin/feature", None)), \ - patch.object(hook, "has_uncommitted_changes", return_value=(False, None)), \ - patch.object(hook, "has_unpushed_commits", return_value=(False, None)): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - assert capsys.readouterr().out == "", "expected no hook output for clean tree" - - def test_error_checking_changes(self, capsys: pytest.CaptureFixture[str]) -> None: - """Error from has_uncommitted_changes -> silent exit 0.""" - with patch.object(hook, "has_uncommitted_changes", return_value=(None, "oops")): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - assert capsys.readouterr().out == "", ( - "expected no hook output when change check errors are suppressed" - ) - - def test_clean_tree_with_unpushed_commits( - self, capsys: pytest.CaptureFixture[str] - ) -> None: - """Clean tree + ahead of upstream -> block with push-only message.""" - with patch.object(hook, "get_upstream_ref", return_value=("origin/feature", None)), \ - patch.object(hook, "has_uncommitted_changes", return_value=(False, None)), \ - patch.object(hook, "has_unpushed_commits", return_value=(True, None)): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - out = json.loads(capsys.readouterr().out) - assert out["decision"] == "block", ( - f"expected block decision but got {out['decision']!r}" - ) - assert "Please push committed changes to origin/feature" in out["reason"], ( - f"expected push reminder but got {out['reason']!r}" - ) - - def test_clean_tree_no_upstream(self, capsys: pytest.CaptureFixture[str]) -> None: - """Clean tree + no upstream -> no ahead check and no output.""" - with patch.object(hook, "get_upstream_ref", return_value=(None, "no upstream")), \ - patch.object(hook, "has_uncommitted_changes", return_value=(False, None)), \ - patch.object(hook, "has_unpushed_commits") as mock_ahead: - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - mock_ahead.assert_not_called() - assert capsys.readouterr().out == "", ( - "expected no hook output when upstream is unavailable" - ) - - def test_error_checking_unpushed_commits( - self, capsys: pytest.CaptureFixture[str] - ) -> None: - """Ahead check errors stay silent to preserve hook contract.""" - with patch.object(hook, "get_upstream_ref", return_value=("origin/feature", None)), \ - patch.object(hook, "has_uncommitted_changes", return_value=(False, None)), \ - patch.object(hook, "has_unpushed_commits", return_value=(None, "oops")): - rc = hook.compush_check(REPO) - assert rc == 0, f"expected compush_check rc 0 but got {rc!r}" - assert capsys.readouterr().out == "", ( - "expected no hook output when ahead check errors are suppressed" - ) - - -# --------------------------------------------------------------------------- -# parse_env - compush flag -# --------------------------------------------------------------------------- - - -class TestParseEnvCompush: - """Tests for the compush flag in parse_env().""" - - def test_compush_set(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("POST_TURN_COMPUSH", "1") - _base, _fetch, _max, compush = hook.parse_env() - assert compush is True, f"expected compush to be True but was {compush!r}" - - def test_compush_unset(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("POST_TURN_COMPUSH", raising=False) - _base, _fetch, _max, compush = hook.parse_env() - assert compush is False, f"expected compush to be False but was {compush!r}" - - def test_compush_truthy_alias(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("POST_TURN_COMPUSH", "yes") - _base, _fetch, _max, compush = hook.parse_env() - assert compush is True, f"expected compush to be True but was {compush!r}" - - -# --------------------------------------------------------------------------- -# run_stop_checks - compush integration -# --------------------------------------------------------------------------- - - -class TestRunStopChecksCompush: - """Integration-level tests for compush in run_stop_checks().""" - - def test_compush_triggers_after_success(self) -> None: - """compush=True + quality pass + dirty -> compush_check called.""" - with patch.object(hook, "repo_root", return_value=(REPO, None)), \ - patch.object(hook, "ensure_base_ref", return_value=(True, None, False)), \ - patch.object(hook, "merge_base", return_value=("abc123", None)), \ - patch.object(hook, "changed_files", return_value=(["src/foo.py"], None)), \ - patch.object(hook, "evaluate_changes", return_value=0), \ - patch.object(hook, "compush_check", return_value=0) as mock_compush, \ - patch("shutil.which", return_value="/usr/bin/git"): - rc = hook.run_stop_checks( - REPO, "origin/main", always_fetch=False, max_out=12000, compush=True - ) - assert rc == 0, f"expected run_stop_checks rc 0 but got {rc!r}" - mock_compush.assert_called_once_with(REPO) - - def test_compush_skipped_when_disabled(self) -> None: - """compush=False -> compush_check not called.""" - with patch.object(hook, "repo_root", return_value=(REPO, None)), \ - patch.object(hook, "ensure_base_ref", return_value=(True, None, False)), \ - patch.object(hook, "merge_base", return_value=("abc123", None)), \ - patch.object(hook, "changed_files", return_value=(["src/foo.py"], None)), \ - patch.object(hook, "evaluate_changes", return_value=0), \ - patch.object(hook, "compush_check") as mock_compush, \ - patch("shutil.which", return_value="/usr/bin/git"): - hook.run_stop_checks( - REPO, "origin/main", always_fetch=False, max_out=12000, compush=False - ) - mock_compush.assert_not_called() - - def test_compush_skipped_on_quality_failure(self) -> None: - """Quality check failure (nonzero rc) -> compush_check not called.""" - with patch.object(hook, "repo_root", return_value=(REPO, None)), \ - patch.object(hook, "ensure_base_ref", return_value=(True, None, False)), \ - patch.object(hook, "merge_base", return_value=("abc123", None)), \ - patch.object(hook, "changed_files", return_value=(["src/foo.py"], None)), \ - patch.object(hook, "evaluate_changes", return_value=1), \ - patch.object(hook, "compush_check") as mock_compush, \ - patch("shutil.which", return_value="/usr/bin/git"): - hook.run_stop_checks( - REPO, "origin/main", always_fetch=False, max_out=12000, compush=True - ) - mock_compush.assert_not_called() - - def test_compush_runs_when_no_files_changed(self) -> None: - """compush=True still runs when there are no changed files to lint.""" - with patch.object(hook, "repo_root", return_value=(REPO, None)), \ - patch.object(hook, "ensure_base_ref", return_value=(True, None, False)), \ - patch.object(hook, "merge_base", return_value=("abc123", None)), \ - patch.object(hook, "changed_files", return_value=([], None)), \ - patch.object(hook, "evaluate_changes") as mock_evaluate, \ - patch.object(hook, "compush_check", return_value=0) as mock_compush, \ - patch("shutil.which", return_value="/usr/bin/git"): - rc = hook.run_stop_checks( - REPO, "origin/main", always_fetch=False, max_out=12000, compush=True - ) - assert rc == 0, f"expected run_stop_checks rc 0 but got {rc!r}" - mock_evaluate.assert_not_called() - mock_compush.assert_called_once_with(REPO) - -# --------------------------------------------------------------------------- -# run() - OSError resilience -# --------------------------------------------------------------------------- - - -class TestRunOSError: - """Tests for run() handling of OSError (e.g. missing cwd).""" - - def test_nonexistent_cwd_returns_error(self) -> None: - """run() with a nonexistent cwd returns rc=1 instead of raising.""" - missing_path = Path("/nonexistent/path") - result = hook.run(["git", "status"], missing_path) - assert result.returncode == 1, ( - f"expected returncode 1 for missing cwd but got {result.returncode!r}" - ) - assert result.stderr, "expected stderr to describe missing cwd" - assert str(missing_path) in result.stderr, ( - f"expected missing path in stderr but got {result.stderr!r}" - ) - - def test_run_stop_checks_nonexistent_cwd(self, capsys: pytest.CaptureFixture[str]) -> None: - """Full pipeline exits cleanly when start_cwd does not exist.""" - with patch("shutil.which", return_value="/usr/bin/git"): - rc = hook.run_stop_checks( - Path("/nonexistent/path"), - "origin/main", - always_fetch=False, - max_out=12000, - ) - assert rc == 0, f"expected run_stop_checks rc 0 but got {rc!r}" - assert capsys.readouterr().out == "", ( - "expected no hook output when start_cwd does not exist" - ) - - -class TestGetMakeTargets: - """Tests for make target enumeration.""" - - def test_missing_make_returns_error(self) -> None: - """Missing `make` surfaces as an enumeration error.""" - with patch.object( - hook, - "run", - side_effect=FileNotFoundError(2, "No such file or directory", "make"), - ): - targets, err = hook.get_make_targets(REPO) - assert targets is None, ( - f"expected no make targets when make is missing but got {targets!r}" - ) - assert err == "make not found on PATH", ( - f"expected make-not-found error but got {err!r}" - ) diff --git a/install-hooks b/install-hooks index e5525491..d28d60db 100755 --- a/install-hooks +++ b/install-hooks @@ -19,8 +19,6 @@ INSTALL_BIN_DIR="${HOME}/.local/bin" INSTALL_HOOK_CMD="${INSTALL_BIN_DIR}/install-hook-cmd" CLAUDE_HOOKS_DIR="${HOME}/.claude/hooks" -# Allow slow checks in large repos or on constrained machines. -STOP_HOOK_TIMEOUT=600 if ! command -v git >/dev/null 2>&1; then echo "Error: git is required but was not found on PATH" >&2 @@ -79,14 +77,6 @@ if [[ "${installed_hooks}" -eq 0 ]]; then echo "Warning: no hooks found in ${HOOKS_SRC}" >&2 fi -stop_hook="${CLAUDE_HOOKS_DIR}/post-turn-quality-stop-hook.py" -if [[ ! -f "${stop_hook}" ]]; then - echo "Error: Stop hook not found at ${stop_hook}" >&2 - exit 1 -fi - -"${INSTALL_HOOK_CMD}" Stop --timeout "${STOP_HOOK_TIMEOUT}" python3 "${stop_hook}" - echo "Hooks installed and registered." echo "Hook command installed to: ${INSTALL_HOOK_CMD}" echo "Hooks directory: ${CLAUDE_HOOKS_DIR}" diff --git a/pyproject.toml b/pyproject.toml index e220e954..51449380 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,17 +3,6 @@ name = "agent-helper-scripts" version = "0.0.0" requires-python = ">=3.13" -[tool.mutmut] -# The only importable Python product code is the Stop-hook script in -# hooks/; its co-located pytest module loads it in-process via importlib -# (Path(__file__)-relative), so mutants are attributable inside mutmut's -# mutants/ sandbox. The tests/ suite exercises shell scripts and -# repository files through root-relative paths the sandbox does not -# copy, so it stays outside the mutation test selection. -source_paths = ["hooks/"] -do_not_mutate = ["hooks/test_*.py"] -pytest_add_cli_args_test_selection = ["hooks/"] - [dependency-groups] dev = [ "cmd-mox", diff --git a/tests/test_workflow_contract.py b/tests/test_workflow_contract.py deleted file mode 100644 index 251dcc2a..00000000 --- a/tests/test_workflow_contract.py +++ /dev/null @@ -1,143 +0,0 @@ -"""Contract tests for the mutation-testing caller workflow. - -The executable logic lives in the ``leynos/shared-actions`` reusable -workflow, which carries its own unit and integration tests; this -repository's caller is declarative configuration. These tests parse the -caller with PyYAML and assert the contract it must uphold: that it -references the correct reusable workflow, pinned to a commit SHA (not a -mutable branch or tag), with the expected permissions, triggers, and -inputs. Dependabot owns the pinned SHA value itself, so these tests do -not assert which SHA is pinned — only that one is. Drift (repointing the -pin at a branch, widening permissions, or losing the hooks/ -configuration) fails CI on the pull request rather than surfacing in a -scheduled or manual run. - -Run via the repository ``make test`` target or directly with pytest. -""" - -from __future__ import annotations - -import re -from pathlib import Path - -import pytest -import yaml - -WORKFLOW_PATH = ( - Path(__file__).resolve().parents[1] - / ".github" - / "workflows" - / "mutation-testing.yml" -) - -pytestmark = pytest.mark.skipif( - not WORKFLOW_PATH.exists(), - reason="workflow file not present in this working copy (e.g. " - "inside mutmut's mutants/ sandbox, which does not copy .github/)", -) - -#: Matches the mutmut reusable workflow pinned to a full 40-character -#: lowercase-hex commit SHA. Dependabot owns the SHA value; this test -#: only asserts the shape of the pin, not which commit it points at. -USES_RE = re.compile( - r"^leynos/shared-actions/\.github/workflows/mutation-mutmut\.yml@" - r"[0-9a-f]{40}$" -) - -EXPECTED_WITH = { - "paths": "hooks/", - "module-prefix-strip": "", -} - - -def _load() -> dict[str, object]: - """Parse the workflow file.""" - return yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")) - - -def _triggers(workflow: dict[str, object]) -> dict[str, object]: - """Return the ``on:`` mapping (PyYAML parses the bare key as True).""" - triggers = workflow.get("on", workflow.get(True)) - assert isinstance(triggers, dict), "the workflow must declare an on: mapping" - return triggers - - -def _mutation_job(workflow: dict[str, object]) -> dict[str, object]: - """Return the single calling job.""" - jobs = workflow.get("jobs") - assert isinstance(jobs, dict), "the workflow must declare a jobs mapping" - assert list(jobs) == ["mutation"], ( - f"expected a single job named 'mutation', found {sorted(jobs or {})}" - ) - return jobs["mutation"] - - -def test_uses_reference_is_pinned_to_a_commit_sha() -> None: - """The job must call the correct shared workflow, pinned to a commit SHA. - - The exact SHA is not asserted: Dependabot owns bumping it, and a test - that pinned the value would fail every Dependabot bump PR in lockstep. - """ - uses = _mutation_job(_load()).get("uses") - assert uses is not None, "jobs.mutation.uses is missing" - assert USES_RE.match(uses), ( - f"jobs.mutation.uses must reference mutation-mutmut.yml pinned to a " - f"full 40-character lowercase-hex commit SHA (not a branch or tag), " - f"got {uses!r}" - ) - - -def test_job_permissions_are_exactly_least_privilege() -> None: - """The job grants contents: read and id-token: write, nothing broader.""" - permissions = _mutation_job(_load()).get("permissions") - assert permissions == {"contents": "read", "id-token": "write"}, ( - "jobs.mutation.permissions must be exactly " - f"{{'contents': 'read', 'id-token': 'write'}}, got {permissions!r}" - ) - - -def test_workflow_default_permissions_are_empty() -> None: - """The workflow-level default token scope is empty.""" - workflow = _load() - assert workflow.get("permissions") == {}, ( - f"top-level permissions must be an empty mapping, got " - f"{workflow.get('permissions')!r}" - ) - - -def test_concurrency_serializes_per_ref_without_cancelling() -> None: - """Runs queue per ref instead of cancelling one another.""" - concurrency = _load().get("concurrency") - assert isinstance(concurrency, dict), "the workflow must declare concurrency" - assert concurrency.get("group") == "mutation-testing-${{ github.ref }}", ( - f"concurrency.group must key on the triggering ref, got " - f"{concurrency.get('group')!r}" - ) - assert concurrency.get("cancel-in-progress") is False, ( - f"concurrency.cancel-in-progress must be false, got " - f"{concurrency.get('cancel-in-progress')!r}" - ) - - -def test_triggers_keep_schedule_and_plain_dispatch() -> None: - """The daily schedule stays; dispatch declares no inputs.""" - triggers = _triggers(_load()) - schedule = triggers.get("schedule") - assert schedule == [{"cron": "50 12 * * *"}], ( - f"on.schedule must be the daily 12:50 UTC cron, got {schedule!r}" - ) - assert "workflow_dispatch" in triggers, "on.workflow_dispatch is missing" - dispatch = triggers.get("workflow_dispatch") or {} - inputs = dispatch.get("inputs") or {} - assert not inputs, ( - f"on.workflow_dispatch must declare no inputs, got {sorted(inputs)}" - ) - - -def test_with_block_carries_the_caller_configuration() -> None: - """The caller passes exactly the hooks/ flat-layout configuration.""" - with_block = _mutation_job(_load()).get("with") - assert with_block == EXPECTED_WITH, ( - f"jobs.mutation.with must be exactly {EXPECTED_WITH!r}, " - f"got {with_block!r}" - )