test_worktree.py
python
| 1 | """Tests for ``muse worktree`` subcommands. |
| 2 | |
| 3 | Covers the acceptance criteria: |
| 4 | - ``muse worktree add`` creates a linked worktree with shared objects store. |
| 5 | - Linked worktrees have independent muse-work/ and .muse gitdir file. |
| 6 | - ``muse worktree list`` shows main + linked worktrees with path, branch, HEAD. |
| 7 | - ``muse worktree remove`` cleans up the directory and registration. |
| 8 | - ``muse worktree prune`` removes stale registrations (directory gone). |
| 9 | - Cannot check out the same branch in two worktrees simultaneously. |
| 10 | """ |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import json |
| 14 | import pathlib |
| 15 | |
| 16 | import pytest |
| 17 | from typer.testing import CliRunner |
| 18 | |
| 19 | from maestro.muse_cli.app import cli |
| 20 | from maestro.muse_cli.commands.worktree import ( |
| 21 | WorktreeInfo, |
| 22 | add_worktree, |
| 23 | list_worktrees, |
| 24 | prune_worktrees, |
| 25 | remove_worktree, |
| 26 | ) |
| 27 | |
| 28 | runner = CliRunner() |
| 29 | |
| 30 | |
| 31 | # --------------------------------------------------------------------------- |
| 32 | # Helpers |
| 33 | # --------------------------------------------------------------------------- |
| 34 | |
| 35 | |
| 36 | def _init_repo(tmp_path: pathlib.Path, *, initial_commit: str = "") -> pathlib.Path: |
| 37 | """Create a minimal .muse/ repo under tmp_path.""" |
| 38 | muse_dir = tmp_path / ".muse" |
| 39 | muse_dir.mkdir() |
| 40 | (muse_dir / "repo.json").write_text( |
| 41 | json.dumps({"repo_id": "test-repo-id", "schema_version": "1"}), |
| 42 | encoding="utf-8", |
| 43 | ) |
| 44 | (muse_dir / "HEAD").write_text("refs/heads/main\n", encoding="utf-8") |
| 45 | refs_dir = muse_dir / "refs" / "heads" |
| 46 | refs_dir.mkdir(parents=True) |
| 47 | (refs_dir / "main").write_text(initial_commit, encoding="utf-8") |
| 48 | return tmp_path |
| 49 | |
| 50 | |
| 51 | def _env(root: pathlib.Path) -> dict[str, str]: |
| 52 | return {"MUSE_REPO_ROOT": str(root)} |
| 53 | |
| 54 | |
| 55 | # --------------------------------------------------------------------------- |
| 56 | # list_worktrees |
| 57 | # --------------------------------------------------------------------------- |
| 58 | |
| 59 | |
| 60 | class TestListWorktrees: |
| 61 | def test_main_only_returns_one_entry(self, tmp_path: pathlib.Path) -> None: |
| 62 | root = _init_repo(tmp_path) |
| 63 | worktrees = list_worktrees(root) |
| 64 | assert len(worktrees) == 1 |
| 65 | assert worktrees[0].is_main |
| 66 | assert worktrees[0].branch == "main" |
| 67 | |
| 68 | def test_main_worktree_path_is_root(self, tmp_path: pathlib.Path) -> None: |
| 69 | root = _init_repo(tmp_path) |
| 70 | worktrees = list_worktrees(root) |
| 71 | assert worktrees[0].path == root |
| 72 | |
| 73 | def test_main_worktree_head_commit(self, tmp_path: pathlib.Path) -> None: |
| 74 | root = _init_repo(tmp_path, initial_commit="abc12345") |
| 75 | worktrees = list_worktrees(root) |
| 76 | assert worktrees[0].head_commit == "abc12345" |
| 77 | |
| 78 | def test_linked_worktrees_appear_after_main( |
| 79 | self, tmp_path: pathlib.Path |
| 80 | ) -> None: |
| 81 | root = _init_repo(tmp_path) |
| 82 | link_path = tmp_path.parent / "linked-wt" |
| 83 | add_worktree(root=root, link_path=link_path, branch="feature/test") |
| 84 | worktrees = list_worktrees(root) |
| 85 | assert len(worktrees) == 2 |
| 86 | assert worktrees[0].is_main |
| 87 | assert not worktrees[1].is_main |
| 88 | assert worktrees[1].branch == "feature/test" |
| 89 | |
| 90 | |
| 91 | # --------------------------------------------------------------------------- |
| 92 | # add_worktree — regression test required by # --------------------------------------------------------------------------- |
| 93 | |
| 94 | |
| 95 | class TestWorktreeAdd: |
| 96 | def test_worktree_add_creates_linked_worktree_with_shared_objects( |
| 97 | self, tmp_path: pathlib.Path |
| 98 | ) -> None: |
| 99 | """Regression: add_worktree creates a linked dir sharing the main .muse/.""" |
| 100 | root = _init_repo(tmp_path) |
| 101 | link_path = tmp_path.parent / "my-feature-wt" |
| 102 | info = add_worktree(root=root, link_path=link_path, branch="feature/guitar") |
| 103 | |
| 104 | assert isinstance(info, WorktreeInfo) |
| 105 | assert info.path == link_path |
| 106 | assert info.branch == "feature/guitar" |
| 107 | assert not info.is_main |
| 108 | |
| 109 | # Directory created. |
| 110 | assert link_path.is_dir() |
| 111 | # muse-work/ created. |
| 112 | assert (link_path / "muse-work").is_dir() |
| 113 | # .muse gitdir file created. |
| 114 | muse_file = link_path / ".muse" |
| 115 | assert muse_file.is_file() |
| 116 | assert "gitdir:" in muse_file.read_text() |
| 117 | |
| 118 | # Registered in main .muse/worktrees/. |
| 119 | wt_dir = root / ".muse" / "worktrees" |
| 120 | entries = list(wt_dir.iterdir()) |
| 121 | assert len(entries) == 1 |
| 122 | registration = entries[0] |
| 123 | assert (registration / "path").read_text().strip() == str(link_path) |
| 124 | assert (registration / "branch").read_text().strip() == "feature/guitar" |
| 125 | |
| 126 | def test_worktree_add_creates_branch_ref_if_absent( |
| 127 | self, tmp_path: pathlib.Path |
| 128 | ) -> None: |
| 129 | root = _init_repo(tmp_path) |
| 130 | link_path = tmp_path.parent / "new-branch-wt" |
| 131 | add_worktree(root=root, link_path=link_path, branch="feature/new") |
| 132 | ref_path = root / ".muse" / "refs" / "heads" / "feature" / "new" |
| 133 | assert ref_path.exists() |
| 134 | |
| 135 | def test_worktree_add_reuses_existing_branch_ref( |
| 136 | self, tmp_path: pathlib.Path |
| 137 | ) -> None: |
| 138 | root = _init_repo(tmp_path, initial_commit="deadbeef") |
| 139 | muse_dir = root / ".muse" |
| 140 | # Pre-create a branch ref. |
| 141 | branch_ref = muse_dir / "refs" / "heads" / "feature" / "existing" |
| 142 | branch_ref.parent.mkdir(parents=True, exist_ok=True) |
| 143 | branch_ref.write_text("deadbeef") |
| 144 | |
| 145 | link_path = tmp_path.parent / "existing-branch-wt" |
| 146 | info = add_worktree(root=root, link_path=link_path, branch="feature/existing") |
| 147 | assert info.branch == "feature/existing" |
| 148 | # Ref still has the correct commit. |
| 149 | assert branch_ref.read_text() == "deadbeef" |
| 150 | |
| 151 | def test_worktree_add_returns_worktree_info( |
| 152 | self, tmp_path: pathlib.Path |
| 153 | ) -> None: |
| 154 | root = _init_repo(tmp_path) |
| 155 | link_path = tmp_path.parent / "info-wt" |
| 156 | info = add_worktree(root=root, link_path=link_path, branch="main2") |
| 157 | assert info.path == link_path |
| 158 | assert info.slug != "" |
| 159 | |
| 160 | def test_worktree_add_path_already_exists_exits_1( |
| 161 | self, tmp_path: pathlib.Path |
| 162 | ) -> None: |
| 163 | root = _init_repo(tmp_path) |
| 164 | existing = tmp_path.parent / "already-exists" |
| 165 | existing.mkdir() |
| 166 | result = runner.invoke( |
| 167 | cli, ["worktree", "add", str(existing), "feature/x"], env=_env(root) |
| 168 | ) |
| 169 | assert result.exit_code == 1 |
| 170 | assert "already exists" in result.output.lower() |
| 171 | |
| 172 | def test_worktree_add_same_branch_twice_exits_1( |
| 173 | self, tmp_path: pathlib.Path |
| 174 | ) -> None: |
| 175 | """Cannot check out the same branch in two worktrees simultaneously.""" |
| 176 | root = _init_repo(tmp_path) |
| 177 | first = tmp_path.parent / "first-wt" |
| 178 | add_worktree(root=root, link_path=first, branch="feature/shared") |
| 179 | |
| 180 | second = tmp_path.parent / "second-wt" |
| 181 | result = runner.invoke( |
| 182 | cli, ["worktree", "add", str(second), "feature/shared"], env=_env(root) |
| 183 | ) |
| 184 | assert result.exit_code == 1 |
| 185 | assert "already checked out" in result.output.lower() |
| 186 | |
| 187 | def test_worktree_add_main_branch_conflicts_exits_1( |
| 188 | self, tmp_path: pathlib.Path |
| 189 | ) -> None: |
| 190 | """Adding a worktree for the branch currently in main exits 1.""" |
| 191 | root = _init_repo(tmp_path) |
| 192 | link_path = tmp_path.parent / "main-conflict-wt" |
| 193 | result = runner.invoke( |
| 194 | cli, ["worktree", "add", str(link_path), "main"], env=_env(root) |
| 195 | ) |
| 196 | assert result.exit_code == 1 |
| 197 | assert "already checked out" in result.output.lower() |
| 198 | |
| 199 | def test_worktree_add_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: |
| 200 | result = runner.invoke( |
| 201 | cli, |
| 202 | ["worktree", "add", str(tmp_path / "new-wt"), "feature/x"], |
| 203 | env={"MUSE_REPO_ROOT": str(tmp_path)}, |
| 204 | ) |
| 205 | assert result.exit_code == 2 |
| 206 | |
| 207 | |
| 208 | # --------------------------------------------------------------------------- |
| 209 | # remove_worktree |
| 210 | # --------------------------------------------------------------------------- |
| 211 | |
| 212 | |
| 213 | class TestWorktreeRemove: |
| 214 | def test_remove_deletes_directory_and_registration( |
| 215 | self, tmp_path: pathlib.Path |
| 216 | ) -> None: |
| 217 | root = _init_repo(tmp_path) |
| 218 | link_path = tmp_path.parent / "remove-me-wt" |
| 219 | add_worktree(root=root, link_path=link_path, branch="feature/removeme") |
| 220 | |
| 221 | assert link_path.exists() |
| 222 | remove_worktree(root=root, link_path=link_path) |
| 223 | assert not link_path.exists() |
| 224 | |
| 225 | wt_dir = root / ".muse" / "worktrees" |
| 226 | remaining = list(wt_dir.iterdir()) |
| 227 | assert len(remaining) == 0 |
| 228 | |
| 229 | def test_remove_preserves_branch_ref_in_main( |
| 230 | self, tmp_path: pathlib.Path |
| 231 | ) -> None: |
| 232 | root = _init_repo(tmp_path) |
| 233 | link_path = tmp_path.parent / "branch-preserved-wt" |
| 234 | add_worktree(root=root, link_path=link_path, branch="feature/keep") |
| 235 | remove_worktree(root=root, link_path=link_path) |
| 236 | |
| 237 | # Branch ref still exists in main repo. |
| 238 | ref_path = root / ".muse" / "refs" / "heads" / "feature" / "keep" |
| 239 | assert ref_path.exists() |
| 240 | |
| 241 | def test_remove_main_worktree_exits_1(self, tmp_path: pathlib.Path) -> None: |
| 242 | root = _init_repo(tmp_path) |
| 243 | result = runner.invoke( |
| 244 | cli, ["worktree", "remove", str(root)], env=_env(root) |
| 245 | ) |
| 246 | assert result.exit_code == 1 |
| 247 | assert "cannot remove" in result.output.lower() |
| 248 | |
| 249 | def test_remove_unregistered_path_exits_1(self, tmp_path: pathlib.Path) -> None: |
| 250 | root = _init_repo(tmp_path) |
| 251 | unregistered = tmp_path.parent / "not-registered" |
| 252 | unregistered.mkdir() |
| 253 | result = runner.invoke( |
| 254 | cli, ["worktree", "remove", str(unregistered)], env=_env(root) |
| 255 | ) |
| 256 | assert result.exit_code == 1 |
| 257 | assert "not a registered" in result.output.lower() |
| 258 | |
| 259 | def test_remove_stale_directory_still_deregisters( |
| 260 | self, tmp_path: pathlib.Path |
| 261 | ) -> None: |
| 262 | """remove works even if the linked directory is already gone.""" |
| 263 | root = _init_repo(tmp_path) |
| 264 | link_path = tmp_path.parent / "gone-wt" |
| 265 | add_worktree(root=root, link_path=link_path, branch="feature/gone") |
| 266 | # Simulate the directory disappearing externally. |
| 267 | import shutil |
| 268 | shutil.rmtree(link_path) |
| 269 | |
| 270 | # remove should still deregister cleanly. |
| 271 | remove_worktree(root=root, link_path=link_path) |
| 272 | wt_dir = root / ".muse" / "worktrees" |
| 273 | assert list(wt_dir.iterdir()) == [] |
| 274 | |
| 275 | |
| 276 | # --------------------------------------------------------------------------- |
| 277 | # prune_worktrees |
| 278 | # --------------------------------------------------------------------------- |
| 279 | |
| 280 | |
| 281 | class TestWorktreePrune: |
| 282 | def test_prune_removes_stale_registration(self, tmp_path: pathlib.Path) -> None: |
| 283 | root = _init_repo(tmp_path) |
| 284 | link_path = tmp_path.parent / "stale-wt" |
| 285 | add_worktree(root=root, link_path=link_path, branch="feature/stale") |
| 286 | # Externally delete the linked directory. |
| 287 | import shutil |
| 288 | shutil.rmtree(link_path) |
| 289 | |
| 290 | pruned = prune_worktrees(root=root) |
| 291 | assert str(link_path) in pruned |
| 292 | |
| 293 | wt_dir = root / ".muse" / "worktrees" |
| 294 | assert list(wt_dir.iterdir()) == [] |
| 295 | |
| 296 | def test_prune_leaves_live_worktrees_intact(self, tmp_path: pathlib.Path) -> None: |
| 297 | root = _init_repo(tmp_path) |
| 298 | live = tmp_path.parent / "live-wt" |
| 299 | add_worktree(root=root, link_path=live, branch="feature/live") |
| 300 | |
| 301 | pruned = prune_worktrees(root=root) |
| 302 | assert len(pruned) == 0 |
| 303 | |
| 304 | worktrees = list_worktrees(root) |
| 305 | assert len(worktrees) == 2 |
| 306 | |
| 307 | def test_prune_empty_worktrees_dir_returns_empty( |
| 308 | self, tmp_path: pathlib.Path |
| 309 | ) -> None: |
| 310 | root = _init_repo(tmp_path) |
| 311 | pruned = prune_worktrees(root=root) |
| 312 | assert pruned == [] |
| 313 | |
| 314 | def test_prune_mixed_live_and_stale(self, tmp_path: pathlib.Path) -> None: |
| 315 | root = _init_repo(tmp_path) |
| 316 | live = tmp_path.parent / "live-wt2" |
| 317 | stale = tmp_path.parent / "stale-wt2" |
| 318 | add_worktree(root=root, link_path=live, branch="feature/live2") |
| 319 | add_worktree(root=root, link_path=stale, branch="feature/stale2") |
| 320 | import shutil |
| 321 | shutil.rmtree(stale) |
| 322 | |
| 323 | pruned = prune_worktrees(root=root) |
| 324 | assert str(stale) in pruned |
| 325 | assert str(live) not in pruned |
| 326 | # Live worktree still registered. |
| 327 | worktrees = list_worktrees(root) |
| 328 | assert any(w.path == live for w in worktrees) |
| 329 | |
| 330 | |
| 331 | # --------------------------------------------------------------------------- |
| 332 | # CLI integration — Typer CliRunner |
| 333 | # --------------------------------------------------------------------------- |
| 334 | |
| 335 | |
| 336 | class TestWorktreeCLI: |
| 337 | def test_worktree_list_shows_main(self, tmp_path: pathlib.Path) -> None: |
| 338 | root = _init_repo(tmp_path) |
| 339 | result = runner.invoke(cli, ["worktree", "list"], env=_env(root)) |
| 340 | assert result.exit_code == 0 |
| 341 | assert "[main]" in result.output |
| 342 | assert "main" in result.output |
| 343 | |
| 344 | def test_worktree_list_shows_linked(self, tmp_path: pathlib.Path) -> None: |
| 345 | root = _init_repo(tmp_path) |
| 346 | link_path = tmp_path.parent / "cli-linked-wt" |
| 347 | add_worktree(root=root, link_path=link_path, branch="feature/cli") |
| 348 | result = runner.invoke(cli, ["worktree", "list"], env=_env(root)) |
| 349 | assert result.exit_code == 0 |
| 350 | assert "feature/cli" in result.output |
| 351 | |
| 352 | def test_worktree_add_cli_success(self, tmp_path: pathlib.Path) -> None: |
| 353 | root = _init_repo(tmp_path) |
| 354 | link_path = tmp_path.parent / "cli-add-wt" |
| 355 | result = runner.invoke( |
| 356 | cli, |
| 357 | ["worktree", "add", str(link_path), "feature/cli-add"], |
| 358 | env=_env(root), |
| 359 | ) |
| 360 | assert result.exit_code == 0, result.output |
| 361 | assert "created" in result.output.lower() |
| 362 | assert link_path.is_dir() |
| 363 | |
| 364 | def test_worktree_remove_cli_success(self, tmp_path: pathlib.Path) -> None: |
| 365 | root = _init_repo(tmp_path) |
| 366 | link_path = tmp_path.parent / "cli-remove-wt" |
| 367 | add_worktree(root=root, link_path=link_path, branch="feature/cli-rm") |
| 368 | result = runner.invoke( |
| 369 | cli, ["worktree", "remove", str(link_path)], env=_env(root) |
| 370 | ) |
| 371 | assert result.exit_code == 0, result.output |
| 372 | assert "removed" in result.output.lower() |
| 373 | assert not link_path.exists() |
| 374 | |
| 375 | def test_worktree_prune_cli_no_stale(self, tmp_path: pathlib.Path) -> None: |
| 376 | root = _init_repo(tmp_path) |
| 377 | result = runner.invoke(cli, ["worktree", "prune"], env=_env(root)) |
| 378 | assert result.exit_code == 0 |
| 379 | assert "no stale" in result.output.lower() |
| 380 | |
| 381 | def test_worktree_prune_cli_removes_stale(self, tmp_path: pathlib.Path) -> None: |
| 382 | root = _init_repo(tmp_path) |
| 383 | link_path = tmp_path.parent / "cli-stale-wt" |
| 384 | add_worktree(root=root, link_path=link_path, branch="feature/cli-stale") |
| 385 | import shutil |
| 386 | shutil.rmtree(link_path) |
| 387 | result = runner.invoke(cli, ["worktree", "prune"], env=_env(root)) |
| 388 | assert result.exit_code == 0 |
| 389 | assert "pruned" in result.output.lower() |
| 390 | |
| 391 | def test_worktree_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: |
| 392 | result = runner.invoke( |
| 393 | cli, ["worktree", "list"], env={"MUSE_REPO_ROOT": str(tmp_path)} |
| 394 | ) |
| 395 | assert result.exit_code == 2 |