From bb142d05e436a57740726ac234e2c15f3c2642d3 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 19:51:45 -0400 Subject: [PATCH] feat: git worktree module Adds hqt.git.worktree with slugify, is_git_repo, WorktreeState, create_worktree, add_worktree_for_branch, worktree_state, and remove_worktree; includes 25 real-git integration tests. Co-Authored-By: Claude Fable 5 --- src/hqt/git/__init__.py | 1 + src/hqt/git/worktree.py | 210 +++++++++++++++++++++++++++++ tests/test_worktree.py | 284 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 495 insertions(+) create mode 100644 src/hqt/git/__init__.py create mode 100644 src/hqt/git/worktree.py create mode 100644 tests/test_worktree.py diff --git a/src/hqt/git/__init__.py b/src/hqt/git/__init__.py new file mode 100644 index 0000000..e91d518 --- /dev/null +++ b/src/hqt/git/__init__.py @@ -0,0 +1 @@ +"""Git utilities for hqt.""" diff --git a/src/hqt/git/worktree.py b/src/hqt/git/worktree.py new file mode 100644 index 0000000..c2137e0 --- /dev/null +++ b/src/hqt/git/worktree.py @@ -0,0 +1,210 @@ +"""Git worktree helpers for hqt. + +All async functions use asyncio.create_subprocess_exec in the style of +hqt.tmux.runner. Failures raise hqt.errors.ServiceError with git's stderr. +""" + +import asyncio +import logging +import re +from dataclasses import dataclass +from pathlib import Path + +from hqt.errors import ServiceError + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Pure helpers +# --------------------------------------------------------------------------- + + +def slugify(text: str) -> str: + """Lowercase; keep [a-z0-9]; collapse runs of anything else to single '-'; + strip leading/trailing '-'.""" + lowered = text.lower() + # Replace any run of non-alphanumeric characters with a single dash. + slugged = re.sub(r"[^a-z0-9]+", "-", lowered) + return slugged.strip("-") + + +# --------------------------------------------------------------------------- +# Internal subprocess helper +# --------------------------------------------------------------------------- + + +async def _git( + *args: str, + cwd: Path | None = None, +) -> tuple[int, str, str]: + """Run git with *args; return (returncode, stdout, stderr).""" + proc = await asyncio.create_subprocess_exec( + "git", + *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + ) + stdout_bytes, stderr_bytes = await proc.communicate() + return proc.returncode or 0, stdout_bytes.decode(), stderr_bytes.decode() + + +# --------------------------------------------------------------------------- +# Public async functions +# --------------------------------------------------------------------------- + + +async def is_git_repo(path: Path) -> bool: + """Return True iff path is inside a git work tree.""" + if not path.exists(): + return False + rc, _, _ = await _git("rev-parse", "--is-inside-work-tree", cwd=path) + return rc == 0 + + +@dataclass +class WorktreeState: + """Snapshot of a worktree's runtime state.""" + + exists: bool + dirty: bool + unique_commits: int + + +def _exclude_file(common_dir: Path) -> Path: + return common_dir / "info" / "exclude" + + +def _ensure_worktrees_excluded(common_dir: Path) -> None: + """Append '.worktrees/' to /info/exclude if not already present.""" + excl = _exclude_file(common_dir) + excl.parent.mkdir(parents=True, exist_ok=True) + if excl.exists(): + lines = excl.read_text().splitlines() + if any(line.strip() == ".worktrees/" for line in lines): + return + with excl.open("a") as fh: + fh.write(".worktrees/\n") + + +async def _resolve_common_dir(repo: Path) -> Path: + """Return the git common dir, resolved to an absolute path.""" + rc, out, err = await _git("rev-parse", "--git-common-dir", cwd=repo) + if rc != 0: + raise ServiceError(f"git rev-parse --git-common-dir failed: {err.strip()}") + raw = out.strip() + p = Path(raw) + if not p.is_absolute(): + p = (repo / p).resolve() + return p + + +async def create_worktree(repo: Path, branch: str) -> Path: + """Create a new worktree at /.worktrees/ on a new branch. + + Steps: + 1. Validate branch name with git check-ref-format. + 2. Ensure .worktrees/ is in the git exclude file. + 3. git worktree add -b . + 4. Return the worktree path. + """ + # Step 1: validate branch name. + rc, _, err = await _git("check-ref-format", "--branch", branch, cwd=repo) + if rc != 0: + raise ServiceError(f"Invalid branch name: {branch!r}") + + # Step 2: exclude .worktrees/ from git tracking. + common_dir = await _resolve_common_dir(repo) + _ensure_worktrees_excluded(common_dir) + + # Step 3: add worktree. + wt_path = repo / ".worktrees" / branch + rc, _, err = await _git("worktree", "add", "-b", branch, str(wt_path), cwd=repo) + if rc != 0: + raise ServiceError(f"git worktree add failed: {err.strip()}") + + return wt_path.resolve() + + +async def add_worktree_for_branch(repo: Path, branch: str) -> Path: + """Recreate a worktree for an existing branch (no -b flag). + + Used when the worktree directory has been removed but the branch still + exists. + """ + common_dir = await _resolve_common_dir(repo) + _ensure_worktrees_excluded(common_dir) + + wt_path = repo / ".worktrees" / branch + rc, _, err = await _git("worktree", "add", str(wt_path), branch, cwd=repo) + if rc != 0: + raise ServiceError(f"git worktree add failed: {err.strip()}") + + return wt_path.resolve() + + +async def worktree_state(repo: Path, path: Path, branch: str) -> WorktreeState: + """Return a WorktreeState snapshot for the given worktree/branch. + + - exists: path exists on disk. + - dirty: non-empty git status --porcelain in the worktree. + - unique_commits: commits reachable from branch but not from any other ref. + """ + exists = path.exists() + + # Dirty check — only meaningful when the worktree exists. + dirty = False + if exists: + rc, out, _ = await _git("status", "--porcelain", cwd=path) + dirty = rc == 0 and bool(out.strip()) + + # Unique commits — count commits on branch not reachable from other refs. + rc, out, _ = await _git( + "for-each-ref", + "--format=%(refname)", + "refs/heads", + "refs/remotes", + cwd=repo, + ) + all_refs: list[str] = [line.strip() for line in out.splitlines() if line.strip()] + branch_ref = f"refs/heads/{branch}" + other_refs = [r for r in all_refs if r != branch_ref] + + if other_refs: + rc, out, _ = await _git( + "rev-list", "--count", branch, "--not", *other_refs, cwd=repo + ) + else: + rc, out, _ = await _git("rev-list", "--count", branch, cwd=repo) + + unique = 0 + if rc == 0 and out.strip().isdigit(): + unique = int(out.strip()) + + return WorktreeState(exists=exists, dirty=dirty, unique_commits=unique) + + +async def remove_worktree(repo: Path, path: Path, branch: str, force: bool) -> None: + """Remove a worktree and best-effort delete its branch. + + Steps: + 1. git worktree remove [--force] — ServiceError on failure. + 2. git branch -d — failure silently ignored. + 3. git worktree prune. + """ + args = ["worktree", "remove"] + if force: + args.append("--force") + args.append(str(path)) + + rc, _, err = await _git(*args, cwd=repo) + if rc != 0: + raise ServiceError(f"git worktree remove failed: {err.strip()}") + + # Best-effort branch deletion; unmerged branches (with unique commits) + # will make this fail — we intentionally ignore the failure. + await _git("branch", "-d", branch, cwd=repo) + + # Prune stale worktree metadata. + await _git("worktree", "prune", cwd=repo) diff --git a/tests/test_worktree.py b/tests/test_worktree.py new file mode 100644 index 0000000..4277438 --- /dev/null +++ b/tests/test_worktree.py @@ -0,0 +1,284 @@ +"""Tests for hqt.git.worktree — real git repos in tmp_path.""" + +import subprocess +from pathlib import Path + +import pytest + +from hqt.errors import ServiceError +from hqt.git.worktree import ( + add_worktree_for_branch, + create_worktree, + is_git_repo, + remove_worktree, + slugify, + worktree_state, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _git(repo: Path, *args: str) -> str: + """Run a git command in repo; return stdout stripped.""" + result = subprocess.run( + ["git", *args], + cwd=repo, + capture_output=True, + text=True, + check=True, + ) + return result.stdout.strip() + + +def _make_repo(path: Path) -> Path: + """Init a bare-minimum git repo with one commit; return path.""" + path.mkdir(parents=True, exist_ok=True) + _git(path, "init") + _git(path, "config", "user.name", "Test User") + _git(path, "config", "user.email", "test@example.com") + (path / "README").write_text("init") + _git(path, "add", "README") + _git(path, "commit", "-m", "init") + return path + + +# --------------------------------------------------------------------------- +# slugify +# --------------------------------------------------------------------------- + + +def test_slugify_lowercase(): + assert slugify("Hello World") == "hello-world" + + +def test_slugify_mixed_case_and_spaces(): + assert slugify("My Feature Branch") == "my-feature-branch" + + +def test_slugify_punctuation(): + assert slugify("hello, world!") == "hello-world" + + +def test_slugify_leading_trailing_junk(): + assert slugify("---foo---") == "foo" + + +def test_slugify_empty_string(): + assert slugify("") == "" + + +def test_slugify_collapse_runs(): + # Multiple consecutive non-alphanumeric chars collapse to one dash. + assert slugify("foo!!!bar___baz") == "foo-bar-baz" + + +def test_slugify_numbers_kept(): + assert slugify("feature-123") == "feature-123" + + +# --------------------------------------------------------------------------- +# is_git_repo +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_is_git_repo_true(tmp_path): + repo = _make_repo(tmp_path / "repo") + assert await is_git_repo(repo) is True + + +@pytest.mark.asyncio +async def test_is_git_repo_plain_dir(tmp_path): + plain = tmp_path / "plain" + plain.mkdir() + assert await is_git_repo(plain) is False + + +@pytest.mark.asyncio +async def test_is_git_repo_missing_dir(tmp_path): + missing = tmp_path / "does_not_exist" + assert await is_git_repo(missing) is False + + +# --------------------------------------------------------------------------- +# create_worktree +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_worktree_creates_directory(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-x") + assert wt.exists() + assert wt == repo / ".worktrees" / "feature-x" + + +@pytest.mark.asyncio +async def test_create_worktree_branch_checked_out(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-x") + branch = _git(wt, "rev-parse", "--abbrev-ref", "HEAD") + assert branch == "feature-x" + + +@pytest.mark.asyncio +async def test_create_worktree_exclude_updated(tmp_path): + repo = _make_repo(tmp_path / "repo") + await create_worktree(repo, "feature-x") + exclude = repo / ".git" / "info" / "exclude" + assert exclude.exists() + assert ".worktrees/" in exclude.read_text() + + +@pytest.mark.asyncio +async def test_create_worktree_exclude_not_duplicated(tmp_path): + repo = _make_repo(tmp_path / "repo") + await create_worktree(repo, "feature-x") + await create_worktree(repo, "feature-y") + exclude = repo / ".git" / "info" / "exclude" + content = exclude.read_text() + assert content.count(".worktrees/") == 1 + + +@pytest.mark.asyncio +async def test_create_worktree_duplicate_branch_raises(tmp_path): + repo = _make_repo(tmp_path / "repo") + await create_worktree(repo, "feature-x") + with pytest.raises(ServiceError): + await create_worktree(repo, "feature-x") + + +@pytest.mark.asyncio +async def test_create_worktree_invalid_branch_name_raises(tmp_path): + repo = _make_repo(tmp_path / "repo") + with pytest.raises(ServiceError, match="Invalid branch name"): + await create_worktree(repo, "foo..bar") + + +# --------------------------------------------------------------------------- +# add_worktree_for_branch +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_add_worktree_for_branch_creates_directory(tmp_path): + repo = _make_repo(tmp_path / "repo") + # Create a branch first then remove the worktree dir manually. + wt = await create_worktree(repo, "feature-z") + _git(repo, "worktree", "remove", "--force", str(wt)) + # Now recreate via add_worktree_for_branch. + wt2 = await add_worktree_for_branch(repo, "feature-z") + assert wt2.exists() + assert wt2 == repo / ".worktrees" / "feature-z" + branch = _git(wt2, "rev-parse", "--abbrev-ref", "HEAD") + assert branch == "feature-z" + + +# --------------------------------------------------------------------------- +# worktree_state +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_worktree_state_fresh(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-a") + state = await worktree_state(repo, wt, "feature-a") + assert state.exists is True + assert state.dirty is False + assert state.unique_commits == 0 + + +@pytest.mark.asyncio +async def test_worktree_state_dirty(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-b") + (wt / "untracked.txt").write_text("new") + state = await worktree_state(repo, wt, "feature-b") + assert state.dirty is True + + +@pytest.mark.asyncio +async def test_worktree_state_unique_commits(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-c") + (wt / "new.txt").write_text("content") + _git(wt, "add", "new.txt") + subprocess.run( + ["git", "commit", "-m", "branch commit"], + cwd=wt, + capture_output=True, + check=True, + ) + state = await worktree_state(repo, wt, "feature-c") + assert state.unique_commits == 1 + + +@pytest.mark.asyncio +async def test_worktree_state_missing_dir(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = repo / ".worktrees" / "ghost" + state = await worktree_state(repo, wt, "ghost") + assert state.exists is False + assert state.dirty is False + assert state.unique_commits == 0 + + +# --------------------------------------------------------------------------- +# remove_worktree +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_remove_worktree_clean_merged(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-d") + await remove_worktree(repo, wt, "feature-d", force=False) + assert not wt.exists() + # Branch should be deleted (no unique commits → git branch -d succeeds). + branches = _git(repo, "branch", "--list", "feature-d") + assert branches == "" + + +@pytest.mark.asyncio +async def test_remove_worktree_dirty_without_force_raises(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-e") + (wt / "dirty.txt").write_text("unsaved") + _git(wt, "add", "dirty.txt") + with pytest.raises(ServiceError): + await remove_worktree(repo, wt, "feature-e", force=False) + assert wt.exists() + + +@pytest.mark.asyncio +async def test_remove_worktree_dirty_with_force_removes(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-f") + (wt / "dirty.txt").write_text("unsaved") + _git(wt, "add", "dirty.txt") + await remove_worktree(repo, wt, "feature-f", force=True) + assert not wt.exists() + + +@pytest.mark.asyncio +async def test_remove_worktree_unique_commits_branch_survives(tmp_path): + repo = _make_repo(tmp_path / "repo") + wt = await create_worktree(repo, "feature-g") + (wt / "new.txt").write_text("content") + _git(wt, "add", "new.txt") + subprocess.run( + ["git", "commit", "-m", "unique"], + cwd=wt, + capture_output=True, + check=True, + ) + # remove_worktree must NOT raise even though branch -d will fail. + await remove_worktree(repo, wt, "feature-g", force=True) + assert not wt.exists() + # Branch still exists because git branch -d failed (unmerged) — silently ignored. + branches = _git(repo, "branch", "--list", "feature-g") + assert "feature-g" in branches