136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
"""Git worktree + branch management for batch isolation.
|
||
|
||
Each batch × track combination gets its own git worktree so that:
|
||
- Track A and Track B never share filesystem state.
|
||
- Build directories are fully independent.
|
||
- Every successful file migration is committed immediately to its branch.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Low-level helper
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _git(args: list[str], cwd: Path | None = None) -> tuple[int, str, str]:
|
||
result = subprocess.run(
|
||
["git"] + args,
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=str(cwd) if cwd else None,
|
||
)
|
||
return result.returncode, result.stdout.strip(), result.stderr.strip()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Branch / worktree lifecycle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def branch_name(batch_id: str, track_id: str) -> str:
|
||
"""Canonical branch name: ``migration/<batch_id>/<track_id>``."""
|
||
return f"migration/{batch_id}/{track_id}"
|
||
|
||
|
||
def setup_worktree(
|
||
batch_id: str,
|
||
track_id: str,
|
||
repo_root: Path,
|
||
workspace_root: Path,
|
||
) -> Path:
|
||
"""Ensure a git worktree exists for *batch_id* × *track_id*.
|
||
|
||
Creates the branch (from HEAD) if it does not yet exist.
|
||
Returns the worktree path.
|
||
"""
|
||
branch = branch_name(batch_id, track_id)
|
||
worktree_path = workspace_root / batch_id / track_id
|
||
|
||
# Create the branch if needed.
|
||
rc, _, _ = _git(["rev-parse", "--verify", branch], cwd=repo_root)
|
||
if rc != 0:
|
||
_git(["branch", branch], cwd=repo_root)
|
||
|
||
# Remove stale worktree entry (if the directory was deleted manually).
|
||
_git(["worktree", "prune"], cwd=repo_root)
|
||
|
||
# Check whether the worktree is already registered.
|
||
_, wt_list, _ = _git(["worktree", "list", "--porcelain"], cwd=repo_root)
|
||
if str(worktree_path) in wt_list:
|
||
return worktree_path
|
||
|
||
worktree_path.parent.mkdir(parents=True, exist_ok=True)
|
||
rc, out, err = _git(
|
||
["worktree", "add", str(worktree_path), branch],
|
||
cwd=repo_root,
|
||
)
|
||
if rc != 0:
|
||
raise RuntimeError(
|
||
f"git worktree add failed for {worktree_path}: {err}"
|
||
)
|
||
return worktree_path
|
||
|
||
|
||
def teardown_worktree(
|
||
batch_id: str,
|
||
track_id: str,
|
||
repo_root: Path,
|
||
workspace_root: Path,
|
||
) -> None:
|
||
"""Remove the worktree (but keep the branch)."""
|
||
worktree_path = workspace_root / batch_id / track_id
|
||
_git(["worktree", "remove", "--force", str(worktree_path)], cwd=repo_root)
|
||
_git(["worktree", "prune"], cwd=repo_root)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Per-file commit helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def commit_migrated_file(
|
||
file_path_in_worktree: Path,
|
||
worktree_path: Path,
|
||
track_id: str,
|
||
) -> None:
|
||
"""Stage and commit a single successfully migrated file."""
|
||
_git(["add", str(file_path_in_worktree)], cwd=worktree_path)
|
||
rel = file_path_in_worktree.relative_to(worktree_path)
|
||
_git(
|
||
["commit", "-m", f"migrate({track_id}): {rel}"],
|
||
cwd=worktree_path,
|
||
)
|
||
|
||
|
||
def commit_generated_tests(
|
||
tests_dir: Path,
|
||
worktree_path: Path,
|
||
track_id: str,
|
||
test_prefix: str,
|
||
) -> None:
|
||
"""Stage and commit generated test files and their CMakeLists entry."""
|
||
_git(["add", str(tests_dir)], cwd=worktree_path)
|
||
_git(
|
||
["commit", "-m", f"test({track_id}): generated tests for {test_prefix}"],
|
||
cwd=worktree_path,
|
||
)
|
||
|
||
|
||
def commit_batch_summary(
|
||
worktree_path: Path,
|
||
track_id: str,
|
||
batch_id: str,
|
||
) -> None:
|
||
"""Final commit for a batch (quarantine files, reports, etc.)."""
|
||
_git(["add", "-A"], cwd=worktree_path)
|
||
rc, _, _ = _git(
|
||
["diff", "--cached", "--quiet"], cwd=worktree_path
|
||
)
|
||
if rc != 0: # there are staged changes
|
||
_git(
|
||
["commit", "-m", f"batch({track_id}): finalise {batch_id}"],
|
||
cwd=worktree_path,
|
||
)
|