|
7 | 7 |
|
8 | 8 | import pytest |
9 | 9 |
|
10 | | -from noah_code.worktree import WorktreeError, WorktreeManager, repo_id_for |
| 10 | +from noah_code import worktree |
| 11 | +from noah_code.worktree import ( |
| 12 | + WorktreeError, |
| 13 | + WorktreeManager, |
| 14 | + git_common_dir, |
| 15 | + primary_checkout, |
| 16 | + repo_id_for, |
| 17 | +) |
11 | 18 |
|
12 | 19 |
|
13 | 20 | def _git(cwd: Path, *args: str) -> None: |
@@ -90,3 +97,260 @@ def wrapped(cwd: Path, *args: str): |
90 | 97 | text=True, |
91 | 98 | ) |
92 | 99 | assert "broken" not in leftover.stdout |
| 100 | + |
| 101 | + |
| 102 | +def test_removes_orphaned_worktree_with_missing_directory(tmp_path: Path) -> None: |
| 103 | + repo = _init_repo(tmp_path / "repo") |
| 104 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 105 | + created = manager.create("ghost") |
| 106 | + |
| 107 | + # Simulate a crash or manual rm -rf: registration exists, directory is gone. |
| 108 | + import shutil |
| 109 | + |
| 110 | + shutil.rmtree(created.directory) |
| 111 | + assert [item.name for item in manager.list()] == ["ghost"] |
| 112 | + |
| 113 | + info = manager.remove("ghost") |
| 114 | + |
| 115 | + assert info.name == "ghost" |
| 116 | + assert manager.list() == [] |
| 117 | + branch = subprocess.run( |
| 118 | + ["git", "show-ref", "--verify", "--quiet", "refs/heads/noah/ghost"], |
| 119 | + cwd=repo, |
| 120 | + check=False, |
| 121 | + ) |
| 122 | + assert branch.returncode != 0 |
| 123 | + |
| 124 | + |
| 125 | +def test_create_surfaces_add_failure_without_leftovers( |
| 126 | + tmp_path: Path, monkeypatch |
| 127 | +) -> None: |
| 128 | + repo = _init_repo(tmp_path / "repo") |
| 129 | + storage = tmp_path / "worktree" |
| 130 | + manager = WorktreeManager(repo, storage) |
| 131 | + real = worktree._git |
| 132 | + |
| 133 | + def wrapped(cwd: Path, *args: str): |
| 134 | + if args[:2] == ("worktree", "add"): |
| 135 | + return subprocess.CompletedProcess( |
| 136 | + args=["git", *args], returncode=1, stdout="", stderr="add blew up" |
| 137 | + ) |
| 138 | + return real(cwd, *args) |
| 139 | + |
| 140 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 141 | + with pytest.raises(WorktreeError, match="add blew up"): |
| 142 | + manager.create("doomed") |
| 143 | + assert manager.list() == [] |
| 144 | + |
| 145 | + |
| 146 | +def test_list_reports_git_failures(tmp_path: Path, monkeypatch) -> None: |
| 147 | + repo = _init_repo(tmp_path / "repo") |
| 148 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 149 | + real = worktree._git |
| 150 | + |
| 151 | + def wrapped(cwd: Path, *args: str): |
| 152 | + if args[:3] == ("worktree", "list", "--porcelain"): |
| 153 | + return subprocess.CompletedProcess( |
| 154 | + args=["git", *args], returncode=128, stdout="", stderr="fatal: broken" |
| 155 | + ) |
| 156 | + return real(cwd, *args) |
| 157 | + |
| 158 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 159 | + with pytest.raises(WorktreeError, match="fatal: broken"): |
| 160 | + manager.list() |
| 161 | + |
| 162 | + |
| 163 | +def test_list_parses_trailing_entry_without_blank_line( |
| 164 | + tmp_path: Path, monkeypatch |
| 165 | +) -> None: |
| 166 | + repo = _init_repo(tmp_path / "repo") |
| 167 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 168 | + manager.create("last-one") |
| 169 | + real = worktree._git |
| 170 | + |
| 171 | + def stripped(cwd: Path, *args: str): |
| 172 | + result = real(cwd, *args) |
| 173 | + if args[:3] == ("worktree", "list", "--porcelain"): |
| 174 | + result.stdout = result.stdout.rstrip("\n") # drop final blank line |
| 175 | + return result |
| 176 | + |
| 177 | + monkeypatch.setattr(worktree, "_git", stripped) |
| 178 | + assert [item.name for item in manager.list()] == ["last-one"] |
| 179 | + |
| 180 | + |
| 181 | +def test_remove_rejects_unknown_and_foreign_paths(tmp_path: Path) -> None: |
| 182 | + repo = _init_repo(tmp_path / "repo") |
| 183 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 184 | + |
| 185 | + # Relative path that resolves under the checkout but was never created. |
| 186 | + with pytest.raises(WorktreeError, match="not a Noah worktree"): |
| 187 | + manager.remove("nowhere/child") |
| 188 | + |
| 189 | + (repo / "plain-dir").mkdir() |
| 190 | + with pytest.raises(WorktreeError, match="not a Noah worktree"): |
| 191 | + manager.remove(repo / "plain-dir") |
| 192 | + |
| 193 | + |
| 194 | +def test_remove_fails_when_git_remove_fails(tmp_path: Path, monkeypatch) -> None: |
| 195 | + repo = _init_repo(tmp_path / "repo") |
| 196 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 197 | + created = manager.create("sticky") |
| 198 | + real = worktree._git |
| 199 | + |
| 200 | + def wrapped(cwd: Path, *args: str): |
| 201 | + if args[:2] == ("worktree", "remove"): |
| 202 | + return subprocess.CompletedProcess( |
| 203 | + args=["git", *args], returncode=1, stdout="", stderr="locked" |
| 204 | + ) |
| 205 | + return real(cwd, *args) |
| 206 | + |
| 207 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 208 | + with pytest.raises(WorktreeError, match="locked"): |
| 209 | + manager.remove(created.directory) |
| 210 | + branch = subprocess.run( |
| 211 | + ["git", "show-ref", "--verify", "--quiet", "refs/heads/noah/sticky"], |
| 212 | + cwd=repo, |
| 213 | + check=False, |
| 214 | + ) |
| 215 | + assert branch.returncode == 0 # untouched when removal fails |
| 216 | + |
| 217 | + |
| 218 | +def test_remove_fails_when_prune_fails(tmp_path: Path, monkeypatch) -> None: |
| 219 | + import shutil |
| 220 | + |
| 221 | + repo = _init_repo(tmp_path / "repo") |
| 222 | + manager = WorktreeManager(repo, tmp_path / "worktree") |
| 223 | + created = manager.create("ghost") |
| 224 | + shutil.rmtree(created.directory) |
| 225 | + real = worktree._git |
| 226 | + |
| 227 | + def wrapped(cwd: Path, *args: str): |
| 228 | + if args[:2] == ("worktree", "prune"): |
| 229 | + return subprocess.CompletedProcess( |
| 230 | + args=["git", *args], returncode=1, stdout="", stderr="prune refused" |
| 231 | + ) |
| 232 | + return real(cwd, *args) |
| 233 | + |
| 234 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 235 | + with pytest.raises(WorktreeError, match="prune refused"): |
| 236 | + manager.remove("ghost") |
| 237 | + |
| 238 | + |
| 239 | +def test_remove_cleans_directory_git_left_behind(tmp_path: Path, monkeypatch) -> None: |
| 240 | + repo = _init_repo(tmp_path / "repo") |
| 241 | + storage = tmp_path / "worktree" |
| 242 | + manager = WorktreeManager(repo, storage) |
| 243 | + created = manager.create("messy") |
| 244 | + real = worktree._git |
| 245 | + |
| 246 | + def wrapped(cwd: Path, *args: str): |
| 247 | + if args[:2] == ("worktree", "remove"): |
| 248 | + # Report success but leave the directory in place. |
| 249 | + return subprocess.CompletedProcess( |
| 250 | + args=["git", *args], returncode=0, stdout="", stderr="" |
| 251 | + ) |
| 252 | + return real(cwd, *args) |
| 253 | + |
| 254 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 255 | + info = manager.remove("messy") |
| 256 | + assert info.name == "messy" |
| 257 | + assert not created.directory.exists() |
| 258 | + |
| 259 | + |
| 260 | +def test_create_gives_up_after_repeated_name_collisions( |
| 261 | + tmp_path: Path, monkeypatch |
| 262 | +) -> None: |
| 263 | + repo = _init_repo(tmp_path / "repo") |
| 264 | + storage = tmp_path / "worktree" |
| 265 | + manager = WorktreeManager(repo, storage) |
| 266 | + occupied = storage / repo_id_for(repo) |
| 267 | + (occupied / "taken").mkdir(parents=True) |
| 268 | + (occupied / "taken-same-name").mkdir() |
| 269 | + monkeypatch.setattr(worktree, "_random_name", lambda: "same-name") |
| 270 | + |
| 271 | + with pytest.raises(WorktreeError, match="unique worktree name"): |
| 272 | + manager.create("taken") |
| 273 | + |
| 274 | + |
| 275 | +def test_hostile_names_stay_inside_storage_root(tmp_path: Path) -> None: |
| 276 | + repo = _init_repo(tmp_path / "repo") |
| 277 | + storage = (tmp_path / "worktree").resolve() |
| 278 | + manager = WorktreeManager(repo, storage) |
| 279 | + |
| 280 | + traversal = manager.create("../../etc/passwd-ish") |
| 281 | + collapsed = manager.create("a/../../b") |
| 282 | + slashed = manager.create("feature/one") |
| 283 | + blankish = manager.create(" ") |
| 284 | + |
| 285 | + for info in (traversal, collapsed, slashed, blankish): |
| 286 | + assert info.directory.parent == storage / repo_id_for(repo) |
| 287 | + assert "/" not in info.name |
| 288 | + assert info.directory.is_relative_to(storage) |
| 289 | + assert info.branch.startswith("noah/") |
| 290 | + |
| 291 | + assert collapsed.name == "a-b" |
| 292 | + assert slashed.name == "feature-one" |
| 293 | + # Blank names fall back to a random adjective-noun pair. |
| 294 | + head, _, tail = blankish.name.partition("-") |
| 295 | + assert head in worktree.ADJECTIVES |
| 296 | + assert tail in worktree.NOUNS |
| 297 | + |
| 298 | + |
| 299 | +def test_candidate_name_collisions_get_suffixed(tmp_path: Path) -> None: |
| 300 | + repo = _init_repo(tmp_path / "repo") |
| 301 | + storage = tmp_path / "worktree" |
| 302 | + manager = WorktreeManager(repo, storage) |
| 303 | + |
| 304 | + first = manager.create("dupe") |
| 305 | + second = manager.create("dupe") |
| 306 | + assert first.name != second.name |
| 307 | + assert second.name.startswith("dupe-") |
| 308 | + |
| 309 | + # A pre-existing branch with the target name also forces a suffix. |
| 310 | + _git(repo, "branch", "noah/taken") |
| 311 | + third = manager.create("taken") |
| 312 | + assert third.name != "taken" |
| 313 | + |
| 314 | + |
| 315 | +def test_git_common_dir_and_primary_checkout_fallbacks(tmp_path: Path, monkeypatch) -> None: |
| 316 | + outside = tmp_path / "not-a-repo" |
| 317 | + outside.mkdir() |
| 318 | + fake_ok_empty = subprocess.CompletedProcess( |
| 319 | + args=["git"], returncode=0, stdout="", stderr="" |
| 320 | + ) |
| 321 | + monkeypatch.setattr(worktree, "_git", lambda *_a, **_k: fake_ok_empty) |
| 322 | + |
| 323 | + assert git_common_dir(outside) is None |
| 324 | + assert primary_checkout(outside) == outside.resolve() |
| 325 | + assert WorktreeManager(outside, tmp_path / "worktree").list() == [] |
| 326 | + |
| 327 | + from noah_code.worktree import family_id, infer_worktree_name, worktree_storage_root |
| 328 | + |
| 329 | + assert family_id(outside, fallback="fallback-id") == "fallback-id" |
| 330 | + session_dir = tmp_path / "sessions" / "abc" |
| 331 | + assert infer_worktree_name(session_dir, worktree_storage_root(session_dir)) == "" |
| 332 | + |
| 333 | + |
| 334 | +def test_rollback_cleans_leftover_directory(tmp_path: Path, monkeypatch) -> None: |
| 335 | + repo = _init_repo(tmp_path / "repo") |
| 336 | + storage = tmp_path / "worktree" |
| 337 | + manager = WorktreeManager(repo, storage) |
| 338 | + real = worktree._git |
| 339 | + |
| 340 | + def wrapped(cwd: Path, *args: str): |
| 341 | + if args[:1] == ("reset",): |
| 342 | + return subprocess.CompletedProcess( |
| 343 | + args=["git", *args], returncode=1, stdout="", stderr="reset failed" |
| 344 | + ) |
| 345 | + if args[:2] == ("worktree", "remove"): |
| 346 | + # Pretend the rollback remove succeeded but left files behind. |
| 347 | + return subprocess.CompletedProcess( |
| 348 | + args=["git", *args], returncode=0, stdout="", stderr="" |
| 349 | + ) |
| 350 | + return real(cwd, *args) |
| 351 | + |
| 352 | + monkeypatch.setattr(worktree, "_git", wrapped) |
| 353 | + with pytest.raises(WorktreeError, match="reset failed"): |
| 354 | + manager.create("crumbs") |
| 355 | + # _rollback rmtree'd the leftover directory. |
| 356 | + assert not (storage / repo_id_for(repo) / "crumbs").exists() |
0 commit comments