From aec96eefd319b5cb9476d28e7693275dfdb2c0d4 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 2 Nov 2026 12:17:36 +0100 Subject: [PATCH] Add todo_hooks closer, drift sweep, PR import Python 3.14 stdlib package wiring git to the todo MCP server: - closer (post-commit): runs the done_check of each Todo: trailer task and close_task's it only on exit 0. It FAILS OPEN: the commit already happened, so a down server, bad token or no gh prints one warning and exits 0 instead of failing the hook. - sweep: nightly drift report (systemd units shipped, not enabled). - import-pr: a merged PR's '## Todo:' items become tasks, deduped. - A done_check runs only if its sha256 is in ~/.config/todo-hooks/trusted-checks; 'Check: manual' notes are never auto-closed. - Hooks exposed in .pre-commit-hooks.yaml (post-commit, post-merge, post-rewrite). Co-Authored-By: Claude Opus 5.5 --- .pre-commit-hooks.yaml | 27 +++ scripts/todo_hooks.sh | 16 ++ todo_hooks/.coveragerc | 14 ++ todo_hooks/README.md | 43 ++++ todo_hooks/__init__.py | 1 + todo_hooks/__main__.py | 45 ++++ todo_hooks/checkrun.py | 62 ++++++ todo_hooks/client.py | 189 ++++++++++++++++ todo_hooks/closer.py | 76 +++++++ todo_hooks/context.py | 61 +++++ todo_hooks/gh.py | 69 ++++++ todo_hooks/gitrepo.py | 88 ++++++++ todo_hooks/import_pr.py | 135 +++++++++++ todo_hooks/notes.py | 83 +++++++ todo_hooks/paths.py | 43 ++++ todo_hooks/pr_parse.py | 43 ++++ todo_hooks/pyproject.toml | 177 +++++++++++++++ todo_hooks/sweep.py | 159 +++++++++++++ todo_hooks/systemd/todo-drift.service | 9 + todo_hooks/systemd/todo-drift.timer | 9 + todo_hooks/tests/__init__.py | 1 + todo_hooks/tests/_closer_helpers.py | 36 +++ todo_hooks/tests/_fakes.py | 239 ++++++++++++++++++++ todo_hooks/tests/_pr_helpers.py | 25 +++ todo_hooks/tests/conftest.py | 99 +++++++++ todo_hooks/tests/test_client.py | 198 +++++++++++++++++ todo_hooks/tests/test_closer.py | 175 +++++++++++++++ todo_hooks/tests/test_closer_failures.py | 101 +++++++++ todo_hooks/tests/test_gitrepo_checkrun.py | 142 ++++++++++++ todo_hooks/tests/test_import_pr.py | 128 +++++++++++ todo_hooks/tests/test_import_pr_commits.py | 162 ++++++++++++++ todo_hooks/tests/test_main.py | 173 +++++++++++++++ todo_hooks/tests/test_notes_trust.py | 127 +++++++++++ todo_hooks/tests/test_pr_parse_gh.py | 73 ++++++ todo_hooks/tests/test_sweep.py | 247 +++++++++++++++++++++ todo_hooks/trust.py | 46 ++++ todo_hooks/trust_cmd.py | 60 +++++ 37 files changed, 3381 insertions(+) create mode 100644 .pre-commit-hooks.yaml create mode 100755 scripts/todo_hooks.sh create mode 100644 todo_hooks/.coveragerc create mode 100644 todo_hooks/README.md create mode 100644 todo_hooks/__init__.py create mode 100644 todo_hooks/__main__.py create mode 100644 todo_hooks/checkrun.py create mode 100644 todo_hooks/client.py create mode 100644 todo_hooks/closer.py create mode 100644 todo_hooks/context.py create mode 100644 todo_hooks/gh.py create mode 100644 todo_hooks/gitrepo.py create mode 100644 todo_hooks/import_pr.py create mode 100644 todo_hooks/notes.py create mode 100644 todo_hooks/paths.py create mode 100644 todo_hooks/pr_parse.py create mode 100644 todo_hooks/pyproject.toml create mode 100644 todo_hooks/sweep.py create mode 100644 todo_hooks/systemd/todo-drift.service create mode 100644 todo_hooks/systemd/todo-drift.timer create mode 100644 todo_hooks/tests/__init__.py create mode 100644 todo_hooks/tests/_closer_helpers.py create mode 100644 todo_hooks/tests/_fakes.py create mode 100644 todo_hooks/tests/_pr_helpers.py create mode 100644 todo_hooks/tests/conftest.py create mode 100644 todo_hooks/tests/test_client.py create mode 100644 todo_hooks/tests/test_closer.py create mode 100644 todo_hooks/tests/test_closer_failures.py create mode 100644 todo_hooks/tests/test_gitrepo_checkrun.py create mode 100644 todo_hooks/tests/test_import_pr.py create mode 100644 todo_hooks/tests/test_import_pr_commits.py create mode 100644 todo_hooks/tests/test_main.py create mode 100644 todo_hooks/tests/test_notes_trust.py create mode 100644 todo_hooks/tests/test_pr_parse_gh.py create mode 100644 todo_hooks/tests/test_sweep.py create mode 100644 todo_hooks/trust.py create mode 100644 todo_hooks/trust_cmd.py diff --git a/.pre-commit-hooks.yaml b/.pre-commit-hooks.yaml new file mode 100644 index 0000000..59a80cb --- /dev/null +++ b/.pre-commit-hooks.yaml @@ -0,0 +1,27 @@ +# Hooks other repos consume with `repo: https://github.com/kuhyx/utils`. +# All three are post-* stages, so each clone must install them (see +# todo_hooks/README.md). They never fail a git operation: the work has +# already happened, so server/gh trouble prints one warning and exits 0. +- id: todo-closer + name: close Todo trailer tasks + entry: scripts/todo_hooks.sh closer + language: script + stages: [post-commit] + always_run: true + pass_filenames: false + +- id: todo-import-pr-merge + name: import PR Todo sections (pull) + entry: scripts/todo_hooks.sh import-pr post-merge + language: script + stages: [post-merge] + always_run: true + pass_filenames: false + +- id: todo-import-pr-rewrite + name: import PR Todo sections (pull --rebase) + entry: scripts/todo_hooks.sh import-pr post-rewrite + language: script + stages: [post-rewrite] + always_run: true + pass_filenames: false diff --git a/scripts/todo_hooks.sh b/scripts/todo_hooks.sh new file mode 100755 index 0000000..44442c2 --- /dev/null +++ b/scripts/todo_hooks.sh @@ -0,0 +1,16 @@ +#!/bin/bash + +# ============================================================================ +# Run one todo_hooks subcommand (closer | sweep | trust | import-pr) from this +# utils checkout. This is the entry the .pre-commit-hooks.yaml hooks and the +# todo-drift systemd unit call, so a consuming repo needs no install step. +# +# Usage: scripts/todo_hooks.sh [args...] +# ============================================================================ + +set -euo pipefail + +UTILS_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +readonly UTILS_ROOT + +PYTHONPATH="$UTILS_ROOT${PYTHONPATH:+:$PYTHONPATH}" exec python3 -m todo_hooks "$@" diff --git a/todo_hooks/.coveragerc b/todo_hooks/.coveragerc new file mode 100644 index 0000000..2e7a90e --- /dev/null +++ b/todo_hooks/.coveragerc @@ -0,0 +1,14 @@ +[run] +branch = True +source = todo_hooks +omit = + todo_hooks/tests/* + +[report] +fail_under = 100 +show_missing = true +exclude_lines = + pragma: no cover + if __name__ == .__main__.: + if TYPE_CHECKING: + \.\.\. diff --git a/todo_hooks/README.md b/todo_hooks/README.md new file mode 100644 index 0000000..f4dff57 --- /dev/null +++ b/todo_hooks/README.md @@ -0,0 +1,43 @@ +# todo_hooks + +Wires git to the todo app's MCP server (stdlib only, Python 3.14). + +| Command | What it does | +|---|---| +| `closer` | post-commit: for each `Todo: ` trailer of HEAD, run the note's `done_check` and `close_task` on exit 0 | +| `sweep [--dry-run] [--report P] [--trust-file P]` | nightly: list open notes whose trusted `done_check` already passes, write `~/.local/state/todo-drift/report.txt` | +| `trust ` / `trust --seed-open` | approve a note's `done_check` (or all currently open ones) | +| `import-pr post-merge \| post-rewrite [rebase] \| --range A..B` | add a task per `- title \| done_check: cmd` item under a merged PR's `## Todo:` heading | + +Run as `scripts/todo_hooks.sh ` or `PYTHONPATH= python3 -m todo_hooks `. + +## Safety + +- A `done_check` runs only if `sha256(done_check.strip())` is in `~/.config/todo-hooks/trusted-checks`. Cloud sessions can add notes, so an untrusted check is never executed. +- The note's `repo` must equal `kuhyx/`. +- A note with a `Check: manual` line is never auto-closed; it is listed as "close by hand". +- Hooks fail open: server down, bad token or no `gh` prints one warning and exits 0. + +## Server + +`TODO_MCP_URL` (default `http://127.0.0.1:8735/mcp`), token from `TODO_MCP_TOKEN` else `~/.config/todo-mcp/client-token` (0600). Handshake: `initialize`, `notifications/initialized`, then `tools/call`, all with the `Mcp-Session-Id` the first reply returns. + +## Wiring a repo + +```yaml +repos: + - repo: https://github.com/kuhyx/utils + rev: main + hooks: [{id: todo-closer}, {id: todo-import-pr-merge}, {id: todo-import-pr-rewrite}] +default_install_hook_types: [pre-commit, pre-push, post-commit, post-merge, post-rewrite] +``` + +Each clone needs the post-* hook types installed: the `default_install_hook_types` line above, or `pre-commit install -t post-commit -t post-merge -t post-rewrite`. + +## Nightly sweep + +`systemd/todo-drift.service` and `.timer` (03:00) are files only; install with `systemctl --user link` plus `enable --now todo-drift.timer`. + +## Tests + +`python3 -m pytest todo_hooks -q --cov=todo_hooks --cov-config=todo_hooks/.coveragerc` diff --git a/todo_hooks/__init__.py b/todo_hooks/__init__.py new file mode 100644 index 0000000..3c54e11 --- /dev/null +++ b/todo_hooks/__init__.py @@ -0,0 +1 @@ +"""Wire git to the todo app's MCP server: closer, drift sweep, PR import.""" diff --git a/todo_hooks/__main__.py b/todo_hooks/__main__.py new file mode 100644 index 0000000..53690d2 --- /dev/null +++ b/todo_hooks/__main__.py @@ -0,0 +1,45 @@ +"""``python3 -m todo_hooks {closer|sweep|trust|import-pr}``.""" + +from __future__ import annotations + +import sys +from typing import TYPE_CHECKING + +from todo_hooks import closer, import_pr, sweep, trust_cmd +from todo_hooks.context import Context + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + from pathlib import Path + + from todo_hooks.client import TaskClient + +USAGE = "usage: python3 -m todo_hooks {closer|sweep|trust|import-pr} [args]" + + +def main( + argv: list[str] | None = None, + *, + env: Mapping[str, str] | None = None, + cwd: Path | None = None, + stdin: str | None = None, + make_client: Callable[[], TaskClient] | None = None, +) -> int: + """Dispatch one subcommand; returns the process exit code.""" + args = list(sys.argv[1:] if argv is None else argv) + if not args or args[0] not in {"closer", "sweep", "trust", "import-pr"}: + print(USAGE) + return 2 + ctx = Context.build(env, cwd, make_client) + command, rest = args[0], args[1:] + if command == "closer": + return closer.run(ctx) + if command == "sweep": + return sweep.main(ctx, rest) + if command == "trust": + return trust_cmd.run(ctx, rest) + return import_pr.run(ctx, rest, stdin=stdin) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/todo_hooks/checkrun.py b/todo_hooks/checkrun.py new file mode 100644 index 0000000..ef84674 --- /dev/null +++ b/todo_hooks/checkrun.py @@ -0,0 +1,62 @@ +"""Run a done_check with a hard timeout, in its own process group.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import signal +import subprocess +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + +CHECK_TIMEOUT = 55.0 + + +@dataclass(frozen=True) +class CheckResult: + """Exit code, or None on timeout or failure to start.""" + + code: int | None + timed_out: bool + + @property + def passed(self) -> bool: + return self.code == 0 + + def describe(self) -> str: + if self.timed_out: + return f"timed out after {CHECK_TIMEOUT:g}s" + return "could not start" if self.code is None else f"exit {self.code}" + + +def run_check( + command: str, + cwd: Path, + *, + timeout: float = CHECK_TIMEOUT, + wrapper: Path | None = None, +) -> CheckResult: + """``bash -c command`` from `cwd` without GIT_* env; kill the group on timeout.""" + argv = [str(wrapper)] if wrapper else [] + argv += ["bash", "-c", command] + env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} + try: + proc = subprocess.Popen( + argv, + cwd=cwd, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError: + return CheckResult(None, timed_out=False) + try: + return CheckResult(proc.wait(timeout=timeout), timed_out=False) + except subprocess.TimeoutExpired: + os.killpg(proc.pid, signal.SIGKILL) + proc.wait() + return CheckResult(None, timed_out=True) diff --git a/todo_hooks/client.py b/todo_hooks/client.py new file mode 100644 index 0000000..3277798 --- /dev/null +++ b/todo_hooks/client.py @@ -0,0 +1,189 @@ +"""MCP-over-HTTP JSON-RPC client for the todo server (stdlib only).""" + +from __future__ import annotations + +import contextlib +import http.client +import json +from typing import TYPE_CHECKING, Any, Protocol, Self +import urllib.error +import urllib.request + +if TYPE_CHECKING: + from collections.abc import Mapping + from pathlib import Path + +DEFAULT_URL = "http://127.0.0.1:8735/mcp" +DEFAULT_TIMEOUT = 5.0 +_PROTOCOL = "2025-03-26" +_HTTP_AUTH = (401, 403) +_HTTP_SERVER_ERROR = 500 + + +class McpError(Exception): + """Any failure talking to the todo MCP server.""" + + +class McpUnavailable(McpError): + """Server down, timed out, or answering 5xx.""" + + +class McpAuthError(McpError): + """No token, or the server rejected it.""" + + +class McpToolError(McpError): + """The tool ran but reported an error.""" + + +class TaskClient(Protocol): + """What the hooks need from a client; tests substitute a fake.""" + + def call(self, tool: str, arguments: dict[str, str] | None = None) -> str: + """Call a tool and return its text.""" + ... + + def close(self) -> None: + """Release the server session.""" + ... + + +def load_token(env: Mapping[str, str], token_file: Path) -> str: + """Return ``TODO_MCP_TOKEN`` from env, else the token file's content.""" + token = env.get("TODO_MCP_TOKEN", "").strip() + if not token: + try: + token = token_file.read_text(encoding="utf-8").strip() + except OSError: + token = "" + if not token: + msg = f"no token: set TODO_MCP_TOKEN or create {token_file}" + raise McpAuthError(msg) + return token + + +def _parse_reply(content_type: str, body: str) -> dict[str, Any]: + candidates = [body] + if "text/event-stream" in content_type: + candidates = [ + row.removeprefix("data:").strip() + for row in body.splitlines() + if row.startswith("data:") + ] + for raw in candidates: + try: + message = json.loads(raw) + except json.JSONDecodeError: + continue + if isinstance(message, dict): + return message + msg = f"unparseable response: {body[:80]!r}" + raise McpError(msg) + + +class McpClient: + """One lazily initialised MCP session; ``close`` it (or use ``with``).""" + + def __init__(self, url: str, token: str, timeout: float = DEFAULT_TIMEOUT) -> None: + if not url.startswith(("http://", "https://")): + msg = f"unsupported MCP url: {url}" + raise McpError(msg) + self.url = url + self._token = token + self._timeout = timeout + self._session: str | None = None + self._ready = False + self._next_id = 0 + + @classmethod + def from_env( + cls, env: Mapping[str, str], token_file: Path, timeout: float = DEFAULT_TIMEOUT + ) -> McpClient: + """Build from ``TODO_MCP_URL`` and the token (env or file).""" + url = env.get("TODO_MCP_URL") or DEFAULT_URL + return cls(url, load_token(env, token_file), timeout) + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + def _send( + self, payload: dict[str, Any] | None, method: str = "POST" + ) -> tuple[str, str, Mapping[str, str]]: + headers = { + "Authorization": f"Bearer {self._token}", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + } + if self._session: + headers["Mcp-Session-Id"] = self._session + data = json.dumps(payload).encode() if payload is not None else None + request = urllib.request.Request( + self.url, data=data, headers=headers, method=method + ) + try: + with urllib.request.urlopen(request, timeout=self._timeout) as reply: + body = reply.read().decode("utf-8", "replace") + return reply.headers.get("Content-Type", ""), body, reply.headers + except urllib.error.HTTPError as err: + err.close() + if err.code in _HTTP_AUTH: + msg = f"server rejected the token (HTTP {err.code})" + raise McpAuthError(msg) from err + msg = f"HTTP {err.code}" + if err.code >= _HTTP_SERVER_ERROR: + raise McpUnavailable(msg) from err + raise McpError(msg) from err + except (urllib.error.URLError, OSError, http.client.HTTPException) as err: + msg = f"cannot reach {self.url}: {getattr(err, 'reason', err)}" + raise McpUnavailable(msg) from err + + def _rpc(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + self._next_id += 1 + payload = { + "jsonrpc": "2.0", + "id": self._next_id, + "method": method, + "params": params, + } + ctype, body, headers = self._send(payload) + if method == "initialize": + self._session = headers.get("Mcp-Session-Id") + message = _parse_reply(ctype, body) + if "error" in message: + raise McpError(str(message["error"].get("message", message["error"]))) + return message.get("result", {}) + + def _handshake(self) -> None: + info = {"name": "todo-hooks", "version": "1"} + self._rpc( + "initialize", + {"protocolVersion": _PROTOCOL, "capabilities": {}, "clientInfo": info}, + ) + self._send({"jsonrpc": "2.0", "method": "notifications/initialized"}) + self._ready = True + + def call(self, tool: str, arguments: dict[str, str] | None = None) -> str: + """Call `tool` and return its text; ``McpToolError`` if it reports one.""" + if not self._ready: + self._handshake() + result = self._rpc("tools/call", {"name": tool, "arguments": arguments or {}}) + text = "\n".join( + part.get("text", "") + for part in result.get("content", []) + if part.get("type") == "text" + ) + if result.get("isError"): + raise McpToolError(text) + return text + + def close(self) -> None: + """End the server session (best effort, never raises).""" + if self._session is None: + return + with contextlib.suppress(McpError): + self._send(None, method="DELETE") + self._session = None + self._ready = False diff --git a/todo_hooks/closer.py b/todo_hooks/closer.py new file mode 100644 index 0000000..deb3800 --- /dev/null +++ b/todo_hooks/closer.py @@ -0,0 +1,76 @@ +"""Post-commit closer: close ``Todo:`` trailer tasks whose check passes.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks.checkrun import CHECK_TIMEOUT, run_check +from todo_hooks.client import McpError, McpToolError +from todo_hooks.context import describe +from todo_hooks.gitrepo import expected_repo, head_sha, head_todo_ids, toplevel +from todo_hooks.notes import Task, parse_detail +from todo_hooks.trust import is_trusted + +if TYPE_CHECKING: + from pathlib import Path + + from todo_hooks.client import TaskClient + from todo_hooks.context import Context + + +def _refusal(ctx: Context, root: Path, task: Task) -> str | None: + """Why this task must not be auto-closed, or None when it may run.""" + expected = expected_repo(root) + id8 = task.id[:8] + if task.status == "done": + return f"{id8} already done" + if task.repo != expected: + return f"{id8} belongs to {task.repo or 'no repo'}, not {expected}; left open" + if task.is_manual: + return f"{id8} close by hand (Check: manual)" + if not task.done_check: + return f"{id8} has no done_check; close by hand" + if not is_trusted(ctx.paths.trust_file, task.done_check): + return f"{id8} untrusted done_check; run: python3 -m todo_hooks trust {id8}" + return None + + +def _verdict(ctx: Context, root: Path, task_id: str, client: TaskClient) -> str: + """One line saying what happened to the task; closes it on a trusted pass.""" + try: + task = parse_detail(client.call("get_task", {"id": task_id})) + except McpToolError as err: + return f"{task_id[:8]} not found: {err}" + refusal = _refusal(ctx, root, task) + if refusal: + return refusal + result = run_check(task.done_check or "", root, timeout=CHECK_TIMEOUT) + if not result.passed: + return f"{task.id[:8]} left open: done_check {result.describe()}" + evidence = f"{head_sha(root)} done_check exit 0" + try: + client.call("close_task", {"id": task.id, "evidence": evidence}) + except McpToolError as err: + return f"{task.id[:8]} close failed: {err}" + return f"closed {task.id[:8]}" + + +def run(ctx: Context) -> int: + """Always 0: the commit already happened, so a failure here must stay quiet.""" + root = toplevel(ctx.cwd) + ids = head_todo_ids(root) if root else [] + if root is None or not ids: + return 0 + try: + client = ctx.make_client() + except McpError as err: + ctx.say(f"warning: {describe(err)}; nothing closed") + return 0 + try: + for task_id in ids: + ctx.say(_verdict(ctx, root, task_id, client)) + except McpError as err: + ctx.say(f"warning: {describe(err)}; nothing closed") + finally: + client.close() + return 0 diff --git a/todo_hooks/context.py b/todo_hooks/context.py new file mode 100644 index 0000000..ab82d72 --- /dev/null +++ b/todo_hooks/context.py @@ -0,0 +1,61 @@ +"""Everything a subcommand needs, injectable so tests never touch real state.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path +from typing import TYPE_CHECKING + +from todo_hooks.client import McpAuthError, McpClient, McpError, McpUnavailable +from todo_hooks.paths import Paths + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + from todo_hooks.client import TaskClient + +PREFIX = "todo-hooks: " + + +@dataclass +class Context: + """Environment, working directory, paths, client factory and output sink.""" + + env: Mapping[str, str] + cwd: Path + paths: Paths + make_client: Callable[[], TaskClient] + out: Callable[[str], None] + + @classmethod + def build( + cls, + env: Mapping[str, str] | None = None, + cwd: Path | None = None, + make_client: Callable[[], TaskClient] | None = None, + ) -> Context: + """Real defaults: ``os.environ``, the cwd, ``~`` paths, a live client.""" + use_env = dict(os.environ) if env is None else dict(env) + paths = Paths.default(use_env) + + def live() -> TaskClient: + return McpClient.from_env(use_env, paths.token_file) + + return cls( + env=use_env, + cwd=cwd or Path.cwd(), + paths=paths, + make_client=make_client or live, + out=print, + ) + + def say(self, message: str) -> None: + self.out(PREFIX + message) + + +def describe(err: McpError) -> str: + """Server-side failures read as 'unavailable'; tool errors verbatim.""" + if isinstance(err, McpUnavailable | McpAuthError): + return f"todo server unavailable ({err})" + return str(err) diff --git a/todo_hooks/gh.py b/todo_hooks/gh.py new file mode 100644 index 0000000..ca604fd --- /dev/null +++ b/todo_hooks/gh.py @@ -0,0 +1,69 @@ +"""Find the merged pull requests of a commit through the ``gh`` CLI.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import subprocess +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + +GH_TIMEOUT = 20.0 + + +class GhError(Exception): + """gh missing, offline, failing or answering nonsense.""" + + +@dataclass(frozen=True) +class PullRequest: + """A pull request GitHub links to a commit.""" + + number: int + body: str + merged: bool + + +def run_gh(args: list[str]) -> str: + """Run ``gh ``; stdout, or GhError on any failure.""" + try: + done = subprocess.run( + ["gh", *args], + check=False, + capture_output=True, + text=True, + timeout=GH_TIMEOUT, + stdin=subprocess.DEVNULL, + ) + except FileNotFoundError as err: + msg = "gh not installed" + raise GhError(msg) from err + except subprocess.TimeoutExpired as err: + msg = "gh timed out" + raise GhError(msg) from err + if done.returncode != 0: + raise GhError(done.stderr.strip() or f"gh exit {done.returncode}") + return done.stdout + + +def merged_prs( + owner: str, + name: str, + sha: str, + run: Callable[[list[str]], str] | None = None, +) -> list[PullRequest]: + """Pull requests GitHub links to commit `sha`.""" + reply = (run or run_gh)(["api", f"repos/{owner}/{name}/commits/{sha}/pulls"]) + try: + data = json.loads(reply) + return [ + PullRequest( + int(pr["number"]), pr.get("body") or "", bool(pr.get("merged_at")) + ) + for pr in data + ] + except (ValueError, TypeError, KeyError) as err: + msg = "unparseable gh output" + raise GhError(msg) from err diff --git a/todo_hooks/gitrepo.py b/todo_hooks/gitrepo.py new file mode 100644 index 0000000..d808d28 --- /dev/null +++ b/todo_hooks/gitrepo.py @@ -0,0 +1,88 @@ +"""Git helpers: repo root, origin slug, HEAD sha and ``Todo:`` trailers.""" + +from __future__ import annotations + +from pathlib import Path +import re +import subprocess +from urllib.parse import urlsplit + +OWNER = "kuhyx" +_TRAILER_ID = re.compile(r"^[0-9a-fA-F][0-9a-fA-F-]{7,35}$") +_GIT_TIMEOUT = 15 +_SLUG_PARTS = 2 + + +def git(root: Path, *args: str) -> str | None: + """Run git in `root`; stripped stdout, or None on any failure.""" + try: + done = subprocess.run( + ["git", *args], + cwd=root, + check=False, + capture_output=True, + text=True, + timeout=_GIT_TIMEOUT, + stdin=subprocess.DEVNULL, + ) + except OSError, subprocess.TimeoutExpired: + return None + return done.stdout.strip() if done.returncode == 0 else None + + +def toplevel(cwd: Path) -> Path | None: + out = git(cwd, "rev-parse", "--show-toplevel") + return Path(out).resolve() if out else None + + +def _path_of(url: str) -> str: + url = url.strip().rstrip("/").removesuffix(".git") + if "://" in url: + return urlsplit(url).path.strip("/") + if ":" in url and not url.startswith("/"): + return url.split(":", 1)[1].strip("/") + return url + + +def parse_slug(url: str) -> tuple[str, str] | None: + """(owner, name) from an scp-style, ssh:// or https:// remote URL.""" + if "://" not in url and (url.startswith("/") or ":" not in url): + return None + parts = [seg for seg in _path_of(url).split("/") if seg] + return (parts[-2], parts[-1]) if len(parts) >= _SLUG_PARTS else None + + +def origin_url(root: Path) -> str | None: + return git(root, "remote", "get-url", "origin") + + +def origin_slug(root: Path) -> tuple[str, str] | None: + url = origin_url(root) + return parse_slug(url) if url else None + + +def repo_name(root: Path) -> str: + """Name from the origin URL; the checkout basename without a remote.""" + url = origin_url(root) + name = _path_of(url).rsplit("/", 1)[-1] if url else "" + return name or root.name + + +def expected_repo(root: Path) -> str: + """The ``kuhyx/`` value a task for this repository must carry.""" + return f"{OWNER}/{repo_name(root)}" + + +def head_sha(root: Path) -> str: + return git(root, "rev-parse", "--short=7", "HEAD") or "" + + +def head_todo_ids(root: Path) -> list[str]: + """Task ids from HEAD's ``Todo:`` trailers (first token, deduplicated).""" + out = git(root, "log", "-1", "--format=%(trailers:key=Todo,valueonly)") or "" + ids: list[str] = [] + for row in out.splitlines(): + token = row.split()[0] if row.strip() else "" + if _TRAILER_ID.match(token) and token not in ids: + ids.append(token) + return ids diff --git a/todo_hooks/import_pr.py b/todo_hooks/import_pr.py new file mode 100644 index 0000000..d83791d --- /dev/null +++ b/todo_hooks/import_pr.py @@ -0,0 +1,135 @@ +"""Import ``## Todo:`` items from the PRs of newly pulled commits as tasks.""" + +from __future__ import annotations + +import select +import sys +from typing import TYPE_CHECKING + +from todo_hooks import gh +from todo_hooks.client import McpError +from todo_hooks.context import describe +from todo_hooks.gitrepo import expected_repo, git, origin_slug, toplevel +from todo_hooks.notes import parse_list +from todo_hooks.pr_parse import TodoItem, parse_todo_items + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.client import TaskClient + from todo_hooks.context import Context + +MAX_COMMITS = 30 +USAGE = ( + "usage: python3 -m todo_hooks import-pr " + "post-merge | post-rewrite [rebase] | --range A..B" +) + + +def _rev_list(root: Path, spec: str) -> list[str]: + out = git(root, "rev-list", f"--max-count={MAX_COMMITS}", spec) + return out.splitlines() if out else [] + + +def commits_post_merge(root: Path) -> list[str]: + """Commits a pull brought in: ORIG_HEAD..HEAD.""" + return _rev_list(root, "ORIG_HEAD..HEAD") + + +def commits_post_rewrite(root: Path, command: str, pairs: str) -> list[str]: + """New shas from git's ``old new`` stdin lines; only for a rebase.""" + if command != "rebase": + return [] + shas = [row.split()[1] for row in pairs.splitlines() if len(row.split()) >= 2] # noqa: PLR2004 + return shas[:MAX_COMMITS] or commits_post_merge(root) + + +def read_stdin(timeout: float = 1.0) -> str: + """Stdin text if any arrives within `timeout`; never blocks a hook.""" + try: + ready, _, _ = select.select([sys.stdin], [], [], timeout) + except ValueError, OSError: + return "" + return sys.stdin.read() if ready else "" + + +def _exists(client: TaskClient, repo: str, number: int, item: TodoItem) -> bool: + found = parse_list(client.call("search_tasks", {"query": f"PR #{number}"})) + return any(r.repo == repo and r.title == item.title for r in found) + + +def _add(client: TaskClient, repo: str, number: int, item: TodoItem) -> bool: + if _exists(client, repo, number, item): + return False + args = { + "repo": repo, + "title": item.title, + "body": f"Imported from PR #{number} of {repo}.", + "priority": "medium", + } + if item.done_check: + args["done_check"] = item.done_check + client.call("add_task", args) + return True + + +def import_for_commits( + ctx: Context, shas: list[str], runner: Callable[[list[str]], str] | None +) -> int: + """Add a task per new ``## Todo:`` item; warns and returns on gh/server failure.""" + root = toplevel(ctx.cwd) + if not shas or root is None: + return 0 + slug = origin_slug(root) + if slug is None: + ctx.say("no GitHub origin; PR import skipped") + return 0 + repo, seen, total = expected_repo(root), set(), 0 + client: TaskClient | None = None + try: + for sha in shas: + for pr in gh.merged_prs(slug[0], slug[1], sha, runner): + items = parse_todo_items(pr.body) + if not pr.merged or pr.number in seen or not items: + continue + seen.add(pr.number) + client = client or ctx.make_client() + added = sum(_add(client, repo, pr.number, item) for item in items) + if added: + ctx.say(f"imported {added} task(s) from PR #{pr.number}") + total += added + except gh.GhError as err: + ctx.say(f"warning: gh unavailable ({err}); PR import skipped") + except McpError as err: + ctx.say(f"warning: {describe(err)}; PR import skipped") + finally: + if client: + client.close() + return total + + +def run( + ctx: Context, + args: list[str], + gh_runner: Callable[[list[str]], str] | None = None, + stdin: str | None = None, +) -> int: + """Hook entry: always 0 except a usage error (2).""" + mode = args[0] if args else "post-merge" + root = toplevel(ctx.cwd) + if mode == "--range" and len(args) == 2: # noqa: PLR2004 + shas = _rev_list(root, args[1]) if root else [] + elif mode == "post-merge": + shas = commits_post_merge(root) if root else [] + elif mode == "post-rewrite": + command = ( + args[1] if len(args) > 1 else ctx.env.get("PRE_COMMIT_REWRITE_COMMAND", "") + ) + pairs = read_stdin() if stdin is None else stdin + shas = commits_post_rewrite(root, command, pairs) if root else [] + else: + ctx.out(USAGE) + return 2 + import_for_commits(ctx, shas, gh_runner) + return 0 diff --git a/todo_hooks/notes.py b/todo_hooks/notes.py new file mode 100644 index 0000000..7361e0c --- /dev/null +++ b/todo_hooks/notes.py @@ -0,0 +1,83 @@ +"""Parse the todo server's list_tasks and get_task text replies.""" + +from __future__ import annotations + +from dataclasses import dataclass +import re + +_FIELDS = ("id", "priority", "status", "created", "updated", "repo") +_FIELDS += ("done_check", "by", "evidence") +_META_LINE = re.compile(rf"^({'|'.join(_FIELDS)}): ?(.*)$") +_MANUAL = re.compile(r"^\s*check:\s*manual\s*$", re.IGNORECASE | re.MULTILINE) +_COLUMNS = 5 + + +def clean_title(title: str) -> str: + """Drop surrounding space and a leading markdown ``#``.""" + return re.sub(r"^#+\s*", "", title.strip()) + + +@dataclass(frozen=True) +class TaskLine: + """One list_tasks row.""" + + id8: str + status: str + priority: str + repo: str + title: str + + +@dataclass(frozen=True) +class Task: + """One get_task reply: metadata lines, ``---``, then the note text.""" + + id: str + status: str + repo: str | None + done_check: str | None + text: str + + @property + def title(self) -> str: + return clean_title(self.text.split("\n", 1)[0]) + + @property + def is_manual(self) -> bool: + return _MANUAL.search(self.text) is not None + + +def parse_list(text: str) -> list[TaskLine]: + """Rows of tab-separated ``id8 status priority repo title`` lines.""" + rows = [] + for raw in text.splitlines(): + parts = raw.split("\t") + if len(parts) >= _COLUMNS: + title = clean_title("\t".join(parts[4:])) + rows.append(TaskLine(parts[0], parts[1], parts[2], parts[3], title)) + return rows + + +def parse_detail(text: str) -> Task: + """Split a get_task reply into metadata fields and the note text.""" + meta: dict[str, str] = {} + last = "" + lines = text.split("\n") + body: list[str] = [] + for index, row in enumerate(lines): + if row == "---": + body = lines[index + 1 :] + break + found = _META_LINE.match(row) + if found: + last = found.group(1) + meta[last] = found.group(2) + elif last: + meta[last] += "\n" + row + return Task( + id=meta.get("id", ""), + status=meta.get("status", ""), + repo=meta.get("repo"), + done_check=meta.get("done_check"), + text="\n".join(body), + ) diff --git a/todo_hooks/paths.py b/todo_hooks/paths.py new file mode 100644 index 0000000..5ca104d --- /dev/null +++ b/todo_hooks/paths.py @@ -0,0 +1,43 @@ +"""Every on-disk location the hooks use, overridable for tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Mapping + + +@dataclass +class Paths: + """Config, state and lookup locations used by the todo hooks.""" + + token_file: Path + trust_file: Path + report_file: Path + src_root: Path + capped: Path | None + + @classmethod + def default(cls, env: Mapping[str, str], home: Path | None = None) -> Paths: + """Resolve defaults under `home`; ``TODO_*`` env vars override each.""" + base = home if home is not None else Path.home() + capped = base / ".claude" / "scripts" / "capped.sh" + return cls( + token_file=Path( + env.get("TODO_MCP_TOKEN_FILE") + or base / ".config" / "todo-mcp" / "client-token" + ), + trust_file=Path( + env.get("TODO_HOOKS_TRUST_FILE") + or base / ".config" / "todo-hooks" / "trusted-checks" + ), + report_file=Path( + env.get("TODO_HOOKS_REPORT") + or base / ".local" / "state" / "todo-drift" / "report.txt" + ), + src_root=Path(env.get("TODO_HOOKS_SRC_ROOT") or base / "src"), + capped=capped if capped.is_file() else None, + ) diff --git a/todo_hooks/pr_parse.py b/todo_hooks/pr_parse.py new file mode 100644 index 0000000..9239940 --- /dev/null +++ b/todo_hooks/pr_parse.py @@ -0,0 +1,43 @@ +"""Extract ``- title | done_check: cmd`` items from a PR's Todo section.""" + +from __future__ import annotations + +from dataclasses import dataclass +import re + +_HEADING = re.compile(r"^#{1,6}\s*todo:?\s*$", re.IGNORECASE) +_BULLET = re.compile(r"^\s*[-*]\s+(.*\S)\s*$") +_CHECK = re.compile(r"^(.*?)\s*\|\s*done_check:\s*(.*)$", re.IGNORECASE) + + +@dataclass(frozen=True) +class TodoItem: + """One imported task: title and optional done_check.""" + + title: str + done_check: str | None + + +def parse_todo_items(body: str) -> list[TodoItem]: + """Bullets under a ``## Todo:`` heading, up to the first non-bullet text.""" + items: list[TodoItem] = [] + inside = False + for row in body.splitlines(): + if not inside: + inside = bool(_HEADING.match(row)) + continue + bullet = _BULLET.match(row) + if bullet: + item = _item(bullet.group(1)) + if item: + items.append(item) + elif row.strip(): + break + return items + + +def _item(text: str) -> TodoItem | None: + found = _CHECK.match(text) + title, check = (found.group(1), found.group(2)) if found else (text, "") + title = title.strip() + return TodoItem(title, check.strip() or None) if title else None diff --git a/todo_hooks/pyproject.toml b/todo_hooks/pyproject.toml new file mode 100644 index 0000000..af067de --- /dev/null +++ b/todo_hooks/pyproject.toml @@ -0,0 +1,177 @@ +[project] +name = "todo-hooks" +version = "0.1.0" +description = "Git hooks that close, sweep and import todo-app tasks over MCP" +requires-python = ">=3.14" +dependencies = [] + +[tool.ruff] +target-version = "py314" +include = ["*.py", "**/*.py"] +exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"] + +[tool.ruff.lint] +select = ["ALL"] +ignore = [ + "D203", + "D213", + "COM812", + "ISC001", + "S603", + "S607", + "CPY001", + "N818", + "D102", + "D103", + "D105", + "D107", +] +fixable = ["ALL"] +unfixable = [] + +[tool.ruff.lint.per-file-ignores] +"**/client.py" = ["S310"] +"**/__main__.py" = ["T201"] +"**/tests/**/*.py" = ["ARG", "C901", "D", "E501", "EM101", "FBT", "PLC0415", "PLR2004", "PLR0915", "S101", "S105", "SLF001", "TRY003"] +"**/test_*.py" = ["ARG", "D", "E501", "EM101", "FBT", "PLC0415", "PLR2004", "S101", "SLF001", "TRY003"] + +[tool.ruff.lint.pylint] +# Same as [tool.pylint.design] max-args: resolve()'s keyword-only switches. +max-args = 6 + +[tool.ruff.lint.pydocstyle] +convention = "google" + +[tool.ruff.lint.isort] +force-single-line = false +force-sort-within-sections = true +known-first-party = ["todo_hooks"] + +[tool.ruff.lint.flake8-quotes] +docstring-quotes = "double" +inline-quotes = "double" + +[tool.ruff.lint.flake8-tidy-imports] +ban-relative-imports = "all" + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" +skip-magic-trailing-comma = false +line-ending = "auto" +docstring-code-format = true + +[tool.mypy] +python_version = "3.14" +strict = true +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true +disallow_untyped_decorators = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +warn_unreachable = true +disallow_any_unimported = true +disallow_any_explicit = false +disallow_any_generics = true +disallow_subclassing_any = true +strict_equality = true +extra_checks = true +ignore_missing_imports = true +show_error_codes = true +color_output = true +exclude = [".venv/"] + +[tool.pylint.main] +# The repo root on the path, so pylint resolves the package's own imports the +# same way pytest does instead of reporting import-error for them. +init-hook = "import sys; sys.path.insert(0, 'todo_hooks')" +analyse-fallback-blocks = true +persistent = true +jobs = 0 +py-version = "3.14" +ignore = [".venv", "__pycache__"] +# Test suites intentionally use patterns (protected-access, fixture-arg +# shadowing) that don't apply to source code; ruff's per-file-ignores carry +# the equivalent exemptions, so tests stay linted -- by ruff. This is the same +# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has +# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the +# basename of what pylint is *given*, and pre-commit hands it explicit file +# paths, so a directory name in `ignore` never fires under the hook. +ignore-paths = ["^(.*/)?tests/.*$"] +ignore-patterns = [".*\\.pyi$"] + +[tool.pylint.messages_control] +enable = "all" +disable = [] + +[tool.pylint.design] +min-public-methods = 0 +max-module-lines = 1000 +# Earner is a frozen registry record: its fields are the policy (cf. music_theory). +max-attributes = 13 +# resolve()'s keyword-only switches (first_credits, sick, leisure) are the +# public API every consumer calls; bundling them would break each caller. +max-args = 6 + +[tool.pylint.typecheck] +generated-members = [ + ".*\\.assert_called_once_with", + ".*\\.assert_called_once", + ".*\\.assert_called", + ".*\\.assert_not_called", + ".*\\.assert_any_call", + ".*\\.call_args", + ".*\\.call_args_list", + ".*\\.call_count", +] + +[tool.bandit] +exclude_dirs = ["tests", ".venv"] + +[tool.pytest.ini_options] +testpaths = ["todo_hooks/tests"] +python_files = ["test_*.py", "*_test.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "-v", + "--strict-markers", + "--strict-config", + "-ra", + "--cov=todo_hooks", + "--cov-branch", + "--cov-report=term-missing", + "--cov-report=lcov", +] +filterwarnings = [ + "error", + "ignore::DeprecationWarning", + "default::pytest.PytestUnraisableExceptionWarning", +] + +[tool.coverage.run] +source = ["todo_hooks"] +branch = true +omit = [ + "*/__pycache__/*", + "*/tests/*", + "*/.venv/*", +] + +[tool.coverage.report] +fail_under = 100 +show_missing = true +skip_covered = false +exclude_lines = [ + "pragma: no cover", + "raise NotImplementedError", + "raise AssertionError", + "if TYPE_CHECKING:", + 'if __name__ == "__main__":', +] +partial_branches = ["pragma: no branch"] diff --git a/todo_hooks/sweep.py b/todo_hooks/sweep.py new file mode 100644 index 0000000..cbc96a1 --- /dev/null +++ b/todo_hooks/sweep.py @@ -0,0 +1,159 @@ +"""Drift sweep: which open notes already have a passing, trusted done_check.""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass, field, replace +import os +from pathlib import Path +import tempfile +from typing import TYPE_CHECKING + +from todo_hooks.checkrun import CHECK_TIMEOUT, run_check +from todo_hooks.client import McpError, McpToolError +from todo_hooks.context import describe +from todo_hooks.gitrepo import expected_repo +from todo_hooks.notes import Task, parse_detail, parse_list +from todo_hooks.trust import is_trusted + +if TYPE_CHECKING: + from collections.abc import Callable + + from todo_hooks.checkrun import CheckResult + from todo_hooks.context import Context + + +USAGE = ( + "usage: python3 -m todo_hooks sweep [--dry-run] [--report PATH] [--trust-file PATH]" +) + + +@dataclass +class Tally: + """Per-outcome counts plus the report rows they produced.""" + + total: int = 0 + counts: dict[str, int] = field(default_factory=dict) + passing: list[str] = field(default_factory=list) + manual: list[str] = field(default_factory=list) + + def record(self, outcome: str, label: str) -> None: + self.counts[outcome] = self.counts.get(outcome, 0) + 1 + if outcome == "pass": + self.passing.append(label) + elif outcome == "manual": + self.manual.append(label) + + def lines(self, *, dry_run: bool) -> list[str]: + head = "would-run " if dry_run else "" + rows = [head + row for row in self.passing] + rows += [f"# close by hand: {row}" for row in self.manual] + get = self.counts.get + tail = f"{get('untrusted', 0)} untrusted, {get('manual', 0)} manual" + tail += f", {get('nolocal', 0)} no local repo" + if dry_run: + summary = ( + f"# dry-run: {self.total} open notes: " + f"{get('pass', 0)} would run, {tail}" + ) + else: + summary = ( + f"# swept {self.total} open notes: {get('pass', 0)} pass, " + f"{get('fail', 0)} fail, {get('timeout', 0)} timeout, {tail}" + ) + return [*rows, summary] + + +def local_repos(src_root: Path) -> dict[str, Path]: + """Map ``kuhyx/`` to each git checkout directly under src_root.""" + if not src_root.is_dir(): + return {} + return { + expected_repo(child): child + for child in sorted(src_root.iterdir()) + if (child / ".git").exists() + } + + +def _write_report(path: Path, lines: list[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".report-") + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write("\n".join(lines) + "\n") + Path(tmp).replace(path) + + +def run( + ctx: Context, + *, + dry_run: bool = False, + report: Path | None = None, + runner: Callable[..., CheckResult] = run_check, +) -> int: + """Sweep every open repo note; 1 when the server cannot be reached.""" + repos = local_repos(ctx.paths.src_root) + tally = Tally() + try: + client = ctx.make_client() + try: + rows = parse_list(client.call("list_tasks", {"status": "open"})) + tally.total = len(rows) + for row in rows: + root = repos.get(row.repo) + if root is None: + tally.record("nolocal", row.id8) + continue + try: + task = parse_detail(client.call("get_task", {"id": row.id8})) + except McpToolError: + continue + outcome = _outcome(ctx, task, root, runner, dry_run=dry_run) + label = f"{row.id8} {row.repo} {row.title}" + tally.record(outcome, label) + finally: + client.close() + except McpError as err: + ctx.say(f"sweep failed: {describe(err)}") + return 1 + lines = tally.lines(dry_run=dry_run) + _write_report(report or ctx.paths.report_file, lines) + for line in lines: + ctx.out(line) + return 0 + + +def _outcome( + ctx: Context, + task: Task, + root: Path, + runner: Callable[..., CheckResult], + *, + dry_run: bool, +) -> str: + check = task.done_check + if task.status == "done" or not check: + return "skip" + if task.is_manual: + return "manual" + if not is_trusted(ctx.paths.trust_file, check): + return "untrusted" + if dry_run: + return "pass" + result = runner(check, root, timeout=CHECK_TIMEOUT, wrapper=ctx.paths.capped) + return "timeout" if result.timed_out else "pass" if result.passed else "fail" + + +def main(ctx: Context, args: list[str]) -> int: + """Parse ``--dry-run``, ``--report PATH``, ``--trust-file PATH`` and run.""" + parser = argparse.ArgumentParser(prog="todo_hooks sweep", exit_on_error=False) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--report", type=Path) + parser.add_argument("--trust-file", type=Path) + try: + opts = parser.parse_args(args) + except argparse.ArgumentError, SystemExit: + ctx.out(USAGE) + return 2 + if opts.trust_file: + ctx.paths = replace(ctx.paths, trust_file=opts.trust_file) + return run(ctx, dry_run=opts.dry_run, report=opts.report) diff --git a/todo_hooks/systemd/todo-drift.service b/todo_hooks/systemd/todo-drift.service new file mode 100644 index 0000000..a9ba629 --- /dev/null +++ b/todo_hooks/systemd/todo-drift.service @@ -0,0 +1,9 @@ +[Unit] +Description=Todo drift sweep: open notes whose done_check already passes + +[Service] +Type=oneshot +ExecStart=%h/src/utils/scripts/todo_hooks.sh sweep +Nice=19 +IOSchedulingClass=idle +TimeoutStartSec=30min diff --git a/todo_hooks/systemd/todo-drift.timer b/todo_hooks/systemd/todo-drift.timer new file mode 100644 index 0000000..ebc3f69 --- /dev/null +++ b/todo_hooks/systemd/todo-drift.timer @@ -0,0 +1,9 @@ +[Unit] +Description=Nightly todo drift sweep + +[Timer] +OnCalendar=*-*-* 03:00:00 +Persistent=false + +[Install] +WantedBy=timers.target diff --git a/todo_hooks/tests/__init__.py b/todo_hooks/tests/__init__.py new file mode 100644 index 0000000..ae21b56 --- /dev/null +++ b/todo_hooks/tests/__init__.py @@ -0,0 +1 @@ +"""todo_hooks tests.""" diff --git a/todo_hooks/tests/_closer_helpers.py b/todo_hooks/tests/_closer_helpers.py new file mode 100644 index 0000000..49c4bfc --- /dev/null +++ b/todo_hooks/tests/_closer_helpers.py @@ -0,0 +1,36 @@ +"""Setup shared by the closer tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks.trust import add_trusted + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + from todo_hooks.paths import Paths + from todo_hooks.tests._fakes import FakeClient + +ID = "0123abcd-0000-4000-8000-000000000001" +ID2 = "4567ef01-0000-4000-8000-000000000002" +MSG = "feat: x\n\nTodo: 0123abcd" + + +def setup( # noqa: PLR0913 + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, + client: FakeClient, + *, + check: str | None = "true", + trust: bool = True, + message: str = MSG, +) -> tuple[Path, Context, list[str]]: + root = make_repo("demo", origin="git@github.com:kuhyx/demo.git", message=message) + if trust and check: + add_trusted(paths.trust_file, check, "test") + ctx, lines = make_ctx(client, root) + return root, ctx, lines diff --git a/todo_hooks/tests/_fakes.py b/todo_hooks/tests/_fakes.py new file mode 100644 index 0000000..dcabc95 --- /dev/null +++ b/todo_hooks/tests/_fakes.py @@ -0,0 +1,239 @@ +"""Test doubles: a scriptable task client and a stdlib fake MCP HTTP server.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import threading +import time +from typing import TYPE_CHECKING, Any + +from todo_hooks.client import McpToolError + +if TYPE_CHECKING: + from collections.abc import Callable + +TOKEN = "test-token" +SESSION = "sess-1" + + +def detail( + task_id: str, + *, + repo: str = "kuhyx/demo", + done_check: str | None = "true", + text: str = "# Title\n\nBody", + status: str = "todo", + by: str = "claude", +) -> str: + """Build a get_task reply the way the Dart server formats it.""" + lines = [f"id: {task_id}", "priority: medium", f"status: {status}"] + lines += ["created: 2026-11-02T10:00:00", "updated: 2026-11-02T10:00:00"] + lines.append(f"repo: {repo}") + if done_check is not None: + lines.append(f"done_check: {done_check}") + lines += [f"by: {by}", "---", text] + return "\n".join(lines) + + +def line(task_id: str, repo: str = "kuhyx/demo", title: str = "# Title") -> str: + """One list_tasks line: id8, status, priority, repo, title.""" + return "\t".join([task_id[:8], "todo", "medium", repo, title]) + + +@dataclass +class FakeClient: + """Answers tool calls from a dict of tool -> reply (str, callable, or error).""" + + replies: dict[str, Any] = field(default_factory=dict) + calls: list[tuple[str, dict[str, str]]] = field(default_factory=list) + + def call(self, tool: str, arguments: dict[str, str] | None = None) -> str: + """Record the call and return or raise the scripted reply.""" + args = arguments or {} + self.calls.append((tool, args)) + reply = self.replies.get(tool, "") + if callable(reply): + reply = reply(args) + if isinstance(reply, Exception): + raise reply + return str(reply) + + def close(self) -> None: + """Match the real client; nothing to release.""" + + def tools(self) -> list[str]: + """Names of the tools called, in order.""" + return [name for name, _ in self.calls] + + +def task_client(*details: str, **extra: Any) -> FakeClient: # noqa: ANN401 + """A client whose get_task answers by id prefix from `details`.""" + by_id = {d.splitlines()[0].removeprefix("id: "): d for d in details} + + def get_task(args: dict[str, str]) -> str | Exception: + for task_id, text in by_id.items(): + if task_id.startswith(args["id"]): + return text + return McpToolError("error: id: no live note") + + return FakeClient({"get_task": get_task, "close_task": "closed x", **extra}) + + +class FakeMcp: + """A loopback server speaking just enough Streamable HTTP like mcp_dart.""" + + def __init__(self) -> None: + self.tools: dict[str, Callable[[dict[str, Any]], str]] = {} + self.requests: list[dict[str, Any]] = [] + self.mode = "sse" # sse | json | garbage | rpc-error | tool-error + self.status = 200 + self.delay = 0.0 + self.stateless = False + self.token = TOKEN + owner = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_args: object) -> None: + return + + def _send( + self, + code: int, + body: str, + ctype: str, + extra: dict[str, str] | None = None, + ) -> None: + data = body.encode() + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + for key, value in (extra or {}).items(): + self.send_header(key, value) + self.end_headers() + self.wfile.write(data) + + def do_DELETE(self) -> None: + owner.requests.append( + {"method": "DELETE", "session": self.headers.get("Mcp-Session-Id")} + ) + self._send(200, "", "text/plain") + + def do_POST(self) -> None: + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + owner.requests.append( + { + "method": body.get("method"), + "body": body, + "session": self.headers.get("Mcp-Session-Id"), + "accept": self.headers.get("Accept"), + } + ) + if self.headers.get("Authorization") != f"Bearer {owner.token}": + self._send(403, "Forbidden", "text/plain") + return + if owner.delay: + time.sleep(owner.delay) + if owner.status != 200: + self._send(owner.status, "boom", "text/plain") + return + self._answer(body) + + def _answer(self, body: dict[str, Any]) -> None: + method = body.get("method") + if method == "notifications/initialized": + self._send(202, "", "text/plain") + elif method == "initialize": + result = {"result": {"protocolVersion": "2025-03-26"}} + self._reply( + body, + result, + {} if owner.stateless else {"Mcp-Session-Id": SESSION}, + ) + elif ( + not owner.stateless + and self.headers.get("Mcp-Session-Id") != SESSION + ): + err = {"error": {"code": -32000, "message": "no session"}} + self._reply(body, err) + else: + self._tool(body) + + def _tool(self, body: dict[str, Any]) -> None: + params = body["params"] + if owner.mode == "rpc-error": + self._reply(body, {"error": {"code": -32601, "message": "nope"}}) + elif owner.mode == "garbage": + self._send(200, "not json at all", "application/json") + elif owner.mode == "no-data": + self._send(200, "event: ping\n\n", "text/event-stream") + else: + text = owner.tools[params["name"]](params["arguments"]) + result = { + "content": [{"type": "text", "text": text}], + "isError": owner.mode == "tool-error", + } + self._reply(body, {"result": result}) + + def _reply( + self, + body: dict[str, Any], + part: dict[str, Any], + extra: dict[str, str] | None = None, + ) -> None: + payload = json.dumps({"jsonrpc": "2.0", "id": body.get("id"), **part}) + if owner.mode == "json": + self._send(200, payload, "application/json", extra) + else: + sse = f"event: message\ndata: {payload}\n\n" + self._send(200, sse, "text/event-stream", extra) + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + + @property + def url(self) -> str: + """The /mcp endpoint on the ephemeral port.""" + return f"http://127.0.0.1:{self._server.server_address[1]}/mcp" + + def start(self) -> FakeMcp: + """Begin serving in a daemon thread.""" + self._thread.start() + return self + + def stop(self) -> None: + """Stop serving and release the port.""" + self._server.shutdown() + self._server.server_close() + + +BODY = """Intro text. + +## Todo: +- Add retry to sync | done_check: pytest -q tests/test_retry.py +- Document the flag +* Star bullet | done_check: true + +## Other +- not a todo | done_check: false +""" + + +class FakeGh: + """Scripted `gh api` answers by commit sha.""" + + def __init__(self, pulls: dict[str, list[dict[str, object]]]) -> None: + self.pulls = pulls + self.calls: list[list[str]] = [] + + def __call__(self, args: list[str]) -> str: + self.calls.append(args) + sha = args[1].split("/")[4] + return json.dumps(self.pulls.get(sha, [])) + + +def make_pr( + number: int, body: str | None = BODY, merged: str | None = "2026-11-01" +) -> dict[str, object]: + return {"number": number, "body": body, "merged_at": merged} diff --git a/todo_hooks/tests/_pr_helpers.py b/todo_hooks/tests/_pr_helpers.py new file mode 100644 index 0000000..c115a34 --- /dev/null +++ b/todo_hooks/tests/_pr_helpers.py @@ -0,0 +1,25 @@ +"""Helpers shared by the PR import tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks.tests._fakes import FakeClient +from todo_hooks.tests.conftest import git + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + +def search_client(existing: str = "no tasks match") -> FakeClient: + return FakeClient({"search_tasks": existing, "add_task": "added x"}) + + +def pr_repo(make_repo: Callable[..., Path], n: int = 2) -> Path: + root = make_repo("demo", origin="git@github.com:kuhyx/demo.git") + for i in range(n): + (root / f"g{i}").write_text("x") + git(root, "add", f"g{i}") + git(root, "commit", "-q", "-m", f"c{i}") + return root diff --git a/todo_hooks/tests/conftest.py b/todo_hooks/tests/conftest.py new file mode 100644 index 0000000..70fec5e --- /dev/null +++ b/todo_hooks/tests/conftest.py @@ -0,0 +1,99 @@ +"""Shared fixtures: fake MCP server, temp git repos, tmp-rooted Paths.""" + +from __future__ import annotations + +from pathlib import Path +import subprocess +import sys +from typing import TYPE_CHECKING + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from todo_hooks.context import Context +from todo_hooks.paths import Paths +from todo_hooks.tests._fakes import FakeMcp + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + +_GIT_ENV = ["-c", "user.name=t", "-c", "user.email=t@t", "-c", "commit.gpgsign=false"] + + +@pytest.fixture +def mcp() -> Iterator[FakeMcp]: + """A running fake MCP server on an ephemeral port.""" + server = FakeMcp().start() + yield server + server.stop() + + +@pytest.fixture +def paths(tmp_path: Path) -> Paths: + """Every path rooted in tmp_path, never in the real home.""" + return Paths( + token_file=tmp_path / "cfg" / "client-token", + trust_file=tmp_path / "cfg" / "trusted-checks", + report_file=tmp_path / "state" / "report.txt", + src_root=tmp_path / "src", + capped=None, + ) + + +def git(cwd: Path, *args: str) -> str: + """Run git in `cwd` with a fixed identity and return stdout.""" + done = subprocess.run( + ["git", *_GIT_ENV, *args], + cwd=cwd, + check=True, + capture_output=True, + text=True, + ) + return done.stdout.strip() + + +@pytest.fixture +def make_repo(tmp_path: Path) -> Callable[..., Path]: + """Factory: init a repo with an optional origin URL and one commit.""" + + def _make( + name: str = "demo", + *, + origin: str | None = None, + message: str = "init", + parent: Path | None = None, + ) -> Path: + root = (parent or tmp_path / "repos") / name + root.mkdir(parents=True) + git(root, "init", "-q", "-b", "main") + if origin: + git(root, "remote", "add", "origin", origin) + (root / "f.txt").write_text("x") + git(root, "add", "f.txt") + git(root, "commit", "-q", "-m", message) + return root + + return _make + + +@pytest.fixture +def make_ctx(paths: Paths) -> Callable[..., tuple[Context, list[str]]]: + """Factory: a Context around a fake client, with captured output lines.""" + + def _make( + client: object, + cwd: Path, + env: dict[str, str] | None = None, + ) -> tuple[Context, list[str]]: + lines: list[str] = [] + ctx = Context( + env=env or {}, + cwd=cwd, + paths=paths, + make_client=lambda: client, # type: ignore[arg-type,return-value] + out=lines.append, + ) + return ctx, lines + + return _make diff --git a/todo_hooks/tests/test_client.py b/todo_hooks/tests/test_client.py new file mode 100644 index 0000000..c58c89f --- /dev/null +++ b/todo_hooks/tests/test_client.py @@ -0,0 +1,198 @@ +"""Tests for the MCP-over-HTTP client against the stdlib fake server.""" + +from __future__ import annotations + +import socket +from typing import TYPE_CHECKING + +import pytest + +from todo_hooks.client import ( + DEFAULT_URL, + McpAuthError, + McpClient, + McpError, + McpToolError, + McpUnavailable, + _parse_reply, + load_token, +) +from todo_hooks.tests._fakes import SESSION, TOKEN, FakeMcp + +if TYPE_CHECKING: + from pathlib import Path + + +def _client(mcp: FakeMcp, timeout: float = 5.0) -> McpClient: + return McpClient(mcp.url, TOKEN, timeout=timeout) + + +def test_load_token_prefers_env(tmp_path: Path) -> None: + file = tmp_path / "t" + file.write_text("from-file\n") + assert load_token({"TODO_MCP_TOKEN": " from-env "}, file) == "from-env" + + +def test_load_token_falls_back_to_file(tmp_path: Path) -> None: + file = tmp_path / "t" + file.write_text("from-file\n") + assert load_token({}, file) == "from-file" + + +def test_load_token_missing_raises(tmp_path: Path) -> None: + with pytest.raises(McpAuthError, match="no token"): + load_token({}, tmp_path / "absent") + + +def test_load_token_empty_file_raises(tmp_path: Path) -> None: + file = tmp_path / "t" + file.write_text("\n") + with pytest.raises(McpAuthError): + load_token({}, file) + + +def test_from_env_uses_url_override(tmp_path: Path) -> None: + file = tmp_path / "t" + file.write_text("tok") + client = McpClient.from_env({"TODO_MCP_URL": "http://x/mcp"}, file) + assert client.url == "http://x/mcp" + assert McpClient.from_env({}, file).url == DEFAULT_URL + + +def test_handshake_then_call(mcp: FakeMcp) -> None: + mcp.tools["list_tasks"] = lambda args: f"saw {args['status']}" + client = _client(mcp) + assert client.call("list_tasks", {"status": "open"}) == "saw open" + methods = [r["method"] for r in mcp.requests] + assert methods == ["initialize", "notifications/initialized", "tools/call"] + assert mcp.requests[1]["session"] == SESSION + assert mcp.requests[2]["session"] == SESSION + assert "text/event-stream" in mcp.requests[0]["accept"] + + +def test_second_call_reuses_session(mcp: FakeMcp) -> None: + mcp.tools["get_task"] = lambda args: args["id"] + client = _client(mcp) + client.call("get_task", {"id": "a"}) + client.call("get_task", {"id": "b"}) + assert [r["method"] for r in mcp.requests].count("initialize") == 1 + + +def test_json_content_type_is_parsed(mcp: FakeMcp) -> None: + mcp.mode = "json" + mcp.tools["get_task"] = lambda _a: "plain" + assert _client(mcp).call("get_task", {"id": "x"}) == "plain" + + +def test_no_arguments_sends_empty_object(mcp: FakeMcp) -> None: + mcp.tools["list_tasks"] = str + assert _client(mcp).call("list_tasks") == "{}" + + +def test_bad_token_is_auth_error(mcp: FakeMcp) -> None: + with pytest.raises(McpAuthError): + McpClient(mcp.url, "wrong").call("list_tasks") + + +def test_server_error_is_unavailable(mcp: FakeMcp) -> None: + mcp.status = 500 + with pytest.raises(McpUnavailable): + _client(mcp).call("list_tasks") + + +def test_other_http_status_is_mcp_error(mcp: FakeMcp) -> None: + mcp.status = 404 + with pytest.raises(McpError) as info: + _client(mcp).call("list_tasks") + assert not isinstance(info.value, McpUnavailable) + + +def test_connection_refused_is_unavailable() -> None: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + with pytest.raises(McpUnavailable): + McpClient(f"http://127.0.0.1:{port}/mcp", "t", timeout=1).call("x") + + +def test_timeout_is_unavailable(mcp: FakeMcp) -> None: + mcp.delay = 1.0 + with pytest.raises(McpUnavailable): + _client(mcp, timeout=0.2).call("list_tasks") + + +def test_tool_error_flag_raises(mcp: FakeMcp) -> None: + mcp.mode = "tool-error" + mcp.tools["close_task"] = lambda _a: "error: evidence: required" + with pytest.raises(McpToolError, match="evidence"): + _client(mcp).call("close_task", {"id": "x"}) + + +def test_rpc_error_raises(mcp: FakeMcp) -> None: + mcp.mode = "rpc-error" + with pytest.raises(McpError, match="nope"): + _client(mcp).call("list_tasks") + + +def test_garbage_body_raises(mcp: FakeMcp) -> None: + mcp.mode = "garbage" + with pytest.raises(McpError, match="unparseable"): + _client(mcp).call("list_tasks") + + +def test_sse_without_data_raises(mcp: FakeMcp) -> None: + mcp.mode = "no-data" + with pytest.raises(McpError, match="unparseable"): + _client(mcp).call("list_tasks") + + +def test_stateless_server_needs_no_session(mcp: FakeMcp) -> None: + mcp.stateless = True + mcp.tools["x"] = lambda _a: "ok" + assert _client(mcp).call("x") == "ok" + assert mcp.requests[-1]["session"] is None + + +def test_unknown_session_is_mcp_error(mcp: FakeMcp) -> None: + mcp.tools["x"] = lambda _a: "ok" + client = _client(mcp) + client.call("x") + client._session = "stale" + with pytest.raises(McpError, match="no session"): + client.call("x") + + +def test_close_deletes_session_once(mcp: FakeMcp) -> None: + mcp.tools["x"] = lambda _a: "ok" + client = _client(mcp) + client.close() # nothing opened: no request + assert mcp.requests == [] + client.call("x") + client.close() + client.close() + assert [r["method"] for r in mcp.requests].count("DELETE") == 1 + + +def test_close_swallows_errors(mcp: FakeMcp) -> None: + mcp.tools["x"] = lambda _a: "ok" + client = _client(mcp) + client.call("x") + mcp.stop() + client.close() # must not raise + + +def test_context_manager_closes(mcp: FakeMcp) -> None: + mcp.tools["x"] = lambda _a: "ok" + with _client(mcp) as client: + client.call("x") + assert mcp.requests[-1]["method"] == "DELETE" + + +def test_non_object_json_is_unparseable() -> None: + with pytest.raises(McpError, match="unparseable"): + _parse_reply("application/json", "[1, 2]") + + +def test_non_http_url_is_refused() -> None: + with pytest.raises(McpError, match="unsupported"): + McpClient("file:///etc/passwd", "t") diff --git a/todo_hooks/tests/test_closer.py b/todo_hooks/tests/test_closer.py new file mode 100644 index 0000000..3c0e447 --- /dev/null +++ b/todo_hooks/tests/test_closer.py @@ -0,0 +1,175 @@ +"""Tests for the post-commit closer.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from todo_hooks import closer +from todo_hooks.tests._fakes import detail, task_client +from todo_hooks.tests.conftest import git + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + from todo_hooks.paths import Paths + +from todo_hooks.tests._closer_helpers import ID +from todo_hooks.tests._closer_helpers import setup as _setup + + +def test_no_trailers_never_contacts_server( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = make_repo() + ctx, lines = make_ctx(None, root) + ctx.make_client = lambda: pytest.fail("client built without trailers") # type: ignore[assignment,return-value] + assert closer.run(ctx) == 0 + assert lines == [] + + +def test_not_a_git_repo_is_silent( + tmp_path: Path, make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + plain = tmp_path / "plain" + plain.mkdir() + ctx, lines = make_ctx(None, plain) + assert closer.run(ctx) == 0 + assert lines == [] + + +def test_closes_on_exit_zero( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, done_check="true")) + root, ctx, lines = _setup(make_repo, make_ctx, paths, client) + assert closer.run(ctx) == 0 + sha = git(root, "rev-parse", "--short=7", "HEAD") + assert client.calls[-1] == ( + "close_task", + {"id": ID, "evidence": f"{sha} done_check exit 0"}, + ) + assert lines == ["todo-hooks: closed 0123abcd"] + + +def test_runs_from_repo_root_even_in_subdir( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + check = "test -f f.txt" + client = task_client(detail(ID, done_check=check)) + root, ctx, _ = _setup(make_repo, make_ctx, paths, client, check=check) + (root / "sub").mkdir() + ctx.cwd = root / "sub" + closer.run(ctx) + assert "close_task" in client.tools() + + +def test_nonzero_leaves_open( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, done_check="exit 4")) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, check="exit 4") + closer.run(ctx) + assert "close_task" not in client.tools() + assert lines == ["todo-hooks: 0123abcd left open: done_check exit 4"] + + +def test_timeout_leaves_open( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(closer, "CHECK_TIMEOUT", 0.3) + client = task_client(detail(ID, done_check="sleep 20")) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, check="sleep 20") + closer.run(ctx) + assert "close_task" not in client.tools() + assert "timed out" in lines[0] + + +def test_untrusted_check_is_not_executed( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, + tmp_path: Path, +) -> None: + marker = tmp_path / "ran" + check = f"touch {marker}" + client = task_client(detail(ID, done_check=check)) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, check=check, trust=False) + closer.run(ctx) + assert not marker.exists() + assert "close_task" not in client.tools() + assert lines == [ + "todo-hooks: 0123abcd untrusted done_check; run: python3 -m todo_hooks trust 0123abcd" + ] + + +def test_repo_mismatch_is_not_executed( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, repo="kuhyx/other")) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + closer.run(ctx) + assert "close_task" not in client.tools() + assert "belongs to kuhyx/other, not kuhyx/demo" in lines[0] + + +def test_note_without_repo_is_mismatch( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + raw = f"id: {ID}\nstatus: todo\ndone_check: true\n---\nT" + client = task_client(raw) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + closer.run(ctx) + assert "close_task" not in client.tools() + assert "no repo" in lines[0] + + +def test_manual_note_is_listed_not_closed( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, text="# T\n\nCheck: manual")) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + closer.run(ctx) + assert "close_task" not in client.tools() + assert lines == ["todo-hooks: 0123abcd close by hand (Check: manual)"] + + +def test_note_without_done_check( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, done_check=None)) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, check=None) + closer.run(ctx) + assert lines == ["todo-hooks: 0123abcd has no done_check; close by hand"] + + +def test_already_done_is_skipped( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID, status="done")) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + closer.run(ctx) + assert "close_task" not in client.tools() + assert lines == ["todo-hooks: 0123abcd already done"] diff --git a/todo_hooks/tests/test_closer_failures.py b/todo_hooks/tests/test_closer_failures.py new file mode 100644 index 0000000..6f4b826 --- /dev/null +++ b/todo_hooks/tests/test_closer_failures.py @@ -0,0 +1,101 @@ +"""Tests for the closer when the server or the close call misbehaves.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks import closer +from todo_hooks.client import McpAuthError, McpToolError, McpUnavailable +from todo_hooks.tests._closer_helpers import ID, ID2, MSG +from todo_hooks.tests._closer_helpers import setup as _setup +from todo_hooks.tests._fakes import FakeClient, detail, task_client +from todo_hooks.trust import add_trusted + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + from todo_hooks.paths import Paths + + +def test_unknown_id_is_reported_and_next_continues( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client(detail(ID2)) + msg = "feat\n\nTodo: 0123abcd\nTodo: 4567ef01" + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, message=msg) + closer.run(ctx) + assert lines[0].startswith("todo-hooks: 0123abcd not found") + assert lines[1] == "todo-hooks: closed 4567ef01" + + +def test_server_down_warns_once_and_exits_zero( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = FakeClient({"get_task": McpUnavailable("connection refused")}) + msg = "feat\n\nTodo: 0123abcd\nTodo: 4567ef01" + _, ctx, lines = _setup(make_repo, make_ctx, paths, client, message=msg) + assert closer.run(ctx) == 0 + assert client.tools() == ["get_task"] + assert lines == [ + "todo-hooks: warning: todo server unavailable (connection refused); nothing closed" + ] + + +def test_auth_failure_warns( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = FakeClient({"get_task": McpAuthError("bad token")}) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + assert closer.run(ctx) == 0 + assert len(lines) == 1 + assert "warning" in lines[0] + assert "bad token" in lines[0] + + +def test_client_construction_failure_warns( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + _, ctx, lines = _setup(make_repo, make_ctx, paths, FakeClient()) + + def boom() -> FakeClient: + raise McpAuthError("no token") + + ctx.make_client = boom # type: ignore[assignment] + assert closer.run(ctx) == 0 + assert "no token" in lines[0] + + +def test_close_refused_warns( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + client = task_client( + detail(ID), close_task=McpToolError("error: evidence: required") + ) + _, ctx, lines = _setup(make_repo, make_ctx, paths, client) + closer.run(ctx) + assert lines == ["todo-hooks: 0123abcd close failed: error: evidence: required"] + + +def test_no_remote_uses_toplevel_basename( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + root = make_repo("solo", message=MSG) + add_trusted(paths.trust_file, "true", "t") + client = task_client(detail(ID, repo="kuhyx/solo")) + ctx, _ = make_ctx(client, root) + closer.run(ctx) + assert "close_task" in client.tools() diff --git a/todo_hooks/tests/test_gitrepo_checkrun.py b/todo_hooks/tests/test_gitrepo_checkrun.py new file mode 100644 index 0000000..9b3437b --- /dev/null +++ b/todo_hooks/tests/test_gitrepo_checkrun.py @@ -0,0 +1,142 @@ +"""Tests for git helpers and the done_check runner.""" + +from __future__ import annotations + +import stat +import time +from typing import TYPE_CHECKING + +import pytest + +from todo_hooks.checkrun import CHECK_TIMEOUT, run_check +from todo_hooks.gitrepo import ( + expected_repo, + head_sha, + head_todo_ids, + origin_slug, + parse_slug, + repo_name, + toplevel, +) +from todo_hooks.tests.conftest import git + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + +@pytest.mark.parametrize( + ("url", "slug"), + [ + ("git@github.com:kuhyx/utils.git", ("kuhyx", "utils")), + ("https://github.com/kuhyx/utils", ("kuhyx", "utils")), + ("https://github.com/kuhyx/utils.git/", ("kuhyx", "utils")), + ("ssh://git@github.com/kuhyx/utils.git", ("kuhyx", "utils")), + ("https://user:tok@github.com/o/n.git", ("o", "n")), + ("/srv/git/local.git", None), + ("nonsense", None), + ("", None), + ], +) +def test_parse_slug(url: str, slug: tuple[str, str] | None) -> None: + assert parse_slug(url) == slug + + +def test_toplevel_and_non_repo(make_repo: Callable[..., Path], tmp_path: Path) -> None: + root = make_repo() + (root / "sub").mkdir() + assert toplevel(root / "sub") == root.resolve() + assert toplevel(tmp_path / "nowhere-at-all") is None + plain = tmp_path / "plain" + plain.mkdir() + assert toplevel(plain) is None + + +def test_repo_name_from_origin_else_basename(make_repo: Callable[..., Path]) -> None: + with_origin = make_repo("dirname", origin="git@github.com:kuhyx/real-name.git") + assert repo_name(with_origin) == "real-name" + assert origin_slug(with_origin) == ("kuhyx", "real-name") + assert expected_repo(with_origin) == "kuhyx/real-name" + bare = make_repo("solo") + assert origin_slug(bare) is None + assert repo_name(bare) == "solo" + assert expected_repo(bare) == "kuhyx/solo" + odd = make_repo("odd", origin="/srv/odd.git") + assert repo_name(odd) == "odd" + + +def test_head_sha_is_short(make_repo: Callable[..., Path]) -> None: + root = make_repo() + assert git(root, "rev-parse", "HEAD").startswith(head_sha(root)) + assert len(head_sha(root)) == 7 + + +def test_head_todo_ids(make_repo: Callable[..., Path]) -> None: + root = make_repo( + message=( + "subject\n\nbody\n\n" + "Todo: 0123abcd\nTodo: 4567ef01 extra words\nTodo: ../../etc\n" + "Signed-off-by: x " + ) + ) + assert head_todo_ids(root) == ["0123abcd", "4567ef01"] + + +def test_head_todo_ids_none(make_repo: Callable[..., Path]) -> None: + assert head_todo_ids(make_repo()) == [] + + +def test_run_check_exit_codes(tmp_path: Path) -> None: + ok = run_check("exit 0", tmp_path) + assert (ok.code, ok.timed_out, ok.passed) == (0, False, True) + bad = run_check("exit 3", tmp_path) + assert (bad.code, bad.passed) == (3, False) + assert CHECK_TIMEOUT == 55 + + +def test_run_check_uses_cwd_and_bash(tmp_path: Path) -> None: + (tmp_path / "marker").write_text("x") + assert run_check("test -f marker && [[ a == a ]]", tmp_path).passed + + +def test_run_check_timeout_kills_group(tmp_path: Path) -> None: + start = time.monotonic() + result = run_check("sleep 30 & wait", tmp_path, timeout=0.3) + assert result.timed_out + assert result.code is None + assert not result.passed + assert time.monotonic() - start < 5 + + +def test_run_check_wrapper_prefixes_command(tmp_path: Path) -> None: + log = tmp_path / "log" + wrapper = tmp_path / "wrap.sh" + wrapper.write_text(f'#!/bin/sh\necho "$1 $2" > "{log}"\nexec "$@"\n') + wrapper.chmod(wrapper.stat().st_mode | stat.S_IEXEC) + assert run_check("exit 0", tmp_path, wrapper=wrapper).passed + assert log.read_text().strip() == "bash -c" + + +def test_run_check_missing_shell_cwd(tmp_path: Path) -> None: + result = run_check("true", tmp_path / "gone") + assert result.code is None + assert not result.timed_out + assert not result.passed + + +def test_run_check_strips_git_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("GIT_INDEX_FILE", "/nonexistent") + monkeypatch.setenv("KEEP_ME", "1") + assert run_check( + 'test -z "$GIT_INDEX_FILE" && test "$KEEP_ME" = 1', tmp_path + ).passed + + +def test_describe(tmp_path: Path) -> None: + assert run_check("exit 5", tmp_path).describe() == "exit 5" + assert run_check("true", tmp_path / "gone").describe() == "could not start" + assert ( + run_check("sleep 9", tmp_path, timeout=0.1).describe() == "timed out after 55s" + ) diff --git a/todo_hooks/tests/test_import_pr.py b/todo_hooks/tests/test_import_pr.py new file mode 100644 index 0000000..3c5917d --- /dev/null +++ b/todo_hooks/tests/test_import_pr.py @@ -0,0 +1,128 @@ +"""Tests for the PR `## Todo:` importer.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks import gh, import_pr +from todo_hooks.client import McpUnavailable +from todo_hooks.tests._fakes import FakeClient, FakeGh, line, make_pr +from todo_hooks.tests._pr_helpers import pr_repo, search_client +from todo_hooks.tests.conftest import git + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + + +def test_import_adds_tasks_with_pr_reference( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + client = search_client() + ctx, out = make_ctx(client, root) + count = import_pr.import_for_commits(ctx, [sha], FakeGh({sha: [make_pr(7)]})) + assert count == 3 + adds = [a for t, a in client.calls if t == "add_task"] + assert adds[0] == { + "repo": "kuhyx/demo", + "title": "Add retry to sync", + "body": "Imported from PR #7 of kuhyx/demo.", + "done_check": "pytest -q tests/test_retry.py", + "priority": "medium", + } + assert "done_check" not in adds[1] + assert "imported 3 task(s) from PR #7" in out[-1] + + +def test_dedupe_skips_same_pr_and_title( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + existing = line("abcdef12-x", "kuhyx/demo", "# Document the flag") + client = search_client(existing) + ctx, _ = make_ctx(client, root) + count = import_pr.import_for_commits(ctx, [sha], FakeGh({sha: [make_pr(7)]})) + assert count == 2 + queries = [a["query"] for t, a in client.calls if t == "search_tasks"] + assert queries == ["PR #7"] * 3 + titles = [a["title"] for t, a in client.calls if t == "add_task"] + assert "Document the flag" not in titles + + +def test_same_pr_on_many_commits_imported_once( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 2) + shas = git(root, "rev-list", "HEAD").splitlines() + client = search_client() + ctx, _ = make_ctx(client, root) + gh_fake = FakeGh({s: [make_pr(7)] for s in shas}) + assert import_pr.import_for_commits(ctx, shas, gh_fake) == 3 + + +def test_unmerged_empty_and_bodyless_prs_are_ignored( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + client = search_client() + ctx, out = make_ctx(client, root) + prs = [ + make_pr(1, merged=None), + make_pr(2, body=None), + make_pr(3, body="nothing here"), + ] + assert import_pr.import_for_commits(ctx, [sha], FakeGh({sha: prs})) == 0 + assert client.calls == [] + assert out == [] + + +def test_non_github_origin_is_skipped( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = make_repo("local", origin="/srv/local.git") + ctx, out = make_ctx(search_client(), root) + assert import_pr.import_for_commits(ctx, ["abc"], FakeGh({})) == 0 + assert "no GitHub origin" in out[0] + + +def test_gh_failure_warns_and_stops( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + + def broken(_args: list[str]) -> str: + raise gh.GhError("gh not installed") + + ctx, out = make_ctx(search_client(), root) + assert import_pr.import_for_commits(ctx, ["a", "b"], broken) == 0 + assert out == [ + "todo-hooks: warning: gh unavailable (gh not installed); PR import skipped" + ] + + +def test_bad_gh_json_warns( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + ctx, out = make_ctx(search_client(), root) + assert import_pr.import_for_commits(ctx, ["a"], lambda _a: "{not json") == 0 + assert "PR import skipped" in out[0] + + +def test_server_down_warns_and_exits_cleanly( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + client = FakeClient({"search_tasks": McpUnavailable("refused")}) + ctx, out = make_ctx(client, root) + assert import_pr.import_for_commits(ctx, [sha], FakeGh({sha: [make_pr(7)]})) == 0 + assert out == [ + "todo-hooks: warning: todo server unavailable (refused); PR import skipped" + ] diff --git a/todo_hooks/tests/test_import_pr_commits.py b/todo_hooks/tests/test_import_pr_commits.py new file mode 100644 index 0000000..50a1da2 --- /dev/null +++ b/todo_hooks/tests/test_import_pr_commits.py @@ -0,0 +1,162 @@ +"""Tests for choosing which commits to scan, and the hook entry point.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks import import_pr +from todo_hooks.tests._fakes import FakeGh, line, make_pr +from todo_hooks.tests._pr_helpers import pr_repo, search_client +from todo_hooks.tests.conftest import git + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + import pytest + + from todo_hooks.context import Context + + +def test_commits_post_merge_uses_orig_head(make_repo: Callable[..., Path]) -> None: + root = pr_repo(make_repo, 2) + old = git(root, "rev-parse", "HEAD~2") + git(root, "update-ref", "ORIG_HEAD", old) + shas = import_pr.commits_post_merge(root) + assert shas == git(root, "rev-list", "ORIG_HEAD..HEAD").splitlines() + assert len(shas) == 2 + + +def test_commits_post_merge_without_orig_head(make_repo: Callable[..., Path]) -> None: + assert import_pr.commits_post_merge(make_repo()) == [] + + +def test_commits_post_merge_is_capped( + make_repo: Callable[..., Path], monkeypatch: pytest.MonkeyPatch +) -> None: + root = pr_repo(make_repo, 3) + git(root, "update-ref", "ORIG_HEAD", git(root, "rev-parse", "HEAD~3")) + monkeypatch.setattr(import_pr, "MAX_COMMITS", 2) + assert len(import_pr.commits_post_merge(root)) == 2 + + +def test_commits_post_rewrite_only_for_rebase(make_repo: Callable[..., Path]) -> None: + root = pr_repo(make_repo, 1) + pairs = "oldsha1 newsha1\noldsha2 newsha2 extra\n\n" + assert import_pr.commits_post_rewrite(root, "rebase", pairs) == [ + "newsha1", + "newsha2", + ] + assert import_pr.commits_post_rewrite(root, "amend", pairs) == [] + assert import_pr.commits_post_rewrite(root, "", pairs) == [] + + +def test_commits_post_rewrite_without_stdin_falls_back( + make_repo: Callable[..., Path], +) -> None: + root = pr_repo(make_repo, 2) + git(root, "update-ref", "ORIG_HEAD", git(root, "rev-parse", "HEAD~2")) + assert len(import_pr.commits_post_rewrite(root, "rebase", "")) == 2 + + +def test_main_post_merge_end_to_end( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + git(root, "update-ref", "ORIG_HEAD", git(root, "rev-parse", "HEAD~1")) + client = search_client() + ctx, _ = make_ctx(client, root) + assert ( + import_pr.run(ctx, ["post-merge"], FakeGh({sha: [make_pr(9)]}), stdin="") == 0 + ) + assert client.tools().count("add_task") == 3 + + +def test_main_post_rewrite_uses_env_command( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + client = search_client() + ctx, _ = make_ctx(client, root, {"PRE_COMMIT_REWRITE_COMMAND": "rebase"}) + assert ( + import_pr.run( + ctx, ["post-rewrite"], FakeGh({sha: [make_pr(9)]}), stdin=f"old {sha}\n" + ) + == 0 + ) + assert client.tools().count("add_task") == 3 + + +def test_main_explicit_range_and_bad_mode( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + client = search_client() + ctx, out = make_ctx(client, root) + assert ( + import_pr.run( + ctx, ["--range", "HEAD~1..HEAD"], FakeGh({sha: [make_pr(9)]}), stdin="" + ) + == 0 + ) + assert client.tools().count("add_task") == 3 + assert import_pr.run(ctx, ["bogus"], FakeGh({}), stdin="") == 2 + assert "usage" in out[-1] + + +def test_not_in_git_repo_is_quiet( + tmp_path: Path, make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + plain = tmp_path / "plain" + plain.mkdir() + ctx, out = make_ctx(search_client(), plain) + assert import_pr.run(ctx, ["post-merge"], FakeGh({}), stdin="") == 0 + assert out == [] + + +def test_read_stdin_variants(monkeypatch: pytest.MonkeyPatch) -> None: + import io + import os + + read_end, write_end = os.pipe() + os.write(write_end, b"old new\n") + os.close(write_end) + monkeypatch.setattr("sys.stdin", os.fdopen(read_end, "r")) + assert import_pr.read_stdin() == "old new\n" + + idle_read, idle_write = os.pipe() + monkeypatch.setattr("sys.stdin", os.fdopen(idle_read, "r")) + assert import_pr.read_stdin(timeout=0.05) == "" + os.close(idle_write) + + monkeypatch.setattr("sys.stdin", io.StringIO("no fileno")) + assert import_pr.read_stdin() == "" + + +def test_run_post_rewrite_reads_real_stdin_when_not_given( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = pr_repo(make_repo, 1) + monkeypatch.setattr(import_pr, "read_stdin", lambda: "") + ctx, out = make_ctx(search_client(), root, {}) + assert import_pr.run(ctx, ["post-rewrite", "amend"], FakeGh({})) == 0 + assert out == [] + + +def test_nothing_new_to_import_is_silent( + make_repo: Callable[..., Path], make_ctx: Callable[..., tuple[Context, list[str]]] +) -> None: + root = pr_repo(make_repo, 1) + sha = git(root, "rev-parse", "HEAD") + existing = "\n".join( + line("abcdef12-x", "kuhyx/demo", f"# {t}") + for t in ("Add retry to sync", "Document the flag", "Star bullet") + ) + ctx, out = make_ctx(search_client(existing), root) + assert import_pr.import_for_commits(ctx, [sha], FakeGh({sha: [make_pr(7)]})) == 0 + assert out == [] diff --git a/todo_hooks/tests/test_main.py b/todo_hooks/tests/test_main.py new file mode 100644 index 0000000..4f42967 --- /dev/null +++ b/todo_hooks/tests/test_main.py @@ -0,0 +1,173 @@ +"""Tests for the CLI dispatcher and the trust subcommand.""" + +from __future__ import annotations + +import sys +from typing import TYPE_CHECKING + +import pytest + +from todo_hooks import __main__ as cli +from todo_hooks import trust_cmd +from todo_hooks.client import McpToolError, McpUnavailable +from todo_hooks.tests._fakes import ( + TOKEN, + FakeClient, + FakeMcp, + detail, + line, + task_client, +) +from todo_hooks.trust import check_hash, is_trusted + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + from todo_hooks.paths import Paths + +ID = "0123abcd-0000-4000-8000-000000000001" +ID2 = "4567ef01-0000-4000-8000-000000000002" + + +def test_trust_one( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + client = task_client(detail(ID, done_check="make test")) + ctx, out = make_ctx(client, tmp_path) + assert trust_cmd.run(ctx, ["0123abcd"]) == 0 + assert is_trusted(paths.trust_file, "make test") + assert out == [ + f"todo-hooks: trusted 0123abcd (sha256 {check_hash('make test')[:12]})" + ] + + +def test_trust_one_failures( + make_ctx: Callable[..., tuple[Context, list[str]]], tmp_path: Path +) -> None: + ctx, out = make_ctx(task_client(detail(ID, done_check=None)), tmp_path) + assert trust_cmd.run(ctx, ["0123abcd"]) == 1 + assert "has no done_check" in out[-1] + assert trust_cmd.run(ctx, ["ffffffff"]) == 1 + assert "not found" in out[-1] + assert trust_cmd.run(ctx, []) == 2 + assert "usage" in out[-1] + assert trust_cmd.run(ctx, ["a", "b"]) == 2 + + +def test_trust_server_down( + make_ctx: Callable[..., tuple[Context, list[str]]], tmp_path: Path +) -> None: + ctx, out = make_ctx(FakeClient({"get_task": McpUnavailable("refused")}), tmp_path) + assert trust_cmd.run(ctx, ["0123abcd"]) == 1 + assert "unavailable" in out[-1] + + +def test_seed_open_trusts_all_open_with_checks( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + client = task_client( + detail(ID, done_check="first"), + detail(ID2, done_check=None), + list_tasks="\n".join([line(ID), line(ID2), "garbage"]), + ) + ctx, out = make_ctx(client, tmp_path) + assert trust_cmd.run(ctx, ["--seed-open"]) == 0 + assert is_trusted(paths.trust_file, "first") + assert out[-1] == "todo-hooks: seeded 1 new check(s) from 2 open note(s)" + # Re-seeding is idempotent. + assert trust_cmd.run(ctx, ["--seed-open"]) == 0 + assert out[-1] == "todo-hooks: seeded 0 new check(s) from 2 open note(s)" + assert client.calls[0] == ("list_tasks", {"status": "open"}) + + +def test_seed_open_tool_error( + make_ctx: Callable[..., tuple[Context, list[str]]], tmp_path: Path +) -> None: + ctx, out = make_ctx(FakeClient({"list_tasks": McpToolError("error: x")}), tmp_path) + assert trust_cmd.run(ctx, ["--seed-open"]) == 1 + assert "error: x" in out[-1] + + +def test_main_usage_and_unknown(capsys: pytest.CaptureFixture[str]) -> None: + assert cli.main([]) == 2 + assert cli.main(["wat"]) == 2 + assert "usage: python3 -m todo_hooks" in capsys.readouterr().out + + +def test_main_closer_end_to_end_against_fake_server( + mcp: FakeMcp, + make_repo: Callable[..., Path], + paths: Paths, + capsys: pytest.CaptureFixture[str], +) -> None: + root = make_repo( + "demo", origin="git@github.com:kuhyx/demo.git", message="x\n\nTodo: 0123abcd" + ) + mcp.tools["get_task"] = lambda _a: detail(ID, done_check="true") + closed: list[dict[str, object]] = [] + mcp.tools["close_task"] = lambda a: closed.append(a) or "closed" + env = { + "TODO_MCP_URL": mcp.url, + "TODO_MCP_TOKEN": TOKEN, + "TODO_HOOKS_TRUST_FILE": str(paths.trust_file), + } + assert cli.main(["trust", "0123abcd"], env=env, cwd=root) == 0 + assert cli.main(["closer"], env=env, cwd=root) == 0 + assert closed[0]["id"] == ID + assert "closed 0123abcd" in capsys.readouterr().out + + +def test_main_closer_without_token_fails_open( + make_repo: Callable[..., Path], paths: Paths, capsys: pytest.CaptureFixture[str] +) -> None: + root = make_repo("demo", message="x\n\nTodo: 0123abcd") + env = {"TODO_MCP_TOKEN_FILE": str(paths.token_file)} + assert cli.main(["closer"], env=env, cwd=root) == 0 + assert "warning" in capsys.readouterr().out + + +def test_main_sweep_flags( + mcp: FakeMcp, + paths: Paths, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + mcp.tools["list_tasks"] = lambda _a: "no tasks match" + env = {"TODO_MCP_URL": mcp.url, "TODO_MCP_TOKEN": TOKEN} + report = tmp_path / "r.txt" + args = [ + "sweep", + "--report", + str(report), + "--dry-run", + "--trust-file", + str(tmp_path / "t"), + ] + assert cli.main(args, env=env, cwd=tmp_path) == 0 + assert report.read_text().startswith("# dry-run: 0 open notes") + assert "dry-run" in capsys.readouterr().out + assert cli.main(["sweep", "--bogus"], env=env, cwd=tmp_path) == 2 + + +def test_main_import_pr_dispatch( + make_repo: Callable[..., Path], + paths: Paths, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = make_repo("demo", origin="git@github.com:kuhyx/demo.git") + monkeypatch.setattr("todo_hooks.gh.run_gh", lambda _a: "[]") + env = {"TODO_MCP_TOKEN_FILE": str(paths.token_file)} + assert cli.main(["import-pr", "post-merge"], env=env, cwd=root, stdin="") == 0 + + +def test_module_entrypoint_runs(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sys.argv", ["todo_hooks"]) + monkeypatch.delitem(sys.modules, "todo_hooks.__main__", raising=False) + import runpy + + with pytest.raises(SystemExit) as info: + runpy.run_module("todo_hooks", run_name="__main__", alter_sys=True) + assert info.value.code == 2 diff --git a/todo_hooks/tests/test_notes_trust.py b/todo_hooks/tests/test_notes_trust.py new file mode 100644 index 0000000..fd3b37e --- /dev/null +++ b/todo_hooks/tests/test_notes_trust.py @@ -0,0 +1,127 @@ +"""Tests for note parsing, the trust store and path defaults.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +import pytest + +from todo_hooks.notes import clean_title, parse_detail, parse_list +from todo_hooks.paths import Paths +from todo_hooks.tests._fakes import detail, line +from todo_hooks.trust import add_trusted, check_hash, is_trusted, load_trusted + +ID = "0123abcd-0000-4000-8000-000000000001" + + +def test_parse_list_rows_and_empty() -> None: + text = line(ID, title="# First") + "\n" + line("ffff0000-x", "kuhyx/b", "Second") + rows = parse_list(text) + assert [(r.id8, r.repo, r.title) for r in rows] == [ + ("0123abcd", "kuhyx/demo", "First"), + ("ffff0000", "kuhyx/b", "Second"), + ] + assert parse_list("no tasks match") == [] + assert parse_list("garbage line\n\n") == [] + + +def test_parse_detail_fields() -> None: + task = parse_detail(detail(ID, done_check="make test", text="# T\n\nbody line")) + assert task.id == ID + assert task.status == "todo" + assert task.repo == "kuhyx/demo" + assert task.done_check == "make test" + assert task.title == "T" + assert task.text.endswith("body line") + assert not task.is_manual + + +def test_parse_detail_without_optional_fields() -> None: + raw = f"id: {ID}\nstatus: todo\n---\nTitle only" + task = parse_detail(raw) + assert task.repo is None + assert task.done_check is None + assert task.title == "Title only" + + +def test_parse_detail_multiline_done_check() -> None: + raw = f"id: {ID}\nstatus: todo\ndone_check: a &&\n b\nby: claude\n---\nT" + assert parse_detail(raw).done_check == "a &&\n b" + + +def test_parse_detail_without_separator() -> None: + task = parse_detail(f"id: {ID}\nstatus: todo") + assert task.text == "" + + +@pytest.mark.parametrize( + ("body", "manual"), + [ + ("T\n\nCheck: manual", True), + ("T\n\n check: Manual \nmore", True), + ("T\n\nCheck: manual-ish", False), + ("T\n\nNot Check: manual", False), + ], +) +def test_manual_marker(body: str, manual: bool) -> None: + assert parse_detail(detail(ID, text=body)).is_manual is manual + + +def test_clean_title() -> None: + assert clean_title("# Hello ") == "Hello" + assert clean_title("Hello") == "Hello" + + +def test_check_hash_is_sha256_of_stripped_text() -> None: + expected = hashlib.sha256(b"make test").hexdigest() + assert check_hash(" make test\n") == expected + + +def test_trust_roundtrip(tmp_path: Path) -> None: + file = tmp_path / "sub" / "trusted" + assert load_trusted(file) == set() + assert not is_trusted(file, "true") + assert add_trusted(file, "true", "0123abcd kuhyx/demo") is True + assert add_trusted(file, "true", "again") is False + assert is_trusted(file, "true") + assert not is_trusted(file, "false") + assert check_hash("true") in file.read_text() + + +def test_trust_file_ignores_comments_and_blank(tmp_path: Path) -> None: + file = tmp_path / "t" + file.write_text( + f"# header\n\n{check_hash('x')} # note\n {check_hash('y').upper()}\n" + ) + assert load_trusted(file) == {check_hash("x"), check_hash("y")} + + +def test_paths_default_and_env_overrides(tmp_path: Path) -> None: + capped = tmp_path / ".claude" / "scripts" / "capped.sh" + capped.parent.mkdir(parents=True) + capped.write_text("#!/bin/sh\n") + base = Paths.default({}, tmp_path) + assert base.token_file == tmp_path / ".config/todo-mcp/client-token" + assert base.trust_file == tmp_path / ".config/todo-hooks/trusted-checks" + assert base.report_file == tmp_path / ".local/state/todo-drift/report.txt" + assert base.src_root == tmp_path / "src" + assert base.capped == capped + env = { + "TODO_HOOKS_TRUST_FILE": "/t", + "TODO_HOOKS_REPORT": "/r", + "TODO_HOOKS_SRC_ROOT": "/s", + "TODO_MCP_TOKEN_FILE": "/k", + } + over = Paths.default(env, tmp_path / "nohome") + assert (over.trust_file, over.report_file, over.src_root) == ( + Path("/t"), + Path("/r"), + Path("/s"), + ) + assert over.token_file == Path("/k") + assert over.capped is None + + +def test_parse_detail_ignores_text_before_metadata() -> None: + assert parse_detail(f"stray line\nid: {ID}\n---\nT").id == ID diff --git a/todo_hooks/tests/test_pr_parse_gh.py b/todo_hooks/tests/test_pr_parse_gh.py new file mode 100644 index 0000000..09701fe --- /dev/null +++ b/todo_hooks/tests/test_pr_parse_gh.py @@ -0,0 +1,73 @@ +"""Tests for the PR Todo-section parser and the gh wrapper.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from todo_hooks import gh +from todo_hooks.pr_parse import TodoItem, parse_todo_items +from todo_hooks.tests._fakes import BODY, FakeGh, make_pr + +if TYPE_CHECKING: + from pathlib import Path + + +def test_parse_items() -> None: + assert parse_todo_items(BODY) == [ + TodoItem("Add retry to sync", "pytest -q tests/test_retry.py"), + TodoItem("Document the flag", None), + TodoItem("Star bullet", "true"), + ] + + +@pytest.mark.parametrize( + "body", + ["", "no heading", "## Todos\n- x", "## Todo:\n\nprose only\n- late item"], +) +def test_parse_items_none(body: str) -> None: + assert parse_todo_items(body) == [] + + +def test_parse_items_variants() -> None: + body = "### todo\n- | done_check: x\n- Spaced | done_check: a | b \n" + assert parse_todo_items(body) == [TodoItem("Spaced", "a | b")] + + +def test_parse_items_stops_at_next_heading_and_survives_none() -> None: + assert parse_todo_items("## Todo:\n- one\n# Next\n- two") == [TodoItem("one", None)] + + +def test_run_gh_success_failure_and_missing( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fake = tmp_path / "gh" + fake.write_text( + '#!/bin/sh\nif [ "$1" = fail ]; then echo oops >&2; exit 1; fi\necho "[]"\n' + ) + fake.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:/usr/bin") + assert gh.run_gh(["ok"]) == "[]\n" + with pytest.raises(gh.GhError, match="oops"): + gh.run_gh(["fail"]) + monkeypatch.setenv("PATH", str(tmp_path / "empty")) + with pytest.raises(gh.GhError, match="not installed"): + gh.run_gh(["ok"]) + + +def test_run_gh_timeout(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + fake = tmp_path / "gh" + fake.write_text("#!/bin/sh\nsleep 20\n") + fake.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:/usr/bin") + monkeypatch.setattr(gh, "GH_TIMEOUT", 0.3) + with pytest.raises(gh.GhError, match="timed out"): + gh.run_gh(["x"]) + + +def test_merged_prs_parses_fields() -> None: + fake = FakeGh({"abc": [make_pr(5), {"number": 6}]}) + prs = gh.merged_prs("kuhyx", "demo", "abc", fake) + assert [(p.number, p.merged) for p in prs] == [(5, True), (6, False)] + assert fake.calls[0] == ["api", "repos/kuhyx/demo/commits/abc/pulls"] diff --git a/todo_hooks/tests/test_sweep.py b/todo_hooks/tests/test_sweep.py new file mode 100644 index 0000000..fc43cf5 --- /dev/null +++ b/todo_hooks/tests/test_sweep.py @@ -0,0 +1,247 @@ +"""Tests for the nightly drift sweep.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks import sweep +from todo_hooks.checkrun import CheckResult +from todo_hooks.client import McpToolError, McpUnavailable +from todo_hooks.tests._fakes import FakeClient, detail, line +from todo_hooks.trust import add_trusted + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from todo_hooks.context import Context + from todo_hooks.paths import Paths + +A = "aaaaaaaa-0000-4000-8000-000000000001" +B = "bbbbbbbb-0000-4000-8000-000000000002" +C = "cccccccc-0000-4000-8000-000000000003" +D = "dddddddd-0000-4000-8000-000000000004" +E = "eeeeeeee-0000-4000-8000-000000000005" +F = "ffffffff-0000-4000-8000-000000000006" + +Fixture = "tuple[Context, list[str], FakeClient]" + + +def _client(*tasks: tuple[str, str]) -> FakeClient: + """`tasks` are (list line, detail) pairs.""" + details = {d.splitlines()[0].removeprefix("id: ")[:8]: d for _, d in tasks} + return FakeClient( + { + "list_tasks": "\n".join(row for row, _ in tasks) or "no tasks match", + "get_task": lambda args: details[args["id"]], + } + ) + + +def _task(task_id: str, repo: str, title: str, **kw: object) -> tuple[str, str]: + return ( + line(task_id, repo, f"# {title}"), + detail(task_id, repo=repo, text=f"# {title}\n\nbody", **kw), # type: ignore[arg-type] + ) + + +def _runner(results: dict[str, CheckResult]) -> Callable[..., CheckResult]: + seen: list[tuple[str, Path, float, Path | None]] = [] + + def run( + command: str, cwd: Path, *, timeout: float = 55, wrapper: Path | None = None + ) -> CheckResult: + seen.append((command, cwd, timeout, wrapper)) + return results[command] + + run.seen = seen # type: ignore[attr-defined] + return run + + +def test_local_repos_matches_origin_not_dirname( + make_repo: Callable[..., Path], paths: Paths +) -> None: + odd = make_repo( + "weird-dir", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + plain = make_repo("plain", parent=paths.src_root) + (paths.src_root / "not-a-repo").mkdir() + (paths.src_root / "file.txt").write_text("x") + found = sweep.local_repos(paths.src_root) + assert found["kuhyx/demo"] == odd + assert found["kuhyx/plain"] == plain + assert "kuhyx/not-a-repo" not in found + + +def test_local_repos_missing_root(tmp_path: Path) -> None: + assert sweep.local_repos(tmp_path / "none") == {} + + +def test_sweep_reports_only_passing( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + root = make_repo( + "weird", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + for check in ("pass-cmd", "fail-cmd", "slow-cmd", "nocmd-untrusted", "manual-cmd"): + add_trusted(paths.trust_file, check, "t") + client = _client( + _task(A, "kuhyx/demo", "Passes", done_check="pass-cmd"), + _task(B, "kuhyx/demo", "Fails", done_check="fail-cmd"), + _task(C, "kuhyx/demo", "Slow", done_check="slow-cmd"), + _task(D, "kuhyx/demo", "Untrusted", done_check="evil"), + ( + line(E, "kuhyx/demo", "# Manual"), + detail(E, text="# Manual\n\nCheck: manual", done_check="manual-cmd"), + ), + _task(F, "kuhyx/absent", "Not local", done_check="pass-cmd"), + ) + runner = _runner( + { + "pass-cmd": CheckResult(0, False), + "fail-cmd": CheckResult(2, False), + "slow-cmd": CheckResult(None, True), + } + ) + ctx, out = make_ctx(client, root) + assert sweep.run(ctx, runner=runner) == 0 + report = paths.report_file.read_text().splitlines() + assert report[0] == "aaaaaaaa kuhyx/demo Passes" + assert report[1] == "# close by hand: eeeeeeee kuhyx/demo Manual" + assert report[-1] == ( + "# swept 6 open notes: 1 pass, 1 fail, 1 timeout, 1 untrusted, 1 manual, 1 no local repo" + ) + assert len(report) == 3 + assert [c[0] for c in runner.seen] == ["pass-cmd", "fail-cmd", "slow-cmd"] # type: ignore[attr-defined] + assert all(c[1] == root and c[2] == 55 for c in runner.seen) # type: ignore[attr-defined] + assert out[-1] == report[-1] + + +def test_sweep_skips_done_and_checkless( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + root = make_repo( + "demo", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + client = _client( + _task(A, "kuhyx/demo", "Done", status="done", done_check="x"), + _task(B, "kuhyx/demo", "NoCheck", done_check=None), + ) + ctx, _ = make_ctx(client, root) + sweep.run(ctx, runner=_runner({})) + report = paths.report_file.read_text().splitlines() + assert report == [ + "# swept 2 open notes: 0 pass, 0 fail, 0 timeout, 0 untrusted, 0 manual, 0 no local repo" + ] + + +def test_sweep_uses_capped_wrapper( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, + tmp_path: Path, +) -> None: + root = make_repo( + "demo", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + add_trusted(paths.trust_file, "c", "t") + client = _client(_task(A, "kuhyx/demo", "T", done_check="c")) + wrapper = tmp_path / "capped.sh" + object.__setattr__(paths, "capped", wrapper) + runner = _runner({"c": CheckResult(0, False)}) + ctx, _ = make_ctx(client, root) + sweep.run(ctx, runner=runner) + assert runner.seen[0][3] == wrapper # type: ignore[attr-defined] + + +def test_dry_run_runs_nothing( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + root = make_repo( + "demo", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + add_trusted(paths.trust_file, "c", "t") + client = _client( + _task(A, "kuhyx/demo", "Would", done_check="c"), + _task(B, "kuhyx/demo", "Untrusted", done_check="zz"), + ) + + def explode(*_a: object, **_k: object) -> CheckResult: + raise AssertionError("dry run executed a check") + + ctx, _ = make_ctx(client, root) + assert sweep.run(ctx, dry_run=True, runner=explode) == 0 + report = paths.report_file.read_text().splitlines() + assert report[0] == "would-run aaaaaaaa kuhyx/demo Would" + assert report[-1].startswith("# dry-run: 2 open notes: 1 would run") + + +def test_sweep_empty_list( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + ctx, _ = make_ctx(_client(), tmp_path) + assert sweep.run(ctx) == 0 + assert paths.report_file.read_text().startswith("# swept 0 open notes") + + +def test_server_down_exits_one_and_keeps_old_report( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + paths.report_file.parent.mkdir(parents=True) + paths.report_file.write_text("old\n") + client = FakeClient({"list_tasks": McpUnavailable("refused")}) + ctx, out = make_ctx(client, tmp_path) + assert sweep.run(ctx) == 1 + assert paths.report_file.read_text() == "old\n" + assert "unavailable" in out[0] + + +def test_report_override_path( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + ctx, _ = make_ctx(_client(), tmp_path) + target = tmp_path / "elsewhere" / "r.txt" + assert sweep.run(ctx, report=target) == 0 + assert target.exists() + assert not paths.report_file.exists() + + +def test_note_lookup_error_skips_that_note( + make_repo: Callable[..., Path], + make_ctx: Callable[..., tuple[Context, list[str]]], + paths: Paths, +) -> None: + root = make_repo( + "demo", origin="git@github.com:kuhyx/demo.git", parent=paths.src_root + ) + client = FakeClient( + { + "list_tasks": line(A, "kuhyx/demo"), + "get_task": McpToolError("error: id: ambiguous"), + } + ) + ctx, _ = make_ctx(client, root) + assert sweep.run(ctx) == 0 + assert paths.report_file.read_text().startswith("# swept 1 open notes: 0 pass") + + +def test_main_rejects_unknown_flag( + make_ctx: Callable[..., tuple[Context, list[str]]], tmp_path: Path +) -> None: + ctx, out = make_ctx(_client(), tmp_path) + assert sweep.main(ctx, ["--nope"]) == 2 + assert "usage" in out[0] + + +def test_main_without_flags_uses_configured_paths( + make_ctx: Callable[..., tuple[Context, list[str]]], paths: Paths, tmp_path: Path +) -> None: + ctx, _ = make_ctx(_client(), tmp_path) + assert sweep.main(ctx, []) == 0 + assert paths.report_file.exists() diff --git a/todo_hooks/trust.py b/todo_hooks/trust.py new file mode 100644 index 0000000..d2f3ce8 --- /dev/null +++ b/todo_hooks/trust.py @@ -0,0 +1,46 @@ +"""Allow-list of done_check hashes that may be executed on this machine.""" + +from __future__ import annotations + +import hashlib +import os +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + + +def check_hash(done_check: str) -> str: + return hashlib.sha256(done_check.strip().encode()).hexdigest() + + +def load_trusted(path: Path) -> set[str]: + """Hashes in `path`: first token per line, ``#`` lines ignored.""" + try: + text = path.read_text(encoding="utf-8") + except OSError: + return set() + return { + row.split()[0].lower() + for row in text.splitlines() + if row.strip() and not row.lstrip().startswith("#") + } + + +def is_trusted(path: Path, done_check: str) -> bool: + return check_hash(done_check) in load_trusted(path) + + +def add_trusted(path: Path, done_check: str, label: str) -> bool: + """Append the hash plus an audit label; False if it was already listed.""" + digest = check_hash(done_check) + if digest in load_trusted(path): + return False + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + existing = path.read_text(encoding="utf-8") if path.exists() else "" + prefix = "\n" if existing and not existing.endswith("\n") else "" + note = " ".join(label.split()) + fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) + with os.fdopen(fd, "a", encoding="utf-8") as handle: + handle.write(f"{prefix}{digest} # {note}\n") + return True diff --git a/todo_hooks/trust_cmd.py b/todo_hooks/trust_cmd.py new file mode 100644 index 0000000..d5baa27 --- /dev/null +++ b/todo_hooks/trust_cmd.py @@ -0,0 +1,60 @@ +"""``trust `` and ``trust --seed-open``: approve done_checks for execution.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from todo_hooks.client import McpError, McpToolError +from todo_hooks.context import describe +from todo_hooks.notes import parse_detail, parse_list +from todo_hooks.trust import add_trusted, check_hash + +if TYPE_CHECKING: + from todo_hooks.client import TaskClient + from todo_hooks.context import Context + +USAGE = "usage: python3 -m todo_hooks trust | --seed-open" + + +def _trust_one(ctx: Context, client: TaskClient, task_id: str) -> int: + try: + task = parse_detail(client.call("get_task", {"id": task_id})) + except McpToolError as err: + ctx.say(f"{task_id} not found: {err}") + return 1 + if not task.done_check: + ctx.say(f"{task_id} has no done_check; nothing to trust") + return 1 + add_trusted(ctx.paths.trust_file, task.done_check, f"{task_id[:8]} {task.repo}") + ctx.say(f"trusted {task_id[:8]} (sha256 {check_hash(task.done_check)[:12]})") + return 0 + + +def _seed_open(ctx: Context, client: TaskClient) -> int: + rows = parse_list(client.call("list_tasks", {"status": "open"})) + added = 0 + for row in rows: + task = parse_detail(client.call("get_task", {"id": row.id8})) + if task.done_check: + label = f"{row.id8} {row.repo}" + added += add_trusted(ctx.paths.trust_file, task.done_check, label) + ctx.say(f"seeded {added} new check(s) from {len(rows)} open note(s)") + return 0 + + +def run(ctx: Context, args: list[str]) -> int: + """Trust one note's check, or every currently open one.""" + if len(args) != 1: + ctx.out(USAGE) + return 2 + try: + client = ctx.make_client() + try: + if args[0] == "--seed-open": + return _seed_open(ctx, client) + return _trust_one(ctx, client, args[0]) + finally: + client.close() + except McpError as err: + ctx.say(f"trust failed: {describe(err)}") + return 1