gabriel / muse public
test_cmd_checkout.py python
1,694 lines 73.4 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
1 """Tests for ``muse checkout``.
2
3 Coverage tiers
4 --------------
5 Unit — parser flags, dead-code removal, docstring schema.
6 Integration — switch, create, already_on, detach, --dry-run, conflict resolution.
7 End-to-end — full CLI invocations: text and JSON output, all operations.
8 Security — ANSI injection in target, error routing to stderr.
9 Stress — checkout under high file counts, concurrent checkouts.
10 """
11
12 from __future__ import annotations
13
14 import json
15 import os
16 import pathlib
17 import subprocess
18 import threading
19 import time
20 from typing import TYPE_CHECKING
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner, InvokeResult
25 from muse.core._types import short_id
26 from muse.core.store import get_head_commit_id, read_current_branch
27 from muse.cli.config import read_branch_meta
28
29 if TYPE_CHECKING:
30 import argparse
31
32 runner = CliRunner()
33
34 # ──────────────────────────────────────────────────────────────────────────────
35 # Helpers
36 # ──────────────────────────────────────────────────────────────────────────────
37
38
39 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
40 saved = os.getcwd()
41 try:
42 os.chdir(repo)
43 return runner.invoke(None, args)
44 finally:
45 os.chdir(saved)
46
47
48 def _checkout(repo: pathlib.Path, *extra: str) -> InvokeResult:
49 return _invoke(repo, ["checkout", *extra])
50
51
52 def _commit(repo: pathlib.Path, *extra: str) -> InvokeResult:
53 return _invoke(repo, ["commit", *extra])
54
55
56 def _branch(repo: pathlib.Path, *extra: str) -> InvokeResult:
57 return _invoke(repo, ["branch", *extra])
58
59
60 @pytest.fixture()
61 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
62 """Initialised repo with one commit on ``main``."""
63 saved = os.getcwd()
64 try:
65 os.chdir(tmp_path)
66 runner.invoke(None, ["init"])
67 finally:
68 os.chdir(saved)
69 (tmp_path / "a.py").write_text("x = 1\n")
70 _commit(tmp_path, "-m", "initial")
71 return tmp_path
72
73
74 @pytest.fixture()
75 def two_branch_repo(repo: pathlib.Path) -> pathlib.Path:
76 """Repo with ``main`` and ``feat`` branches, each with unique content."""
77 _branch(repo, "feat")
78 _checkout(repo, "feat")
79 (repo / "feat.py").write_text("f = 1\n")
80 _commit(repo, "-m", "feat commit")
81 _checkout(repo, "main")
82 return repo
83
84
85 # ──────────────────────────────────────────────────────────────────────────────
86 # Unit — parser flags
87 # ──────────────────────────────────────────────────────────────────────────────
88
89
90 class TestRegisterFlags:
91 def _parse(self, *args: str) -> "argparse.Namespace":
92 import argparse
93
94 from muse.cli.commands.checkout import register
95
96 p = argparse.ArgumentParser()
97 sub = p.add_subparsers()
98 register(sub)
99 return p.parse_args(["checkout", *args])
100
101 def test_default_json_out_is_false(self) -> None:
102 ns = self._parse("main")
103 assert ns.json_out is False
104
105 def test_json_flag_sets_json_out(self) -> None:
106 ns = self._parse("main", "--json")
107 assert ns.json_out is True
108
109 def test_j_shorthand_sets_json_out(self) -> None:
110 ns = self._parse("main", "-j")
111 assert ns.json_out is True
112
113 def test_create_flag(self) -> None:
114 ns = self._parse("-b", "new")
115 assert ns.create is True
116
117 def test_force_flag(self) -> None:
118 ns = self._parse("main", "--force")
119 assert ns.force is True
120
121 def test_force_short_flag(self) -> None:
122 ns = self._parse("main", "-f")
123 assert ns.force is True
124
125 def test_dry_run_flag(self) -> None:
126 ns = self._parse("main", "--dry-run")
127 assert ns.dry_run is True
128
129 def test_dry_run_short_flag(self) -> None:
130 ns = self._parse("main", "-n")
131 assert ns.dry_run is True
132
133 def test_dry_run_default_false(self) -> None:
134 ns = self._parse("main")
135 assert ns.dry_run is False
136
137 def test_ours_flag(self) -> None:
138 ns = self._parse("--ours", "file.py")
139 assert ns.resolve_ours is True
140
141 def test_theirs_flag(self) -> None:
142 ns = self._parse("--theirs", "file.py")
143 assert ns.resolve_theirs is True
144
145 def test_all_flag(self) -> None:
146 ns = self._parse("--ours", "--all")
147 assert ns.resolve_all is True
148
149 def test_target_optional(self) -> None:
150 ns = self._parse()
151 assert ns.target is None
152
153
154 # ──────────────────────────────────────────────────────────────────────────────
155 # Unit — dead-code removal
156 # ──────────────────────────────────────────────────────────────────────────────
157
158
159 class TestDeadCodeRemoved:
160 def test_read_current_branch_wrapper_removed(self) -> None:
161 import muse.cli.commands.checkout as m
162
163 assert not hasattr(m, "_read_current_branch"), (
164 "_read_current_branch was a dead one-liner wrapper and must be deleted"
165 )
166
167 def test_inline_sanitize_display_import_removed(self) -> None:
168 import inspect
169
170 import muse.cli.commands.checkout as m
171
172 src = inspect.getsource(m.run)
173 assert "sanitize_display as _sd" not in src, (
174 "The inline 'from muse.core.validation import sanitize_display as _sd' "
175 "inside run() was a redundant re-import of the module-level sanitize_display"
176 )
177
178
179 # ──────────────────────────────────────────────────────────────────────────────
180 # Integration — SWITCH (existing branch)
181 # ──────────────────────────────────────────────────────────────────────────────
182
183
184 class TestSwitch:
185 def test_switch_exits_0(self, two_branch_repo: pathlib.Path) -> None:
186 result = _checkout(two_branch_repo, "feat")
187 assert result.exit_code == 0
188
189 def test_switch_changes_branch(self, two_branch_repo: pathlib.Path) -> None:
190 _checkout(two_branch_repo, "feat")
191 assert read_current_branch(two_branch_repo) == "feat"
192
193 def test_switch_text_output(self, two_branch_repo: pathlib.Path) -> None:
194 result = _checkout(two_branch_repo, "feat")
195 assert "feat" in result.output
196 assert "Switched" in result.output
197
198 def test_switch_json_schema(self, two_branch_repo: pathlib.Path) -> None:
199 result = _checkout(two_branch_repo, "feat", "--json")
200 data = json.loads(result.output)
201 assert data["action"] == "switched"
202 assert data["branch"] == "feat"
203 assert data["from_branch"] == "main"
204 assert "commit_id" in data
205 assert data.get("dry_run") is False
206
207 def test_switch_restores_files(self, two_branch_repo: pathlib.Path) -> None:
208 """Files unique to ``feat`` appear after checkout and disappear on return."""
209 _checkout(two_branch_repo, "feat")
210 assert (two_branch_repo / "feat.py").exists()
211 _checkout(two_branch_repo, "main")
212 assert not (two_branch_repo / "feat.py").exists()
213
214 def test_switch_to_nonexistent_exits_1(self, repo: pathlib.Path) -> None:
215 result = _checkout(repo, "does-not-exist")
216 assert result.exit_code == 1
217
218 def test_switch_error_to_stderr(self, repo: pathlib.Path) -> None:
219 result = _checkout(repo, "ghost")
220 assert result.exit_code == 1
221 # Error must appear in stderr, not exclusively stdout
222 assert "not a branch" in (result.stderr or "").lower()
223
224
225 # ──────────────────────────────────────────────────────────────────────────────
226 # Integration — ALREADY_ON
227 # ──────────────────────────────────────────────────────────────────────────────
228
229
230 class TestAlreadyOn:
231 def test_already_on_exits_0(self, repo: pathlib.Path) -> None:
232 result = _checkout(repo, "main")
233 assert result.exit_code == 0
234
235 def test_already_on_text(self, repo: pathlib.Path) -> None:
236 result = _checkout(repo, "main")
237 assert "Already on" in result.output
238
239 def test_already_on_json_schema(self, repo: pathlib.Path) -> None:
240 result = _checkout(repo, "main", "--json")
241 data = json.loads(result.output)
242 assert data["action"] == "already_on"
243 assert data["branch"] == "main"
244 assert data["from_branch"] == "main"
245 assert "commit_id" in data
246
247
248 # ──────────────────────────────────────────────────────────────────────────────
249 # Integration — FORCE on current branch (working-tree restore)
250 # ──────────────────────────────────────────────────────────────────────────────
251
252
253 class TestForceOnCurrentBranch:
254 """checkout --force <current-branch> must restore the working tree to HEAD.
255
256 Regression: previously this was a no-op ('Already on main') regardless of
257 --force. Git's behaviour is to restore missing/modified tracked files even
258 when the branch is already current.
259 """
260
261 def test_force_current_branch_exits_0(self, repo: pathlib.Path) -> None:
262 result = _checkout(repo, "--force", "main")
263 assert result.exit_code == 0
264
265 def test_force_current_branch_restores_deleted_file(self, repo: pathlib.Path) -> None:
266 (repo / "a.py").unlink()
267 assert not (repo / "a.py").exists()
268 _checkout(repo, "--force", "main")
269 assert (repo / "a.py").exists(), "force checkout must restore deleted tracked file"
270
271 def test_force_current_branch_restores_modified_file(self, repo: pathlib.Path) -> None:
272 original = (repo / "a.py").read_text()
273 (repo / "a.py").write_text("corrupted content\n")
274 _checkout(repo, "--force", "main")
275 assert (repo / "a.py").read_text() == original, (
276 "force checkout must restore modified tracked file to HEAD content"
277 )
278
279 def test_force_current_branch_text_output(self, repo: pathlib.Path) -> None:
280 result = _checkout(repo, "--force", "main")
281 assert "restored" in result.output
282
283 def test_force_current_branch_json_action(self, repo: pathlib.Path) -> None:
284 result = _checkout(repo, "--force", "main", "--json")
285 data = json.loads(result.output)
286 assert data["action"] == "restored"
287 assert data["branch"] == "main"
288
289 def test_force_current_branch_dry_run_does_not_restore(
290 self, repo: pathlib.Path
291 ) -> None:
292 (repo / "a.py").unlink()
293 _checkout(repo, "--force", "--dry-run", "main")
294 assert not (repo / "a.py").exists(), (
295 "--dry-run must not actually restore files"
296 )
297
298 def test_force_current_branch_dry_run_json(self, repo: pathlib.Path) -> None:
299 result = _checkout(repo, "--force", "--dry-run", "main", "--json")
300 data = json.loads(result.output)
301 assert data["dry_run"] is True
302 assert data["action"] == "restored"
303
304 def test_without_force_still_noop_on_current_branch(
305 self, repo: pathlib.Path
306 ) -> None:
307 """Without --force, checkout on current branch is still a no-op."""
308 result = _checkout(repo, "main")
309 assert "Already on" in result.output
310
311
312 # ──────────────────────────────────────────────────────────────────────────────
313 # Integration — CREATE (-b)
314 # ──────────────────────────────────────────────────────────────────────────────
315
316
317 class TestCreate:
318 def test_create_exits_0(self, repo: pathlib.Path) -> None:
319 result = _checkout(repo, "-b", "new-branch")
320 assert result.exit_code == 0
321
322 def test_create_switches_to_new_branch(self, repo: pathlib.Path) -> None:
323 _checkout(repo, "-b", "new-branch")
324 assert read_current_branch(repo) == "new-branch"
325
326 def test_create_text_output(self, repo: pathlib.Path) -> None:
327 result = _checkout(repo, "-b", "my-branch")
328 assert "my-branch" in result.output
329
330 def test_create_json_schema(self, repo: pathlib.Path) -> None:
331 result = _checkout(repo, "-b", "json-branch", "--json")
332 data = json.loads(result.output)
333 assert data["action"] == "created"
334 assert data["branch"] == "json-branch"
335 assert data["from_branch"] == "main"
336 assert "commit_id" in data
337 assert data.get("dry_run") is False
338
339 def test_create_duplicate_exits_1(self, repo: pathlib.Path) -> None:
340 _checkout(repo, "-b", "dup")
341 _checkout(repo, "main")
342 result = _checkout(repo, "-b", "dup")
343 assert result.exit_code == 1
344
345 def test_create_duplicate_error_to_stderr(self, repo: pathlib.Path) -> None:
346 _checkout(repo, "-b", "dup2")
347 _checkout(repo, "main")
348 result = _checkout(repo, "-b", "dup2")
349 # Error must appear in stderr
350 assert "already exists" in (result.stderr or "").lower()
351
352 def test_create_invalid_name_exits_1(self, repo: pathlib.Path) -> None:
353 result = _checkout(repo, "-b", "bad..name")
354 assert result.exit_code == 1
355
356 def test_create_invalid_name_error_to_stderr(self, repo: pathlib.Path) -> None:
357 result = _checkout(repo, "-b", "bad..name")
358 assert "Invalid" in (result.stderr or "")
359
360 def test_create_with_dirty_workdir_succeeds(self, repo: pathlib.Path) -> None:
361 """checkout -b must succeed even with uncommitted changes.
362
363 Creating a new branch starts at the current HEAD — no file content
364 changes, so dirty tracked files cannot be overwritten. Blocking
365 here forces an unnecessary shelf/pop dance.
366 """
367 # Dirty the tracked file without committing
368 (repo / "a.py").write_text("x = 2\n")
369 result = _checkout(repo, "-b", "task/dirty-ok")
370 assert result.exit_code == 0
371 assert read_current_branch(repo) == "task/dirty-ok"
372 # The dirty file must still be present (not lost)
373 assert (repo / "a.py").read_text() == "x = 2\n"
374
375
376 # ──────────────────────────────────────────────────────────────────────────────
377 # Integration — DIRTY WORKDIR BLEED-THROUGH (regression)
378 # ──────────────────────────────────────────────────────────────────────────────
379
380
381 @pytest.fixture()
382 def shared_file_repo(tmp_path: pathlib.Path) -> pathlib.Path:
383 """Repo where ``main`` and ``feat`` share the same file at the same content.
384
385 Both branches commit ``shared.py`` with identical content. This is the
386 scenario that triggers the bleed-through bug: a dirty ``shared.py`` would
387 not appear in the delta between the two snapshots, so the old code let it
388 through silently instead of refusing the checkout.
389 """
390 saved = os.getcwd()
391 try:
392 os.chdir(tmp_path)
393 runner.invoke(None, ["init"])
394 finally:
395 os.chdir(saved)
396 (tmp_path / "shared.py").write_text("x = 1\n")
397 _commit(tmp_path, "-m", "initial")
398 # Create feat branch — shared.py is the same on both branches
399 _checkout(tmp_path, "-b", "feat")
400 (tmp_path / "feat_only.py").write_text("f = 1\n")
401 _commit(tmp_path, "-m", "feat commit")
402 _checkout(tmp_path, "main")
403 return tmp_path
404
405
406 class TestDirtyWorkdirBleedThrough:
407 """checkout must refuse when the working tree is dirty.
408
409 Regression: the old ``require_clean_workdir`` only blocked files that
410 the target branch would *overwrite*. Files modified locally but
411 identical on both branches were silently carried through, causing
412 dirty working-tree state to appear on branches the user never
413 touched — exactly the bug that left 80+ modified files on ``main``.
414
415 The fix: always refuse checkout on any dirty tracked file. Users
416 must explicitly commit, shelf, use ``--autoshelf``, or ``--force``.
417 """
418
419 def test_dirty_shared_file_blocks_checkout(
420 self, shared_file_repo: pathlib.Path
421 ) -> None:
422 """A file modified locally but identical on both branches must block checkout."""
423 (shared_file_repo / "shared.py").write_text("x = 999\n")
424 result = _checkout(shared_file_repo, "feat")
425 assert result.exit_code == 1, (
426 "checkout must refuse when shared.py is dirty, "
427 "even though main and feat have the same committed version"
428 )
429
430 def test_dirty_shared_file_error_names_file(
431 self, shared_file_repo: pathlib.Path
432 ) -> None:
433 """The refusal message must name the dirty file."""
434 (shared_file_repo / "shared.py").write_text("x = 999\n")
435 result = _checkout(shared_file_repo, "feat")
436 assert "shared.py" in (result.stderr or ""), (
437 "error message must name the dirty file"
438 )
439
440 def test_dirty_shared_file_json_error(
441 self, shared_file_repo: pathlib.Path
442 ) -> None:
443 """JSON mode must emit a machine-readable error, not bleed through."""
444 (shared_file_repo / "shared.py").write_text("x = 999\n")
445 result = _checkout(shared_file_repo, "feat", "--json")
446 assert result.exit_code == 1
447 data = json.loads(result.output)
448 assert data["error"] == "dirty_workdir"
449 assert "shared.py" in data["files"]
450
451 def test_dirty_shared_file_not_silently_moved(
452 self, shared_file_repo: pathlib.Path
453 ) -> None:
454 """After a refused checkout, we must still be on the original branch."""
455 (shared_file_repo / "shared.py").write_text("x = 999\n")
456 _checkout(shared_file_repo, "feat")
457 assert read_current_branch(shared_file_repo) == "main", (
458 "failed checkout must not switch the branch"
459 )
460
461 def test_deleted_shared_file_blocks_checkout(
462 self, shared_file_repo: pathlib.Path
463 ) -> None:
464 """A tracked file deleted locally but present on both branches must block checkout."""
465 (shared_file_repo / "shared.py").unlink()
466 result = _checkout(shared_file_repo, "feat")
467 assert result.exit_code == 1
468
469 def test_force_bypasses_dirty_check(
470 self, shared_file_repo: pathlib.Path
471 ) -> None:
472 """``--force`` must still bypass the dirty check (discards local changes)."""
473 (shared_file_repo / "shared.py").write_text("x = 999\n")
474 result = _checkout(shared_file_repo, "--force", "feat")
475 assert result.exit_code == 0
476 assert read_current_branch(shared_file_repo) == "feat"
477 # --force restores the target branch's version, discarding local edit
478 assert (shared_file_repo / "shared.py").read_text() == "x = 1\n"
479
480 def test_autoshelf_bypasses_dirty_check(
481 self, shared_file_repo: pathlib.Path
482 ) -> None:
483 """``--autoshelf`` must shelve the dirty file and switch cleanly."""
484 (shared_file_repo / "shared.py").write_text("x = 999\n")
485 result = _checkout(shared_file_repo, "--autoshelf", "feat")
486 assert result.exit_code == 0
487 assert read_current_branch(shared_file_repo) == "feat"
488
489 def test_untracked_file_does_not_block_checkout(
490 self, shared_file_repo: pathlib.Path
491 ) -> None:
492 """A brand-new untracked file must never block checkout."""
493 (shared_file_repo / "new_untracked.py").write_text("new = 1\n")
494 result = _checkout(shared_file_repo, "feat")
495 assert result.exit_code == 0, (
496 "untracked files are never in any snapshot so checkout must allow them"
497 )
498
499 def test_create_branch_allows_dirty_workdir(
500 self, shared_file_repo: pathlib.Path
501 ) -> None:
502 """``-b`` (create) must still succeed with dirty files — no snapshot change."""
503 (shared_file_repo / "shared.py").write_text("x = 999\n")
504 result = _checkout(shared_file_repo, "-b", "task/new")
505 assert result.exit_code == 0, (
506 "creating a branch does not change any file content; dirty tree is fine"
507 )
508
509
510 # ──────────────────────────────────────────────────────────────────────────────
511 # Integration — DETACH HEAD
512 # ──────────────────────────────────────────────────────────────────────────────
513
514
515 class TestDetach:
516 def test_detach_full_sha_exits_0(self, repo: pathlib.Path) -> None:
517 sha = get_head_commit_id(repo, "main")
518 assert sha is not None
519 result = _checkout(repo, sha)
520 assert result.exit_code == 0
521
522 def test_detach_full_sha_text_output(self, repo: pathlib.Path) -> None:
523 sha = get_head_commit_id(repo, "main")
524 assert sha is not None
525 result = _checkout(repo, sha)
526 # Output shows sha256: prefix + 8 hex chars — canonical and algorithm-identifying.
527 assert sha[:len("sha256:") + 8] in result.output
528
529 def test_detach_full_sha_json_schema(self, repo: pathlib.Path) -> None:
530 sha = get_head_commit_id(repo, "main")
531 assert sha is not None
532 result = _checkout(repo, sha, "--json")
533 data = json.loads(result.output)
534 assert data["action"] == "detached"
535 assert data["branch"] is None
536 assert data["commit_id"] == sha
537 assert data["from_branch"] == "main"
538 assert data.get("dry_run") is False
539
540 def test_detach_partial_sha_exits_0(self, repo: pathlib.Path) -> None:
541 sha = get_head_commit_id(repo, "main")
542 assert sha is not None
543 # Pass bare hex prefix to checkout — the command resolves it
544 hex_prefix = short_id(sha, strip=True)
545 result = _checkout(repo, hex_prefix)
546 assert result.exit_code == 0
547
548 def test_detach_partial_sha_points_to_correct_commit(self, repo: pathlib.Path) -> None:
549 """A partial SHA must resolve to the correct commit, not be treated as a branch."""
550 from muse.core.store import get_commits_for_branch, read_current_branch
551 from muse.core.repo import read_repo_id
552
553 (repo / "b.py").write_text("b=1\n")
554 _commit(repo, "-m", "second")
555
556 repo_id = read_repo_id(repo)
557 branch = read_current_branch(repo)
558 commits = get_commits_for_branch(repo, repo_id, branch)
559 first_sha = commits[-1].commit_id # oldest
560
561 # Pass bare hex prefix to checkout — the command resolves it
562 hex_prefix = short_id(first_sha, strip=True)
563 result = _checkout(repo, hex_prefix)
564 assert result.exit_code == 0
565 assert first_sha[:len("sha256:") + 8] in result.output
566
567 def test_detach_bad_ref_exits_1(self, repo: pathlib.Path) -> None:
568 result = _checkout(repo, "deadbeefdeadbeef")
569 assert result.exit_code == 1
570
571 def test_detach_error_to_stderr(self, repo: pathlib.Path) -> None:
572 result = _checkout(repo, "deadbeefdeadbeef")
573 assert "not a branch" in (result.stderr or "").lower()
574
575
576 # ──────────────────────────────────────────────────────────────────────────────
577 # Integration — DRY-RUN
578 # ──────────────────────────────────────────────────────────────────────────────
579
580
581 class TestDryRun:
582 def test_dry_run_switch_exits_0(self, two_branch_repo: pathlib.Path) -> None:
583 result = _checkout(two_branch_repo, "--dry-run", "feat")
584 assert result.exit_code == 0
585
586 def test_dry_run_does_not_switch_branch(self, two_branch_repo: pathlib.Path) -> None:
587 _checkout(two_branch_repo, "--dry-run", "feat")
588 assert read_current_branch(two_branch_repo) == "main"
589
590 def test_dry_run_text_says_would(self, two_branch_repo: pathlib.Path) -> None:
591 result = _checkout(two_branch_repo, "--dry-run", "feat")
592 assert "Would" in result.output
593 assert "feat" in result.output
594
595 def test_dry_run_json_schema(self, two_branch_repo: pathlib.Path) -> None:
596 result = _checkout(two_branch_repo, "--dry-run", "feat", "--json")
597 data = json.loads(result.output)
598 assert data["dry_run"] is True
599 assert data["action"] == "switched"
600 assert data["branch"] == "feat"
601 assert data["from_branch"] == "main"
602
603 def test_dry_run_does_not_restore_files(self, two_branch_repo: pathlib.Path) -> None:
604 """feat.py exists only on feat branch; dry-run must not create it on main."""
605 _checkout(two_branch_repo, "--dry-run", "feat")
606 assert not (two_branch_repo / "feat.py").exists()
607
608 def test_dry_run_create_exits_0(self, repo: pathlib.Path) -> None:
609 result = _checkout(repo, "-b", "dry-branch", "--dry-run")
610 assert result.exit_code == 0
611
612 def test_dry_run_create_does_not_create_branch(self, repo: pathlib.Path) -> None:
613 _checkout(repo, "-b", "dry-branch", "--dry-run")
614 result = _invoke(repo, ["branch", "--json"])
615 names = [b["name"] for b in json.loads(result.output)]
616 assert "dry-branch" not in names
617
618 def test_dry_run_create_json_schema(self, repo: pathlib.Path) -> None:
619 result = _checkout(repo, "-b", "dry-new", "--dry-run", "--json")
620 data = json.loads(result.output)
621 assert data["dry_run"] is True
622 assert data["action"] == "created"
623 assert data["from_branch"] == "main"
624
625 def test_dry_run_detach_exits_0(self, repo: pathlib.Path) -> None:
626 sha = get_head_commit_id(repo, "main")
627 assert sha is not None
628 result = _checkout(repo, "--dry-run", sha)
629 assert result.exit_code == 0
630
631 def test_dry_run_detach_does_not_detach(self, repo: pathlib.Path) -> None:
632 sha = get_head_commit_id(repo, "main")
633 assert sha is not None
634 _checkout(repo, "--dry-run", sha)
635 assert read_current_branch(repo) == "main"
636
637 def test_dry_run_detach_json(self, repo: pathlib.Path) -> None:
638 sha = get_head_commit_id(repo, "main")
639 assert sha is not None
640 result = _checkout(repo, "--dry-run", sha, "--json")
641 data = json.loads(result.output)
642 assert data["dry_run"] is True
643 assert data["action"] == "detached"
644 assert data["branch"] is None
645
646 def test_dry_run_nonexistent_branch_exits_1(self, repo: pathlib.Path) -> None:
647 result = _checkout(repo, "--dry-run", "no-such-branch")
648 assert result.exit_code == 1
649
650 def test_dry_run_already_on_exits_0(self, repo: pathlib.Path) -> None:
651 result = _checkout(repo, "--dry-run", "main")
652 assert result.exit_code == 0
653
654 def test_dry_run_already_on_json(self, repo: pathlib.Path) -> None:
655 result = _checkout(repo, "--dry-run", "main", "--json")
656 data = json.loads(result.output)
657 assert data["dry_run"] is True
658 assert data["action"] == "already_on"
659
660
661 # ──────────────────────────────────────────────────────────────────────────────
662 # Integration — JSON schema consistency
663 # ──────────────────────────────────────────────────────────────────────────────
664
665
666 class TestJsonSchema:
667 REQUIRED_KEYS = {"action", "branch", "commit_id", "from_branch", "dry_run"}
668
669 def test_create_has_all_keys(self, repo: pathlib.Path) -> None:
670 result = _checkout(repo, "-b", "k-test", "--json")
671 data = json.loads(result.output)
672 missing = self.REQUIRED_KEYS - set(data)
673 assert not missing, f"Missing keys in 'created' JSON: {missing}"
674
675 def test_switch_has_all_keys(self, two_branch_repo: pathlib.Path) -> None:
676 result = _checkout(two_branch_repo, "feat", "--json")
677 data = json.loads(result.output)
678 missing = self.REQUIRED_KEYS - set(data)
679 assert not missing, f"Missing keys in 'switched' JSON: {missing}"
680
681 def test_already_on_has_all_keys(self, repo: pathlib.Path) -> None:
682 result = _checkout(repo, "main", "--json")
683 data = json.loads(result.output)
684 missing = self.REQUIRED_KEYS - set(data)
685 assert not missing, f"Missing keys in 'already_on' JSON: {missing}"
686
687 def test_detach_has_all_keys(self, repo: pathlib.Path) -> None:
688 sha = get_head_commit_id(repo, "main")
689 assert sha is not None
690 result = _checkout(repo, sha, "--json")
691 data = json.loads(result.output)
692 missing = self.REQUIRED_KEYS - set(data)
693 assert not missing, f"Missing keys in 'detached' JSON: {missing}"
694
695 def test_detach_branch_is_null(self, repo: pathlib.Path) -> None:
696 sha = get_head_commit_id(repo, "main")
697 assert sha is not None
698 result = _checkout(repo, sha, "--json")
699 data = json.loads(result.output)
700 assert data["branch"] is None
701
702 def test_from_branch_reflects_previous(self, two_branch_repo: pathlib.Path) -> None:
703 _checkout(two_branch_repo, "feat")
704 result = _checkout(two_branch_repo, "main", "--json")
705 data = json.loads(result.output)
706 assert data["from_branch"] == "feat"
707
708
709 # ──────────────────────────────────────────────────────────────────────────────
710 # Integration — validation
711 # ──────────────────────────────────────────────────────────────────────────────
712
713
714 class TestValidation:
715 def test_no_target_exits_1(self, repo: pathlib.Path) -> None:
716 result = _checkout(repo)
717 assert result.exit_code == 1
718
719 def test_no_target_error_to_stderr(self, repo: pathlib.Path) -> None:
720 result = _checkout(repo)
721 assert "Specify" in (result.stderr or "")
722
723 def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None:
724 result = _checkout(repo, "main", "--no-such-flag")
725 assert result.exit_code != 0
726
727 def test_ours_without_theirs_context_exits_1(self, repo: pathlib.Path) -> None:
728 result = _checkout(repo, "--ours", "file.py")
729 assert result.exit_code == 1
730
731 def test_ours_and_theirs_together_exits_1(self, repo: pathlib.Path) -> None:
732 result = _checkout(repo, "--ours", "--theirs", "--all")
733 assert result.exit_code == 1
734
735
736 # ──────────────────────────────────────────────────────────────────────────────
737 # Security — ANSI injection
738 # ──────────────────────────────────────────────────────────────────────────────
739
740
741 class TestSecurityAnsi:
742 def _has_ansi(self, s: str) -> bool:
743 return "\x1b[" in s
744
745 def test_ansi_in_target_sanitized(self, repo: pathlib.Path) -> None:
746 result = _checkout(repo, "\x1b[31mevil\x1b[0m")
747 assert not self._has_ansi(result.output)
748
749 def test_ansi_in_create_name_sanitized(self, repo: pathlib.Path) -> None:
750 result = _checkout(repo, "-b", "\x1b[31mevil\x1b[0m")
751 assert not self._has_ansi(result.output)
752
753 def test_error_not_a_branch_sanitized(self, repo: pathlib.Path) -> None:
754 """The 'not a branch' error message must not echo raw ANSI from target."""
755 result = _checkout(repo, "\x1b[31mnotabranch\x1b[0m")
756 assert not self._has_ansi(result.output)
757 assert not self._has_ansi(result.stderr or "")
758
759 def test_all_errors_to_stderr(self, repo: pathlib.Path) -> None:
760 """Every ❌ error must go to stderr; stderr must contain the error."""
761 error_cases = [
762 ["ghost"],
763 ["-b", "bad..name"],
764 ]
765 for case in error_cases:
766 result = _checkout(repo, *case)
767 assert result.exit_code != 0, f"Expected failure for args {case}"
768 assert "❌" in (result.stderr or ""), (
769 f"Error not in stderr for args {case}: stderr={result.stderr!r}"
770 )
771
772
773 # ──────────────────────────────────────────────────────────────────────────────
774 # Integration — conflict resolution
775 # ──────────────────────────────────────────────────────────────────────────────
776
777
778 class TestConflictResolution:
779 def _setup_merge_conflict(
780 self, repo: pathlib.Path
781 ) -> tuple[str, str]:
782 """Create a merge conflict on ``repo``. Returns (ours_commit, theirs_commit)."""
783 # ours: commit on main
784 (repo / "shared.py").write_text("x = 1\n")
785 _commit(repo, "-m", "main: set x=1")
786 ours_cid = get_head_commit_id(repo, "main") or ""
787
788 # theirs: commit on feature branch
789 _branch(repo, "feat2")
790 _invoke(repo, ["checkout", "feat2"])
791 (repo / "shared.py").write_text("x = 2\n")
792 _commit(repo, "-m", "feat: set x=2")
793 theirs_cid = get_head_commit_id(repo, "feat2") or ""
794
795 _invoke(repo, ["checkout", "main"])
796 # Force a merge conflict via merge_engine internals
797 from muse.core.merge_engine import write_merge_state
798
799 write_merge_state(
800 repo,
801 base_commit="",
802 ours_commit=ours_cid,
803 theirs_commit=theirs_cid,
804 conflict_paths=["shared.py"],
805 other_branch="feat2",
806 )
807 return ours_cid, theirs_cid
808
809 def test_ours_no_merge_state_exits_1(self, repo: pathlib.Path) -> None:
810 result = _checkout(repo, "--ours", "file.py")
811 assert result.exit_code == 1
812
813 def test_theirs_no_merge_state_exits_1(self, repo: pathlib.Path) -> None:
814 result = _checkout(repo, "--theirs", "file.py")
815 assert result.exit_code == 1
816
817 def test_ours_and_theirs_both_exits_1(self, repo: pathlib.Path) -> None:
818 result = _checkout(repo, "--ours", "--theirs", "--all")
819 assert result.exit_code == 1
820
821 def test_ours_resolves_conflict(self, repo: pathlib.Path) -> None:
822 self._setup_merge_conflict(repo)
823 result = _checkout(repo, "--ours", "shared.py")
824 assert result.exit_code == 0
825
826 def test_theirs_resolves_conflict(self, repo: pathlib.Path) -> None:
827 self._setup_merge_conflict(repo)
828 result = _checkout(repo, "--theirs", "shared.py")
829 assert result.exit_code == 0
830
831 def test_resolve_all_ours_json(self, repo: pathlib.Path) -> None:
832 self._setup_merge_conflict(repo)
833 result = _checkout(repo, "--ours", "--all", "--json")
834 assert result.exit_code == 0
835 data = json.loads(result.output)
836 assert data["action"] == "conflict_resolved_all"
837 assert data["side"] == "ours"
838 assert "resolved_count" in data
839 assert "remaining_conflicts" in data
840
841 def test_resolve_all_theirs_json(self, repo: pathlib.Path) -> None:
842 self._setup_merge_conflict(repo)
843 result = _checkout(repo, "--theirs", "--all", "--json")
844 assert result.exit_code == 0
845 data = json.loads(result.output)
846 assert data["action"] == "conflict_resolved_all"
847 assert data["side"] == "theirs"
848
849 def test_resolve_single_file_json(self, repo: pathlib.Path) -> None:
850 self._setup_merge_conflict(repo)
851 result = _checkout(repo, "--ours", "shared.py", "--json")
852 assert result.exit_code == 0
853 data = json.loads(result.output)
854 assert data["action"] == "conflict_resolved"
855 assert data["file"] == "shared.py"
856 assert data["side"] == "ours"
857 assert "remaining_conflicts" in data
858
859 def test_resolve_all_empty_conflicts_exits_0(self, repo: pathlib.Path) -> None:
860 """--ours --all when no conflicts exist still exits 0."""
861 from muse.core.merge_engine import write_merge_state
862
863 ours = get_head_commit_id(repo, "main") or ""
864 write_merge_state(
865 repo,
866 base_commit="",
867 ours_commit=ours,
868 theirs_commit=ours,
869 conflict_paths=[],
870 other_branch="feat",
871 )
872 result = _checkout(repo, "--ours", "--all")
873 assert result.exit_code == 0
874
875 def test_resolve_nonexistent_path_exits_0(self, repo: pathlib.Path) -> None:
876 """A path not in the conflict list is informational, not an error."""
877 self._setup_merge_conflict(repo)
878 result = _checkout(repo, "--ours", "not_conflicted.py")
879 assert result.exit_code == 0
880
881 def test_missing_ours_theirs_without_all_exits_1(self, repo: pathlib.Path) -> None:
882 result = _checkout(repo, "--ours")
883 assert result.exit_code == 1
884
885
886 # ──────────────────────────────────────────────────────────────────────────────
887 # Stress
888 # ──────────────────────────────────────────────────────────────────────────────
889
890
891 @pytest.mark.slow
892 class TestStress:
893 def test_checkout_100_file_branch_fast(self, repo: pathlib.Path) -> None:
894 """Switching between branches with 100 modified files under 2s."""
895 for i in range(100):
896 (repo / f"f{i:03d}.py").write_text(f"x={i}\n")
897 _commit(repo, "-m", "big main")
898 _branch(repo, "big-alt")
899 _checkout(repo, "big-alt")
900 for i in range(100):
901 (repo / f"f{i:03d}.py").write_text(f"y={i}\n")
902 _commit(repo, "-m", "big alt")
903 _checkout(repo, "main")
904
905 t0 = time.perf_counter()
906 result = _checkout(repo, "big-alt")
907 elapsed = (time.perf_counter() - t0) * 1000
908 assert result.exit_code == 0
909 assert elapsed < 2000, f"checkout 100-file branch took {elapsed:.0f}ms (limit 2s)"
910
911 def test_dry_run_100_file_branch_fast(self, repo: pathlib.Path) -> None:
912 """dry-run on 100-file branch should be very fast (no restore)."""
913 for i in range(100):
914 (repo / f"g{i:03d}.py").write_text(f"x={i}\n")
915 _commit(repo, "-m", "big2")
916 _branch(repo, "big2-alt")
917
918 t0 = time.perf_counter()
919 result = _checkout(repo, "--dry-run", "big2-alt")
920 elapsed = (time.perf_counter() - t0) * 1000
921 assert result.exit_code == 0
922 assert elapsed < 500, f"dry-run took {elapsed:.0f}ms (limit 500ms)"
923
924 def test_concurrent_checkouts_separate_repos(self, tmp_path: pathlib.Path) -> None:
925 """Multiple threads checking out branches in separate repos must not interfere."""
926 errors: list[str] = []
927
928 def do_checkout(idx: int) -> None:
929 repo_dir = tmp_path / f"repo_{idx}"
930 repo_dir.mkdir()
931 subprocess.run(["muse", "init"], cwd=str(repo_dir), capture_output=True)
932 (repo_dir / "x.py").write_text(f"x={idx}\n")
933 subprocess.run(
934 ["muse", "commit", "-m", f"base{idx}"],
935 cwd=str(repo_dir), capture_output=True,
936 )
937 subprocess.run(
938 ["muse", "branch", "alt"], cwd=str(repo_dir), capture_output=True
939 )
940 subprocess.run(
941 ["muse", "checkout", "alt"], cwd=str(repo_dir), capture_output=True
942 )
943 (repo_dir / "y.py").write_text(f"y={idx}\n")
944 subprocess.run(
945 ["muse", "commit", "-m", f"alt{idx}"],
946 cwd=str(repo_dir), capture_output=True,
947 )
948 r = subprocess.run(
949 ["muse", "checkout", "main", "--json"],
950 cwd=str(repo_dir), capture_output=True, text=True,
951 )
952 if r.returncode != 0:
953 errors.append(f"repo_{idx}: checkout failed")
954 return
955 data = json.loads(r.stdout)
956 if data.get("action") != "switched":
957 errors.append(f"repo_{idx}: expected switched, got {data.get('action')}")
958
959 threads = [threading.Thread(target=do_checkout, args=(i,)) for i in range(6)]
960 for t in threads:
961 t.start()
962 for t in threads:
963 t.join()
964 assert not errors, "Concurrent checkout errors:\n" + "\n".join(errors)
965
966 def test_repeated_back_and_forth_100_times(self, two_branch_repo: pathlib.Path) -> None:
967 """Switching back and forth 100 times must not corrupt the working tree."""
968 for i in range(50):
969 r1 = _checkout(two_branch_repo, "feat")
970 assert r1.exit_code == 0, f"Iteration {i}: switch to feat failed"
971 assert (two_branch_repo / "feat.py").exists()
972 r2 = _checkout(two_branch_repo, "main")
973 assert r2.exit_code == 0, f"Iteration {i}: switch to main failed"
974
975
976 # ──────────────────────────────────────────────────────────────────────────────
977 # TestCheckoutMerge — muse checkout -m (Cohen Transform carry-forward)
978 # ──────────────────────────────────────────────────────────────────────────────
979
980
981 def _make_diverged_repo(tmp_path: pathlib.Path) -> pathlib.Path:
982 """Repo with main and *other* branches that have diverged file content.
983
984 Layout after setup::
985
986 main: shared.py = "line1\\nline2\\nline3\\n" (committed)
987 other: shared.py = "line1\\nLINE2\\nline3\\n" (committed — different line 2)
988
989 The caller is left on *main* with a dirty working tree.
990 """
991 saved = os.getcwd()
992 try:
993 os.chdir(tmp_path)
994 runner.invoke(None, ["init"])
995 finally:
996 os.chdir(saved)
997
998 (tmp_path / "shared.py").write_text("line1\nline2\nline3\n")
999 _commit(tmp_path, "-m", "initial")
1000
1001 # Create other branch with a different version of shared.py
1002 _branch(tmp_path, "other")
1003 _checkout(tmp_path, "other")
1004 (tmp_path / "shared.py").write_text("line1\nLINE2\nline3\n")
1005 _commit(tmp_path, "-m", "other changes line2")
1006
1007 # Back on main
1008 _checkout(tmp_path, "main")
1009 return tmp_path
1010
1011
1012 class TestCheckoutMergeParser:
1013 """Parser-level tests for the ``-m`` / ``--merge`` flag."""
1014
1015 def _parse(self, *args: str) -> "argparse.Namespace":
1016 import argparse
1017
1018 from muse.cli.commands.checkout import register
1019
1020 parser = argparse.ArgumentParser()
1021 sub = parser.add_subparsers()
1022 register(sub)
1023 return parser.parse_args(["checkout", *args])
1024
1025 def test_merge_short_flag_parsed(self) -> None:
1026 ns = self._parse("-m", "feat")
1027 assert ns.merge is True
1028
1029 def test_merge_long_flag_parsed(self) -> None:
1030 ns = self._parse("--merge", "feat")
1031 assert ns.merge is True
1032
1033 def test_merge_false_by_default(self) -> None:
1034 ns = self._parse("feat")
1035 assert ns.merge is False
1036
1037 def test_merge_and_dry_run_coexist(self) -> None:
1038 ns = self._parse("-m", "--dry-run", "feat")
1039 assert ns.merge is True
1040 assert ns.dry_run is True
1041
1042 def test_merge_and_json_coexist(self) -> None:
1043 ns = self._parse("-m", "--json", "feat")
1044 assert ns.merge is True
1045 assert ns.json_out is True
1046
1047
1048 class TestCheckoutMergeClean:
1049 """Clean-merge scenarios — no conflict markers should appear."""
1050
1051 def test_untracked_file_survives_checkout(self, repo: pathlib.Path) -> None:
1052 """Untracked files must not be disturbed by -m checkout."""
1053 _branch(repo, "feat")
1054 (repo / "untracked.txt").write_text("I am untracked\n")
1055 r = _checkout(repo, "-m", "feat")
1056 assert r.exit_code == 0
1057 assert (repo / "untracked.txt").read_text() == "I am untracked\n"
1058
1059 def test_ours_only_change_carried_cleanly(self, tmp_path: pathlib.Path) -> None:
1060 """When we modify a file that target branch left untouched, the change carries.
1061
1062 'clean' branch diverged by adding a brand-new file, leaving shared.py
1063 identical to main's HEAD. Our uncommitted change to shared.py has no
1064 competition from the target and must merge cleanly.
1065 """
1066 saved = os.getcwd()
1067 try:
1068 os.chdir(tmp_path)
1069 runner.invoke(None, ["init"])
1070 finally:
1071 os.chdir(saved)
1072 (tmp_path / "shared.py").write_text("line1\nline2\nline3\n")
1073 _commit(tmp_path, "-m", "initial")
1074
1075 # Create 'clean' branch — only adds a new file, does NOT touch shared.py
1076 _branch(tmp_path, "clean")
1077 _checkout(tmp_path, "clean")
1078 (tmp_path / "extra.py").write_text("# extra\n")
1079 _commit(tmp_path, "-m", "add extra.py")
1080 _checkout(tmp_path, "main")
1081
1082 # Dirty workdir on main: modify shared.py
1083 (tmp_path / "shared.py").write_text("line1\nline2\nLINE3\n")
1084 r = _checkout(tmp_path, "-m", "clean")
1085 assert r.exit_code == 0
1086 content = (tmp_path / "shared.py").read_text()
1087 assert "LINE3" in content, "Our uncommitted change must be in merged result"
1088
1089 def test_clean_merge_json_output(self, tmp_path: pathlib.Path) -> None:
1090 """JSON output reports clean_merges and empty conflicts on success."""
1091 saved = os.getcwd()
1092 try:
1093 os.chdir(tmp_path)
1094 runner.invoke(None, ["init"])
1095 finally:
1096 os.chdir(saved)
1097 (tmp_path / "shared.py").write_text("line1\nline2\nline3\n")
1098 _commit(tmp_path, "-m", "initial")
1099
1100 _branch(tmp_path, "clean")
1101 _checkout(tmp_path, "clean")
1102 (tmp_path / "extra.py").write_text("# extra\n")
1103 _commit(tmp_path, "-m", "add extra.py")
1104 _checkout(tmp_path, "main")
1105
1106 (tmp_path / "shared.py").write_text("line1\nline2\nLINE3\n")
1107 r = _checkout(tmp_path, "-m", "--json", "clean")
1108 assert r.exit_code == 0
1109 data = json.loads(r.output)
1110 assert data["action"] == "switched"
1111 assert data["branch"] == "clean"
1112 assert isinstance(data["clean_merges"], list)
1113 assert isinstance(data["conflicts"], list)
1114 assert len(data["conflicts"]) == 0
1115
1116 def test_switched_branch_recorded(self, tmp_path: pathlib.Path) -> None:
1117 """After a clean -m checkout the current branch is the target."""
1118 repo = _make_diverged_repo(tmp_path)
1119 (repo / "shared.py").write_text("line1\nline2\nLINE3\n")
1120 _checkout(repo, "-m", "other")
1121 assert read_current_branch(repo) == "other"
1122
1123 def test_no_dirty_files_succeeds_silently(self, repo: pathlib.Path) -> None:
1124 """If the working tree is clean, -m behaves like a normal checkout."""
1125 _branch(repo, "feat")
1126 r = _checkout(repo, "-m", "feat")
1127 assert r.exit_code == 0
1128 assert read_current_branch(repo) == "feat"
1129
1130 def test_dry_run_does_not_switch_branch(self, tmp_path: pathlib.Path) -> None:
1131 """-m --dry-run must not actually switch branches."""
1132 repo = _make_diverged_repo(tmp_path)
1133 (repo / "shared.py").write_text("line1\nline2\nLINE3\n")
1134 r = _checkout(repo, "-m", "--dry-run", "other")
1135 assert r.exit_code == 0
1136 assert read_current_branch(repo) == "main"
1137
1138 def test_dry_run_json_reports_dry_run_true(self, tmp_path: pathlib.Path) -> None:
1139 repo = _make_diverged_repo(tmp_path)
1140 (repo / "shared.py").write_text("line1\nline2\nLINE3\n")
1141 r = _checkout(repo, "-m", "--dry-run", "--json", "other")
1142 assert r.exit_code == 0
1143 data = json.loads(r.output)
1144 assert data["dry_run"] is True
1145 assert data["branch"] == "other"
1146
1147
1148 class TestCheckoutMergeConflict:
1149 """Conflict scenarios — conflict markers and MERGE_STATE must appear."""
1150
1151 def test_conflicting_change_exits_1(self, tmp_path: pathlib.Path) -> None:
1152 """Same line changed on both sides → conflict → exit code 1."""
1153 repo = _make_diverged_repo(tmp_path)
1154 # Also change line2 on main (uncommitted) — same line other branch changed
1155 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1156 r = _checkout(repo, "-m", "other")
1157 assert r.exit_code == 1
1158
1159 def test_conflict_markers_written_to_file(self, tmp_path: pathlib.Path) -> None:
1160 """Conflicting file must contain diff3-style conflict markers after -m."""
1161 repo = _make_diverged_repo(tmp_path)
1162 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1163 _checkout(repo, "-m", "other")
1164 content = (repo / "shared.py").read_text()
1165 assert "<<<<<<<" in content
1166 assert "=======" in content
1167 assert ">>>>>>>" in content
1168
1169 def test_conflict_markers_contain_cohen_action_labels(self, tmp_path: pathlib.Path) -> None:
1170 """Cohen-style action labels ([modified], [inserted], [deleted]) must appear."""
1171 repo = _make_diverged_repo(tmp_path)
1172 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1173 _checkout(repo, "-m", "other")
1174 content = (repo / "shared.py").read_text()
1175 # At least one action label must be present
1176 assert any(label in content for label in ("[modified]", "[inserted]", "[deleted]"))
1177
1178 def test_merge_state_written_on_conflict(self, tmp_path: pathlib.Path) -> None:
1179 """MERGE_STATE.json must exist after a conflicting -m checkout."""
1180 repo = _make_diverged_repo(tmp_path)
1181 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1182 _checkout(repo, "-m", "other")
1183 merge_state_file = repo / ".muse" / "MERGE_STATE.json"
1184 assert merge_state_file.exists()
1185
1186 def test_merge_state_lists_conflict_path(self, tmp_path: pathlib.Path) -> None:
1187 """MERGE_STATE.json must name the conflicting path."""
1188 repo = _make_diverged_repo(tmp_path)
1189 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1190 _checkout(repo, "-m", "other")
1191 data = json.loads((repo / ".muse" / "MERGE_STATE.json").read_text())
1192 assert "shared.py" in data.get("conflict_paths", [])
1193
1194 def test_conflict_json_output_contains_conflicts_list(self, tmp_path: pathlib.Path) -> None:
1195 """JSON output must list the conflict paths even on exit code 1."""
1196 repo = _make_diverged_repo(tmp_path)
1197 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1198 r = _checkout(repo, "-m", "--json", "other")
1199 data = json.loads(r.output)
1200 assert "conflicts" in data
1201 assert len(data["conflicts"]) >= 1
1202
1203 def test_branch_is_switched_despite_conflict(self, tmp_path: pathlib.Path) -> None:
1204 """Even when conflicts exist, we are on the target branch after -m."""
1205 repo = _make_diverged_repo(tmp_path)
1206 (repo / "shared.py").write_text("line1\nOURS_LINE2\nline3\n")
1207 _checkout(repo, "-m", "other")
1208 assert read_current_branch(repo) == "other"
1209
1210
1211 class TestCheckoutMergeErrors:
1212 """Error cases for -m flag."""
1213
1214 def test_merge_with_create_flag_ignored(self, repo: pathlib.Path) -> None:
1215 """``-m -b new-branch`` must NOT invoke _checkout_with_merge (target doesn't exist)."""
1216 # -m with -b falls through to regular create logic; no conflict-carry
1217 r = _checkout(repo, "-m", "-b", "new-branch")
1218 # Should succeed as a regular branch creation (no carry logic for new branches)
1219 assert r.exit_code == 0
1220
1221 def test_merge_missing_branch_errors(self, repo: pathlib.Path) -> None:
1222 """``-m nonexistent`` must exit 1 with an error message."""
1223 r = _checkout(repo, "-m", "nonexistent")
1224 assert r.exit_code == 1
1225
1226 def test_merge_already_on_branch_is_noop(self, repo: pathlib.Path) -> None:
1227 """``-m main`` when already on main must print 'Already on' and exit 0."""
1228 r = _checkout(repo, "-m", "main")
1229 assert r.exit_code == 0
1230 assert "Already on" in r.output or "already" in r.output.lower()
1231
1232 def test_merge_with_invalid_branch_name_errors(self, repo: pathlib.Path) -> None:
1233 """``-m ..invalid`` must exit 1 before attempting any merge."""
1234 r = _checkout(repo, "-m", "../bad/name")
1235 assert r.exit_code == 1
1236
1237
1238 # ──────────────────────────────────────────────────────────────────────────────
1239 # Non-conflicting carry-forward: checkout <branch> with dirty files that the
1240 # target branch does not touch should succeed without --merge or --autoshelf.
1241 #
1242 # This is the core ergonomics fix: an agent editing app.py on dev, then doing
1243 # checkout always refuses a dirty tracked file regardless of whether the target
1244 # branch shares the same committed version — see TestDirtyWorkdirBleedThrough.
1245 # Untracked files are never blocked (they are not in any snapshot).
1246 # ──────────────────────────────────────────────────────────────────────────────
1247
1248
1249 class TestSwitchWithNonConflictingChanges:
1250 @pytest.fixture()
1251 def two_branch_clean(self, tmp_path: pathlib.Path) -> pathlib.Path:
1252 """Repo with main and feat; feat adds a NEW file, leaves shared.py alone."""
1253 saved = os.getcwd()
1254 try:
1255 os.chdir(tmp_path)
1256 runner.invoke(None, ["init"])
1257 finally:
1258 os.chdir(saved)
1259 (tmp_path / "shared.py").write_text("x = 1\n")
1260 _commit(tmp_path, "-m", "initial")
1261 # Create feat branch with an extra file — shared.py is untouched
1262 _checkout(tmp_path, "-b", "feat")
1263 (tmp_path / "feat_only.py").write_text("f = 1\n")
1264 _commit(tmp_path, "-m", "feat commit")
1265 _checkout(tmp_path, "main")
1266 return tmp_path
1267
1268 def test_switch_still_blocks_on_true_conflict(
1269 self, tmp_path: pathlib.Path
1270 ) -> None:
1271 """When the target branch has a different version of a modified file, block."""
1272 saved = os.getcwd()
1273 try:
1274 os.chdir(tmp_path)
1275 runner.invoke(None, ["init"])
1276 finally:
1277 os.chdir(saved)
1278 (tmp_path / "shared.py").write_text("x = 1\n")
1279 _commit(tmp_path, "-m", "initial")
1280 # feat branch changes shared.py
1281 _checkout(tmp_path, "-b", "feat")
1282 (tmp_path / "shared.py").write_text("x = feat\n")
1283 _commit(tmp_path, "-m", "feat changes shared")
1284 _checkout(tmp_path, "main")
1285 # Now dirty shared.py locally — feat has a different version → must block
1286 (tmp_path / "shared.py").write_text("x = local\n")
1287 result = _checkout(tmp_path, "feat")
1288 assert result.exit_code != 0
1289
1290 def test_switch_new_file_carries_through(
1291 self, two_branch_clean: pathlib.Path
1292 ) -> None:
1293 """Brand-new untracked files always carry through (unchanged behaviour)."""
1294 repo = two_branch_clean
1295 (repo / "new_file.py").write_text("new = True\n")
1296 result = _checkout(repo, "feat")
1297 assert result.exit_code == 0
1298 assert (repo / "new_file.py").exists()
1299
1300
1301 # ──────────────────────────────────────────────────────────────────────────────
1302 # Agent supercharge — duration_ms and exit_code in every JSON output
1303 # ──────────────────────────────────────────────────────────────────────────────
1304
1305
1306 class TestElapsed:
1307 """Every JSON output path must include an ``duration_ms`` float."""
1308
1309 def test_switch_json_has_elapsed(self, two_branch_repo: pathlib.Path) -> None:
1310 result = _checkout(two_branch_repo, "feat", "--json")
1311 data = json.loads(result.output)
1312 assert "duration_ms" in data
1313 assert isinstance(data["duration_ms"], float)
1314
1315 def test_create_json_has_elapsed(self, repo: pathlib.Path) -> None:
1316 result = _checkout(repo, "-b", "elapsed-branch", "--json")
1317 data = json.loads(result.output)
1318 assert "duration_ms" in data
1319 assert isinstance(data["duration_ms"], float)
1320
1321 def test_already_on_json_has_elapsed(self, repo: pathlib.Path) -> None:
1322 result = _checkout(repo, "main", "--json")
1323 data = json.loads(result.output)
1324 assert "duration_ms" in data
1325 assert isinstance(data["duration_ms"], float)
1326
1327 def test_detach_json_has_elapsed(self, repo: pathlib.Path) -> None:
1328 sha = get_head_commit_id(repo, "main")
1329 assert sha is not None
1330 result = _checkout(repo, sha, "--json")
1331 data = json.loads(result.output)
1332 assert "duration_ms" in data
1333 assert isinstance(data["duration_ms"], float)
1334
1335 def test_dry_run_switch_json_has_elapsed(self, two_branch_repo: pathlib.Path) -> None:
1336 result = _checkout(two_branch_repo, "--dry-run", "feat", "--json")
1337 data = json.loads(result.output)
1338 assert "duration_ms" in data
1339
1340 def test_dry_run_create_json_has_elapsed(self, repo: pathlib.Path) -> None:
1341 result = _checkout(repo, "-b", "dry-elapsed", "--dry-run", "--json")
1342 data = json.loads(result.output)
1343 assert "duration_ms" in data
1344
1345 def test_dry_run_detach_json_has_elapsed(self, repo: pathlib.Path) -> None:
1346 sha = get_head_commit_id(repo, "main")
1347 assert sha is not None
1348 result = _checkout(repo, "--dry-run", sha, "--json")
1349 data = json.loads(result.output)
1350 assert "duration_ms" in data
1351
1352 def test_restored_json_has_elapsed(self, repo: pathlib.Path) -> None:
1353 result = _checkout(repo, "--force", "main", "--json")
1354 data = json.loads(result.output)
1355 assert "duration_ms" in data
1356
1357 def test_conflict_resolved_all_json_has_elapsed(self, repo: pathlib.Path) -> None:
1358 from muse.core.merge_engine import write_merge_state
1359
1360 ours = get_head_commit_id(repo, "main") or ""
1361 write_merge_state(
1362 repo,
1363 base_commit="",
1364 ours_commit=ours,
1365 theirs_commit=ours,
1366 conflict_paths=["a.py"],
1367 other_branch="feat",
1368 )
1369 result = _checkout(repo, "--ours", "--all", "--json")
1370 data = json.loads(result.output)
1371 assert "duration_ms" in data
1372
1373 def test_conflict_resolved_single_json_has_elapsed(self, repo: pathlib.Path) -> None:
1374 from muse.core.merge_engine import write_merge_state
1375 from muse.core.store import get_head_commit_id as _gci
1376
1377 ours = _gci(repo, "main") or ""
1378 write_merge_state(
1379 repo,
1380 base_commit="",
1381 ours_commit=ours,
1382 theirs_commit=ours,
1383 conflict_paths=["a.py"],
1384 other_branch="feat",
1385 )
1386 result = _checkout(repo, "--ours", "a.py", "--json")
1387 data = json.loads(result.output)
1388 assert "duration_ms" in data
1389
1390
1391 class TestExitCode:
1392 """Every successful JSON output path must include ``exit_code: 0``."""
1393
1394 def test_switch_json_exit_code_0(self, two_branch_repo: pathlib.Path) -> None:
1395 result = _checkout(two_branch_repo, "feat", "--json")
1396 data = json.loads(result.output)
1397 assert data["exit_code"] == 0
1398
1399 def test_create_json_exit_code_0(self, repo: pathlib.Path) -> None:
1400 result = _checkout(repo, "-b", "ec-branch", "--json")
1401 data = json.loads(result.output)
1402 assert data["exit_code"] == 0
1403
1404 def test_already_on_json_exit_code_0(self, repo: pathlib.Path) -> None:
1405 result = _checkout(repo, "main", "--json")
1406 data = json.loads(result.output)
1407 assert data["exit_code"] == 0
1408
1409 def test_detach_json_exit_code_0(self, repo: pathlib.Path) -> None:
1410 sha = get_head_commit_id(repo, "main")
1411 assert sha is not None
1412 result = _checkout(repo, sha, "--json")
1413 data = json.loads(result.output)
1414 assert data["exit_code"] == 0
1415
1416 def test_dry_run_switch_json_exit_code_0(self, two_branch_repo: pathlib.Path) -> None:
1417 result = _checkout(two_branch_repo, "--dry-run", "feat", "--json")
1418 data = json.loads(result.output)
1419 assert data["exit_code"] == 0
1420
1421 def test_dry_run_create_json_exit_code_0(self, repo: pathlib.Path) -> None:
1422 result = _checkout(repo, "-b", "dry-ec", "--dry-run", "--json")
1423 data = json.loads(result.output)
1424 assert data["exit_code"] == 0
1425
1426 def test_dry_run_detach_json_exit_code_0(self, repo: pathlib.Path) -> None:
1427 sha = get_head_commit_id(repo, "main")
1428 assert sha is not None
1429 result = _checkout(repo, "--dry-run", sha, "--json")
1430 data = json.loads(result.output)
1431 assert data["exit_code"] == 0
1432
1433 def test_restored_json_exit_code_0(self, repo: pathlib.Path) -> None:
1434 result = _checkout(repo, "--force", "main", "--json")
1435 data = json.loads(result.output)
1436 assert data["exit_code"] == 0
1437
1438 def test_conflict_resolved_all_json_exit_code_0(self, repo: pathlib.Path) -> None:
1439 from muse.core.merge_engine import write_merge_state
1440
1441 ours = get_head_commit_id(repo, "main") or ""
1442 write_merge_state(
1443 repo,
1444 base_commit="",
1445 ours_commit=ours,
1446 theirs_commit=ours,
1447 conflict_paths=["a.py"],
1448 other_branch="feat",
1449 )
1450 result = _checkout(repo, "--ours", "--all", "--json")
1451 data = json.loads(result.output)
1452 assert data["exit_code"] == 0
1453
1454 def test_conflict_resolved_single_json_exit_code_0(self, repo: pathlib.Path) -> None:
1455 from muse.core.merge_engine import write_merge_state
1456 from muse.core.store import get_head_commit_id as _gci
1457
1458 ours = _gci(repo, "main") or ""
1459 write_merge_state(
1460 repo,
1461 base_commit="",
1462 ours_commit=ours,
1463 theirs_commit=ours,
1464 conflict_paths=["a.py"],
1465 other_branch="feat",
1466 )
1467 result = _checkout(repo, "--ours", "a.py", "--json")
1468 data = json.loads(result.output)
1469 assert data["exit_code"] == 0
1470
1471
1472 class TestJsonSchemaComplete:
1473 """``duration_ms`` and ``exit_code`` must be in ``REQUIRED_KEYS``."""
1474
1475 REQUIRED_KEYS = {
1476 "action", "branch", "commit_id", "from_branch", "dry_run",
1477 "duration_ms", "exit_code",
1478 }
1479
1480 def test_switch_has_complete_schema(self, two_branch_repo: pathlib.Path) -> None:
1481 result = _checkout(two_branch_repo, "feat", "--json")
1482 data = json.loads(result.output)
1483 missing = self.REQUIRED_KEYS - set(data)
1484 assert not missing, f"Missing keys in 'switched' JSON: {missing}"
1485
1486 def test_create_has_complete_schema(self, repo: pathlib.Path) -> None:
1487 result = _checkout(repo, "-b", "schema-branch", "--json")
1488 data = json.loads(result.output)
1489 missing = self.REQUIRED_KEYS - set(data)
1490 assert not missing, f"Missing keys in 'created' JSON: {missing}"
1491
1492 def test_detach_has_complete_schema(self, repo: pathlib.Path) -> None:
1493 sha = get_head_commit_id(repo, "main")
1494 assert sha is not None
1495 result = _checkout(repo, sha, "--json")
1496 data = json.loads(result.output)
1497 missing = self.REQUIRED_KEYS - set(data)
1498 assert not missing, f"Missing keys in 'detached' JSON: {missing}"
1499
1500 def test_already_on_has_complete_schema(self, repo: pathlib.Path) -> None:
1501 result = _checkout(repo, "main", "--json")
1502 data = json.loads(result.output)
1503 missing = self.REQUIRED_KEYS - set(data)
1504 assert not missing, f"Missing keys in 'already_on' JSON: {missing}"
1505
1506
1507 # ──────────────────────────────────────────────────────────────────────────────
1508 # checkout -b --intent / --resumable
1509 # ──────────────────────────────────────────────────────────────────────────────
1510
1511
1512 class TestCheckoutCreateWithMeta:
1513 """``muse checkout -b <name> --intent <text> --resumable`` stores metadata."""
1514
1515 # ── Parser ────────────────────────────────────────────────────────────────
1516
1517 def _parse(self, *args: str) -> "argparse.Namespace":
1518 import argparse
1519 from muse.cli.commands.checkout import register
1520 p = argparse.ArgumentParser()
1521 sub = p.add_subparsers()
1522 register(sub)
1523 return p.parse_args(["checkout", *args])
1524
1525 def test_intent_flag_parsed(self) -> None:
1526 ns = self._parse("-b", "task/x", "--intent", "do the thing")
1527 assert ns.intent == "do the thing"
1528
1529 def test_resumable_flag_parsed(self) -> None:
1530 ns = self._parse("-b", "task/x", "--resumable")
1531 assert ns.resumable is True
1532
1533 def test_intent_default_none(self) -> None:
1534 ns = self._parse("main")
1535 assert ns.intent is None
1536
1537 def test_resumable_default_false(self) -> None:
1538 ns = self._parse("main")
1539 assert ns.resumable is False
1540
1541 def test_intent_without_create_is_error(self, repo: pathlib.Path) -> None:
1542 """--intent without -b should be rejected."""
1543 result = _checkout(repo, "main", "--intent", "oops")
1544 assert result.exit_code != 0
1545
1546 def test_resumable_without_create_is_error(self, repo: pathlib.Path) -> None:
1547 """--resumable without -b should be rejected."""
1548 result = _checkout(repo, "main", "--resumable")
1549 assert result.exit_code != 0
1550
1551 # ── Integration: intent stored ─────────────────────────────────────────
1552
1553 def test_intent_stored_in_branch_meta(self, repo: pathlib.Path) -> None:
1554 result = _checkout(repo, "-b", "task/work", "--intent", "implement auth")
1555 assert result.exit_code == 0
1556 meta = read_branch_meta(repo, "task/work")
1557 assert meta.get("intent") == "implement auth"
1558
1559 def test_resumable_stored_in_branch_meta(self, repo: pathlib.Path) -> None:
1560 result = _checkout(repo, "-b", "task/work", "--resumable")
1561 assert result.exit_code == 0
1562 meta = read_branch_meta(repo, "task/work")
1563 assert meta.get("resumable") is True
1564
1565 def test_intent_and_resumable_together(self, repo: pathlib.Path) -> None:
1566 result = _checkout(
1567 repo, "-b", "task/work", "--intent", "add feature", "--resumable"
1568 )
1569 assert result.exit_code == 0
1570 meta = read_branch_meta(repo, "task/work")
1571 assert meta.get("intent") == "add feature"
1572 assert meta.get("resumable") is True
1573
1574 def test_branch_switched_after_create_with_meta(self, repo: pathlib.Path) -> None:
1575 _checkout(repo, "-b", "task/work", "--intent", "x", "--resumable")
1576 assert read_current_branch(repo) == "task/work"
1577
1578 def test_no_metadata_when_flags_absent(self, repo: pathlib.Path) -> None:
1579 _checkout(repo, "-b", "task/plain")
1580 meta = read_branch_meta(repo, "task/plain")
1581 assert meta.get("intent") is None
1582 assert not meta.get("resumable")
1583
1584 def test_intent_only_no_resumable_set(self, repo: pathlib.Path) -> None:
1585 _checkout(repo, "-b", "task/work", "--intent", "just intent")
1586 meta = read_branch_meta(repo, "task/work")
1587 assert meta.get("intent") == "just intent"
1588 assert not meta.get("resumable")
1589
1590 def test_resumable_only_no_intent_set(self, repo: pathlib.Path) -> None:
1591 _checkout(repo, "-b", "task/work", "--resumable")
1592 meta = read_branch_meta(repo, "task/work")
1593 assert meta.get("resumable") is True
1594 assert meta.get("intent") is None
1595
1596 # ── JSON output still correct ──────────────────────────────────────────
1597
1598 def test_json_action_is_created(self, repo: pathlib.Path) -> None:
1599 result = _checkout(
1600 repo, "-b", "task/work", "--intent", "x", "--resumable", "--json"
1601 )
1602 assert result.exit_code == 0
1603 data = json.loads(result.output)
1604 assert data["action"] == "created"
1605
1606 def test_json_branch_name_correct(self, repo: pathlib.Path) -> None:
1607 result = _checkout(
1608 repo, "-b", "task/work", "--intent", "x", "--json"
1609 )
1610 data = json.loads(result.output)
1611 assert data["branch"] == "task/work"
1612
1613 # ── branch --json listing reflects metadata ────────────────────────────
1614
1615 def test_branch_list_json_shows_intent(self, repo: pathlib.Path) -> None:
1616 _checkout(repo, "-b", "task/work", "--intent", "my intent")
1617 result = _invoke(repo, ["branch", "--json"])
1618 branches = json.loads(result.output)
1619 entry = next(b for b in branches if b["name"] == "task/work")
1620 assert entry.get("intent") == "my intent"
1621
1622 def test_branch_list_json_shows_resumable(self, repo: pathlib.Path) -> None:
1623 _checkout(repo, "-b", "task/work", "--resumable")
1624 result = _invoke(repo, ["branch", "--json"])
1625 branches = json.loads(result.output)
1626 entry = next(b for b in branches if b["name"] == "task/work")
1627 assert entry.get("resumable") is True
1628
1629 def test_resumable_filter_finds_branch(self, repo: pathlib.Path) -> None:
1630 _checkout(repo, "-b", "task/work", "--resumable")
1631 _checkout(repo, "main")
1632 _checkout(repo, "-b", "task/plain")
1633 result = _invoke(repo, ["branch", "--resumable", "--json"])
1634 names = [b["name"] for b in json.loads(result.output)]
1635 assert "task/work" in names
1636 assert "task/plain" not in names
1637
1638 # ── Security: ANSI in intent ───────────────────────────────────────────
1639
1640 def test_ansi_in_intent_sanitized_in_text_output(self, repo: pathlib.Path) -> None:
1641 evil = "\x1b[31mevil\x1b[0m"
1642 result = _checkout(repo, "-b", "task/work", "--intent", evil)
1643 assert result.exit_code == 0
1644 assert "\x1b" not in result.output
1645
1646
1647 # ---------------------------------------------------------------------------
1648 # Flag registration tests
1649 # ---------------------------------------------------------------------------
1650
1651 import argparse as _argparse
1652 from muse.cli.commands.checkout import register as _register_checkout
1653
1654
1655 def _parse_checkout(*args: str) -> _argparse.Namespace:
1656 """Build an argument parser via register() and parse args."""
1657 root_p = _argparse.ArgumentParser()
1658 subs = root_p.add_subparsers(dest="cmd")
1659 _register_checkout(subs)
1660 return root_p.parse_args(["checkout", *args])
1661
1662
1663 class TestRegisterFlags:
1664 def test_default_json_out_is_false(self) -> None:
1665 ns = _parse_checkout("dev")
1666 assert ns.json_out is False
1667
1668 def test_json_flag_sets_json_out(self) -> None:
1669 ns = _parse_checkout("dev", "--json")
1670 assert ns.json_out is True
1671
1672 def test_j_shorthand_sets_json_out(self) -> None:
1673 ns = _parse_checkout("dev", "-j")
1674 assert ns.json_out is True
1675
1676 def test_create_branch_flag(self) -> None:
1677 ns = _parse_checkout("-b", "task/foo")
1678 assert ns.create is True
1679
1680 def test_force_flag(self) -> None:
1681 ns = _parse_checkout("dev", "--force")
1682 assert ns.force is True
1683
1684 def test_f_shorthand_for_force(self) -> None:
1685 ns = _parse_checkout("dev", "-f")
1686 assert ns.force is True
1687
1688 def test_dry_run_flag(self) -> None:
1689 ns = _parse_checkout("dev", "--dry-run")
1690 assert ns.dry_run is True
1691
1692 def test_n_shorthand_for_dry_run(self) -> None:
1693 ns = _parse_checkout("dev", "-n")
1694 assert ns.dry_run is True
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 141 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 144 days ago