felixm/hqt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden git worktree module (review findings)

Browse Source 
- is_git_repo: check out.strip() == "true" (bare repos / .git dirs exit 0 with "false")
- _validate_branch: new shared helper calls check-ref-format; used in all three
  public branch-taking functions to block option-injection (--detach, --all, etc.)
- _ensure_worktrees_excluded: prepend newline when existing content lacks trailing newline
- WorktreeState: make frozen=True (codebase precedent)
- create_worktree invalid-branch ServiceError now includes git stderr
- remove_worktree: log.debug for silently-swallowed branch -d / worktree prune failures
- worktree_state docstring: document git-failure → clean-state behaviour
- Tests: 9 new regression tests; raw subprocess.run commit calls replaced with _commit helper

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Felix Martin
2026-06-10 19:59:16 -04:00
parent 0e7d8ff35b
commit cc49859b08
2 changed files with 109 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/hqt/git/worktree.py
+35 -9
View File
@@ -56,14 +56,14 @@ async def _git(
async def is_git_repo(path: Path) -> bool:
"""Return True iff path is inside a git work tree."""
"""Return True iff path is inside a git work tree (not a bare repo or .git dir)."""
if not path.exists():
return False
rc, _, _ = await _git("rev-parse", "--is-inside-work-tree", cwd=path)
return rc == 0
rc, out, _ = await _git("rev-parse", "--is-inside-work-tree", cwd=path)
return rc == 0 and out.strip() == "true"
@dataclass
@dataclass(frozen=True)
class WorktreeState:
"""Snapshot of a worktree's runtime state."""
@@ -85,6 +85,13 @@ def _ensure_worktrees_excluded(common_dir: Path) -> None:
if any(line.strip() == ".worktrees/" for line in lines):
return
with excl.open("a") as fh:
# If the file is non-empty and lacks a trailing newline, add one first
# to avoid corrupting the last pattern (e.g. "*.log" + ".worktrees/"
# would become "*.log.worktrees/" without the separator).
if excl.exists() and excl.stat().st_size > 0:
existing = excl.read_text()
if not existing.endswith("\n"):
fh.write("\n")
fh.write(".worktrees/\n")
@@ -100,6 +107,16 @@ async def _resolve_common_dir(repo: Path) -> Path:
return p
async def _validate_branch(repo: Path, branch: str) -> None:
"""Validate branch name with git check-ref-format; raise ServiceError if invalid."""
rc, _, err = await _git("check-ref-format", "--branch", branch, cwd=repo)
if rc != 0:
raise ServiceError(
f"Invalid branch name: {branch!r}"
+ (f": {err.strip()}" if err.strip() else "")
)
async def create_worktree(repo: Path, branch: str) -> Path:
"""Create a new worktree at <repo>/.worktrees/<branch> on a new branch.
@@ -110,9 +127,7 @@ async def create_worktree(repo: Path, branch: str) -> 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}")
await _validate_branch(repo, branch)
# Step 2: exclude .worktrees/ from git tracking.
common_dir = await _resolve_common_dir(repo)
@@ -133,6 +148,8 @@ async def add_worktree_for_branch(repo: Path, branch: str) -> Path:
Used when the worktree directory has been removed but the branch still
exists.
"""
await _validate_branch(repo, branch)
common_dir = await _resolve_common_dir(repo)
_ensure_worktrees_excluded(common_dir)
@@ -150,7 +167,12 @@ async def worktree_state(repo: Path, path: Path, branch: str) -> WorktreeState:
- 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.
Git failures (e.g. branch does not exist yet) are treated as clean state:
dirty=False and unique_commits=0.
"""
await _validate_branch(repo, branch)
exists = path.exists()
# Dirty check — only meaningful when the worktree exists.
@@ -204,7 +226,11 @@ async def remove_worktree(repo: Path, path: Path, branch: str, force: bool) -> N
# 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)
rc, _, err = await _git("branch", "-d", branch, cwd=repo)
if rc != 0:
log.debug("git branch -d %r failed (ignored): %s", branch, err.strip())
# Prune stale worktree metadata.
await _git("worktree", "prune", cwd=repo)
rc, _, err = await _git("worktree", "prune", cwd=repo)
if rc != 0:
log.debug("git worktree prune failed (ignored): %s", err.strip())
tests/test_worktree.py
+74 -16
View File
@@ -7,6 +7,7 @@ import pytest
from hqt.errors import ServiceError
from hqt.git.worktree import (
WorktreeState,
add_worktree_for_branch,
create_worktree,
is_git_repo,
@@ -33,6 +34,13 @@ def _git(repo: Path, *args: str) -> str:
return result.stdout.strip()
def _commit(repo: Path, filename: str, message: str) -> None:
"""Create a file, stage it, and commit it in repo."""
(repo / filename).write_text("content")
_git(repo, "add", filename)
_git(repo, "commit", "-m", message)
def _make_repo(path: Path) -> Path:
"""Init a bare-minimum git repo with one commit; return path."""
path.mkdir(parents=True, exist_ok=True)
@@ -103,6 +111,36 @@ async def test_is_git_repo_missing_dir(tmp_path):
assert await is_git_repo(missing) is False
@pytest.mark.asyncio
async def test_is_git_repo_bare_repo_returns_false(tmp_path):
"""git rev-parse --is-inside-work-tree exits 0 and prints 'false' in bare repos."""
bare = tmp_path / "bare.git"
bare.mkdir()
subprocess.run(
["git", "init", "--bare", str(bare)], check=True, capture_output=True
)
assert await is_git_repo(bare) is False
@pytest.mark.asyncio
async def test_is_git_repo_dot_git_dir_returns_false(tmp_path):
"""git rev-parse --is-inside-work-tree exits 0 and prints 'false' inside .git."""
repo = _make_repo(tmp_path / "repo")
dot_git = repo / ".git"
assert await is_git_repo(dot_git) is False
# ---------------------------------------------------------------------------
# WorktreeState frozen dataclass
# ---------------------------------------------------------------------------
def test_worktree_state_is_frozen():
state = WorktreeState(exists=True, dirty=False, unique_commits=0)
with pytest.raises(Exception):
state.exists = False # type: ignore[misc]
# ---------------------------------------------------------------------------
# create_worktree
# ---------------------------------------------------------------------------
@@ -158,6 +196,23 @@ async def test_create_worktree_invalid_branch_name_raises(tmp_path):
await create_worktree(repo, "foo..bar")
@pytest.mark.asyncio
async def test_create_worktree_exclude_no_trailing_newline(tmp_path):
"""Exclude file lacking trailing newline must not corrupt the last pattern."""
repo = _make_repo(tmp_path / "repo")
exclude = repo / ".git" / "info" / "exclude"
exclude.parent.mkdir(parents=True, exist_ok=True)
# Write a pattern WITHOUT a trailing newline.
exclude.write_text("*.log")
await create_worktree(repo, "feature-x")
content = exclude.read_text()
lines = [ln for ln in content.splitlines() if ln.strip()]
assert "*.log" in lines, f"*.log missing from: {content!r}"
assert ".worktrees/" in lines, f".worktrees/ missing from: {content!r}"
# ---------------------------------------------------------------------------
# add_worktree_for_branch
# ---------------------------------------------------------------------------
@@ -177,6 +232,14 @@ async def test_add_worktree_for_branch_creates_directory(tmp_path):
assert branch == "feature-z"
@pytest.mark.asyncio
async def test_add_worktree_for_branch_option_like_name_raises(tmp_path):
"""Option-looking branch names (e.g. --detach) must be rejected before git sees them."""
repo = _make_repo(tmp_path / "repo")
with pytest.raises(ServiceError, match="Invalid branch name"):
await add_worktree_for_branch(repo, "--detach")
# ---------------------------------------------------------------------------
# worktree_state
# ---------------------------------------------------------------------------
@@ -205,14 +268,7 @@ async def test_worktree_state_dirty(tmp_path):
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,
)
_commit(wt, "new.txt", "branch commit")
state = await worktree_state(repo, wt, "feature-c")
assert state.unique_commits == 1
@@ -227,6 +283,15 @@ async def test_worktree_state_missing_dir(tmp_path):
assert state.unique_commits == 0
@pytest.mark.asyncio
async def test_worktree_state_option_like_branch_raises(tmp_path):
"""Option-looking branch names (e.g. --all) must be rejected."""
repo = _make_repo(tmp_path / "repo")
wt = repo / ".worktrees" / "whatever"
with pytest.raises(ServiceError, match="Invalid branch name"):
await worktree_state(repo, wt, "--all")
# ---------------------------------------------------------------------------
# remove_worktree
# ---------------------------------------------------------------------------
@@ -268,14 +333,7 @@ async def test_remove_worktree_dirty_with_force_removes(tmp_path):
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,
)
_commit(wt, "new.txt", "unique")
# remove_worktree must NOT raise even though branch -d will fail.
await remove_worktree(repo, wt, "feature-g", force=True)
assert not wt.exists()