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 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Git utilities for hqt."""
|
||||
@@ -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 <common-dir>/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 <repo>/.worktrees/<branch> 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 <branch> <path>.
|
||||
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] <path> — ServiceError on failure.
|
||||
2. git branch -d <branch> — 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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user