gabriel / muse public
test_cmd_checkout.py python
827 lines 35.9 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 153 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.store import get_head_commit_id, read_current_branch
26
27 if TYPE_CHECKING:
28 import argparse
29
30 runner = CliRunner()
31
32 # ──────────────────────────────────────────────────────────────────────────────
33 # Helpers
34 # ──────────────────────────────────────────────────────────────────────────────
35
36
37 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
38 saved = os.getcwd()
39 try:
40 os.chdir(repo)
41 return runner.invoke(None, args)
42 finally:
43 os.chdir(saved)
44
45
46 def _checkout(repo: pathlib.Path, *extra: str) -> InvokeResult:
47 return _invoke(repo, ["checkout", *extra])
48
49
50 def _commit(repo: pathlib.Path, *extra: str) -> InvokeResult:
51 return _invoke(repo, ["commit", *extra])
52
53
54 def _branch(repo: pathlib.Path, *extra: str) -> InvokeResult:
55 return _invoke(repo, ["branch", *extra])
56
57
58 @pytest.fixture()
59 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
60 """Initialised repo with one commit on ``main``."""
61 saved = os.getcwd()
62 try:
63 os.chdir(tmp_path)
64 runner.invoke(None, ["init"])
65 finally:
66 os.chdir(saved)
67 (tmp_path / "a.py").write_text("x = 1\n")
68 _commit(tmp_path, "-m", "initial")
69 return tmp_path
70
71
72 @pytest.fixture()
73 def two_branch_repo(repo: pathlib.Path) -> pathlib.Path:
74 """Repo with ``main`` and ``feat`` branches, each with unique content."""
75 _branch(repo, "feat")
76 _checkout(repo, "feat")
77 (repo / "feat.py").write_text("f = 1\n")
78 _commit(repo, "-m", "feat commit")
79 _checkout(repo, "main")
80 return repo
81
82
83 # ──────────────────────────────────────────────────────────────────────────────
84 # Unit — parser flags
85 # ──────────────────────────────────────────────────────────────────────────────
86
87
88 class TestRegisterFlags:
89 def _parse(self, *args: str) -> "argparse.Namespace":
90 import argparse
91
92 from muse.cli.commands.checkout import register
93
94 p = argparse.ArgumentParser()
95 sub = p.add_subparsers()
96 register(sub)
97 return p.parse_args(["checkout", *args])
98
99 def test_default_fmt_is_text(self) -> None:
100 ns = self._parse("main")
101 assert ns.fmt == "text"
102
103 def test_json_flag_sets_fmt(self) -> None:
104 ns = self._parse("main", "--json")
105 assert ns.fmt == "json"
106
107 def test_format_json_flag(self) -> None:
108 ns = self._parse("main", "--format", "json")
109 assert ns.fmt == "json"
110
111 def test_create_flag(self) -> None:
112 ns = self._parse("-b", "new")
113 assert ns.create is True
114
115 def test_force_flag(self) -> None:
116 ns = self._parse("main", "--force")
117 assert ns.force is True
118
119 def test_force_short_flag(self) -> None:
120 ns = self._parse("main", "-f")
121 assert ns.force is True
122
123 def test_dry_run_flag(self) -> None:
124 ns = self._parse("main", "--dry-run")
125 assert ns.dry_run is True
126
127 def test_dry_run_short_flag(self) -> None:
128 ns = self._parse("main", "-n")
129 assert ns.dry_run is True
130
131 def test_dry_run_default_false(self) -> None:
132 ns = self._parse("main")
133 assert ns.dry_run is False
134
135 def test_ours_flag(self) -> None:
136 ns = self._parse("--ours", "file.py")
137 assert ns.resolve_ours is True
138
139 def test_theirs_flag(self) -> None:
140 ns = self._parse("--theirs", "file.py")
141 assert ns.resolve_theirs is True
142
143 def test_all_flag(self) -> None:
144 ns = self._parse("--ours", "--all")
145 assert ns.resolve_all is True
146
147 def test_target_optional(self) -> None:
148 ns = self._parse()
149 assert ns.target is None
150
151
152 # ──────────────────────────────────────────────────────────────────────────────
153 # Unit — dead-code removal
154 # ──────────────────────────────────────────────────────────────────────────────
155
156
157 class TestDeadCodeRemoved:
158 def test_read_current_branch_wrapper_removed(self) -> None:
159 import muse.cli.commands.checkout as m
160
161 assert not hasattr(m, "_read_current_branch"), (
162 "_read_current_branch was a dead one-liner wrapper and must be deleted"
163 )
164
165 def test_inline_sanitize_display_import_removed(self) -> None:
166 import inspect
167
168 import muse.cli.commands.checkout as m
169
170 src = inspect.getsource(m.run)
171 assert "sanitize_display as _sd" not in src, (
172 "The inline 'from muse.core.validation import sanitize_display as _sd' "
173 "inside run() was a redundant re-import of the module-level sanitize_display"
174 )
175
176
177 # ──────────────────────────────────────────────────────────────────────────────
178 # Integration — SWITCH (existing branch)
179 # ──────────────────────────────────────────────────────────────────────────────
180
181
182 class TestSwitch:
183 def test_switch_exits_0(self, two_branch_repo: pathlib.Path) -> None:
184 result = _checkout(two_branch_repo, "feat")
185 assert result.exit_code == 0
186
187 def test_switch_changes_branch(self, two_branch_repo: pathlib.Path) -> None:
188 _checkout(two_branch_repo, "feat")
189 assert read_current_branch(two_branch_repo) == "feat"
190
191 def test_switch_text_output(self, two_branch_repo: pathlib.Path) -> None:
192 result = _checkout(two_branch_repo, "feat")
193 assert "feat" in result.output
194 assert "Switched" in result.output
195
196 def test_switch_json_schema(self, two_branch_repo: pathlib.Path) -> None:
197 result = _checkout(two_branch_repo, "feat", "--json")
198 data = json.loads(result.output)
199 assert data["action"] == "switched"
200 assert data["branch"] == "feat"
201 assert data["from_branch"] == "main"
202 assert "commit_id" in data
203 assert data.get("dry_run") is False
204
205 def test_switch_restores_files(self, two_branch_repo: pathlib.Path) -> None:
206 """Files unique to ``feat`` appear after checkout and disappear on return."""
207 _checkout(two_branch_repo, "feat")
208 assert (two_branch_repo / "feat.py").exists()
209 _checkout(two_branch_repo, "main")
210 assert not (two_branch_repo / "feat.py").exists()
211
212 def test_switch_to_nonexistent_exits_1(self, repo: pathlib.Path) -> None:
213 result = _checkout(repo, "does-not-exist")
214 assert result.exit_code == 1
215
216 def test_switch_error_to_stderr(self, repo: pathlib.Path) -> None:
217 result = _checkout(repo, "ghost")
218 assert result.exit_code == 1
219 # Error must appear in stderr, not exclusively stdout
220 assert "not a branch" in (result.stderr or "").lower()
221
222
223 # ──────────────────────────────────────────────────────────────────────────────
224 # Integration — ALREADY_ON
225 # ──────────────────────────────────────────────────────────────────────────────
226
227
228 class TestAlreadyOn:
229 def test_already_on_exits_0(self, repo: pathlib.Path) -> None:
230 result = _checkout(repo, "main")
231 assert result.exit_code == 0
232
233 def test_already_on_text(self, repo: pathlib.Path) -> None:
234 result = _checkout(repo, "main")
235 assert "Already on" in result.output
236
237 def test_already_on_json_schema(self, repo: pathlib.Path) -> None:
238 result = _checkout(repo, "main", "--json")
239 data = json.loads(result.output)
240 assert data["action"] == "already_on"
241 assert data["branch"] == "main"
242 assert data["from_branch"] == "main"
243 assert "commit_id" in data
244
245
246 # ──────────────────────────────────────────────────────────────────────────────
247 # Integration — FORCE on current branch (working-tree restore)
248 # ──────────────────────────────────────────────────────────────────────────────
249
250
251 class TestForceOnCurrentBranch:
252 """checkout --force <current-branch> must restore the working tree to HEAD.
253
254 Regression: previously this was a no-op ('Already on main') regardless of
255 --force. Git's behaviour is to restore missing/modified tracked files even
256 when the branch is already current.
257 """
258
259 def test_force_current_branch_exits_0(self, repo: pathlib.Path) -> None:
260 result = _checkout(repo, "--force", "main")
261 assert result.exit_code == 0
262
263 def test_force_current_branch_restores_deleted_file(self, repo: pathlib.Path) -> None:
264 (repo / "a.py").unlink()
265 assert not (repo / "a.py").exists()
266 _checkout(repo, "--force", "main")
267 assert (repo / "a.py").exists(), "force checkout must restore deleted tracked file"
268
269 def test_force_current_branch_restores_modified_file(self, repo: pathlib.Path) -> None:
270 original = (repo / "a.py").read_text()
271 (repo / "a.py").write_text("corrupted content\n")
272 _checkout(repo, "--force", "main")
273 assert (repo / "a.py").read_text() == original, (
274 "force checkout must restore modified tracked file to HEAD content"
275 )
276
277 def test_force_current_branch_text_output(self, repo: pathlib.Path) -> None:
278 result = _checkout(repo, "--force", "main")
279 assert "restored" in result.output
280
281 def test_force_current_branch_json_action(self, repo: pathlib.Path) -> None:
282 result = _checkout(repo, "--force", "main", "--json")
283 data = json.loads(result.output)
284 assert data["action"] == "restored"
285 assert data["branch"] == "main"
286
287 def test_force_current_branch_dry_run_does_not_restore(
288 self, repo: pathlib.Path
289 ) -> None:
290 (repo / "a.py").unlink()
291 _checkout(repo, "--force", "--dry-run", "main")
292 assert not (repo / "a.py").exists(), (
293 "--dry-run must not actually restore files"
294 )
295
296 def test_force_current_branch_dry_run_json(self, repo: pathlib.Path) -> None:
297 result = _checkout(repo, "--force", "--dry-run", "main", "--json")
298 data = json.loads(result.output)
299 assert data["dry_run"] is True
300 assert data["action"] == "restored"
301
302 def test_without_force_still_noop_on_current_branch(
303 self, repo: pathlib.Path
304 ) -> None:
305 """Without --force, checkout on current branch is still a no-op."""
306 result = _checkout(repo, "main")
307 assert "Already on" in result.output
308
309
310 # ──────────────────────────────────────────────────────────────────────────────
311 # Integration — CREATE (-b)
312 # ──────────────────────────────────────────────────────────────────────────────
313
314
315 class TestCreate:
316 def test_create_exits_0(self, repo: pathlib.Path) -> None:
317 result = _checkout(repo, "-b", "new-branch")
318 assert result.exit_code == 0
319
320 def test_create_switches_to_new_branch(self, repo: pathlib.Path) -> None:
321 _checkout(repo, "-b", "new-branch")
322 assert read_current_branch(repo) == "new-branch"
323
324 def test_create_text_output(self, repo: pathlib.Path) -> None:
325 result = _checkout(repo, "-b", "my-branch")
326 assert "my-branch" in result.output
327
328 def test_create_json_schema(self, repo: pathlib.Path) -> None:
329 result = _checkout(repo, "-b", "json-branch", "--json")
330 data = json.loads(result.output)
331 assert data["action"] == "created"
332 assert data["branch"] == "json-branch"
333 assert data["from_branch"] == "main"
334 assert "commit_id" in data
335 assert data.get("dry_run") is False
336
337 def test_create_duplicate_exits_1(self, repo: pathlib.Path) -> None:
338 _checkout(repo, "-b", "dup")
339 _checkout(repo, "main")
340 result = _checkout(repo, "-b", "dup")
341 assert result.exit_code == 1
342
343 def test_create_duplicate_error_to_stderr(self, repo: pathlib.Path) -> None:
344 _checkout(repo, "-b", "dup2")
345 _checkout(repo, "main")
346 result = _checkout(repo, "-b", "dup2")
347 # Error must appear in stderr
348 assert "already exists" in (result.stderr or "").lower()
349
350 def test_create_invalid_name_exits_1(self, repo: pathlib.Path) -> None:
351 result = _checkout(repo, "-b", "bad..name")
352 assert result.exit_code == 1
353
354 def test_create_invalid_name_error_to_stderr(self, repo: pathlib.Path) -> None:
355 result = _checkout(repo, "-b", "bad..name")
356 assert "Invalid" in (result.stderr or "")
357
358
359 # ──────────────────────────────────────────────────────────────────────────────
360 # Integration — DETACH HEAD
361 # ──────────────────────────────────────────────────────────────────────────────
362
363
364 class TestDetach:
365 def test_detach_full_sha_exits_0(self, repo: pathlib.Path) -> None:
366 sha = get_head_commit_id(repo, "main")
367 assert sha is not None
368 result = _checkout(repo, sha)
369 assert result.exit_code == 0
370
371 def test_detach_full_sha_text_output(self, repo: pathlib.Path) -> None:
372 sha = get_head_commit_id(repo, "main")
373 assert sha is not None
374 result = _checkout(repo, sha)
375 assert sha[:8] in result.output
376
377 def test_detach_full_sha_json_schema(self, repo: pathlib.Path) -> None:
378 sha = get_head_commit_id(repo, "main")
379 assert sha is not None
380 result = _checkout(repo, sha, "--json")
381 data = json.loads(result.output)
382 assert data["action"] == "detached"
383 assert data["branch"] is None
384 assert data["commit_id"] == sha
385 assert data["from_branch"] == "main"
386 assert data.get("dry_run") is False
387
388 def test_detach_partial_sha_exits_0(self, repo: pathlib.Path) -> None:
389 sha = get_head_commit_id(repo, "main")
390 assert sha is not None
391 result = _checkout(repo, sha[:12])
392 assert result.exit_code == 0
393
394 def test_detach_partial_sha_points_to_correct_commit(self, repo: pathlib.Path) -> None:
395 """A partial SHA must resolve to the correct commit, not be treated as a branch."""
396 from muse.core.store import get_commits_for_branch, read_current_branch
397 from muse.core.repo import read_repo_id
398
399 (repo / "b.py").write_text("b=1\n")
400 _commit(repo, "-m", "second")
401
402 repo_id = read_repo_id(repo)
403 branch = read_current_branch(repo)
404 commits = get_commits_for_branch(repo, repo_id, branch)
405 first_sha = commits[-1].commit_id # oldest
406
407 result = _checkout(repo, first_sha[:12])
408 assert result.exit_code == 0
409 assert first_sha[:8] in result.output
410
411 def test_detach_bad_ref_exits_1(self, repo: pathlib.Path) -> None:
412 result = _checkout(repo, "deadbeefdeadbeef")
413 assert result.exit_code == 1
414
415 def test_detach_error_to_stderr(self, repo: pathlib.Path) -> None:
416 result = _checkout(repo, "deadbeefdeadbeef")
417 assert "not a branch" in (result.stderr or "").lower()
418
419
420 # ──────────────────────────────────────────────────────────────────────────────
421 # Integration — DRY-RUN
422 # ──────────────────────────────────────────────────────────────────────────────
423
424
425 class TestDryRun:
426 def test_dry_run_switch_exits_0(self, two_branch_repo: pathlib.Path) -> None:
427 result = _checkout(two_branch_repo, "--dry-run", "feat")
428 assert result.exit_code == 0
429
430 def test_dry_run_does_not_switch_branch(self, two_branch_repo: pathlib.Path) -> None:
431 _checkout(two_branch_repo, "--dry-run", "feat")
432 assert read_current_branch(two_branch_repo) == "main"
433
434 def test_dry_run_text_says_would(self, two_branch_repo: pathlib.Path) -> None:
435 result = _checkout(two_branch_repo, "--dry-run", "feat")
436 assert "Would" in result.output
437 assert "feat" in result.output
438
439 def test_dry_run_json_schema(self, two_branch_repo: pathlib.Path) -> None:
440 result = _checkout(two_branch_repo, "--dry-run", "feat", "--json")
441 data = json.loads(result.output)
442 assert data["dry_run"] is True
443 assert data["action"] == "switched"
444 assert data["branch"] == "feat"
445 assert data["from_branch"] == "main"
446
447 def test_dry_run_does_not_restore_files(self, two_branch_repo: pathlib.Path) -> None:
448 """feat.py exists only on feat branch; dry-run must not create it on main."""
449 _checkout(two_branch_repo, "--dry-run", "feat")
450 assert not (two_branch_repo / "feat.py").exists()
451
452 def test_dry_run_create_exits_0(self, repo: pathlib.Path) -> None:
453 result = _checkout(repo, "-b", "dry-branch", "--dry-run")
454 assert result.exit_code == 0
455
456 def test_dry_run_create_does_not_create_branch(self, repo: pathlib.Path) -> None:
457 _checkout(repo, "-b", "dry-branch", "--dry-run")
458 result = _invoke(repo, ["branch", "--json"])
459 names = [b["name"] for b in json.loads(result.output)]
460 assert "dry-branch" not in names
461
462 def test_dry_run_create_json_schema(self, repo: pathlib.Path) -> None:
463 result = _checkout(repo, "-b", "dry-new", "--dry-run", "--json")
464 data = json.loads(result.output)
465 assert data["dry_run"] is True
466 assert data["action"] == "created"
467 assert data["from_branch"] == "main"
468
469 def test_dry_run_detach_exits_0(self, repo: pathlib.Path) -> None:
470 sha = get_head_commit_id(repo, "main")
471 assert sha is not None
472 result = _checkout(repo, "--dry-run", sha)
473 assert result.exit_code == 0
474
475 def test_dry_run_detach_does_not_detach(self, repo: pathlib.Path) -> None:
476 sha = get_head_commit_id(repo, "main")
477 assert sha is not None
478 _checkout(repo, "--dry-run", sha)
479 assert read_current_branch(repo) == "main"
480
481 def test_dry_run_detach_json(self, repo: pathlib.Path) -> None:
482 sha = get_head_commit_id(repo, "main")
483 assert sha is not None
484 result = _checkout(repo, "--dry-run", sha, "--json")
485 data = json.loads(result.output)
486 assert data["dry_run"] is True
487 assert data["action"] == "detached"
488 assert data["branch"] is None
489
490 def test_dry_run_nonexistent_branch_exits_1(self, repo: pathlib.Path) -> None:
491 result = _checkout(repo, "--dry-run", "no-such-branch")
492 assert result.exit_code == 1
493
494 def test_dry_run_already_on_exits_0(self, repo: pathlib.Path) -> None:
495 result = _checkout(repo, "--dry-run", "main")
496 assert result.exit_code == 0
497
498 def test_dry_run_already_on_json(self, repo: pathlib.Path) -> None:
499 result = _checkout(repo, "--dry-run", "main", "--json")
500 data = json.loads(result.output)
501 assert data["dry_run"] is True
502 assert data["action"] == "already_on"
503
504
505 # ──────────────────────────────────────────────────────────────────────────────
506 # Integration — JSON schema consistency
507 # ──────────────────────────────────────────────────────────────────────────────
508
509
510 class TestJsonSchema:
511 REQUIRED_KEYS = {"action", "branch", "commit_id", "from_branch", "dry_run"}
512
513 def test_create_has_all_keys(self, repo: pathlib.Path) -> None:
514 result = _checkout(repo, "-b", "k-test", "--json")
515 data = json.loads(result.output)
516 missing = self.REQUIRED_KEYS - set(data)
517 assert not missing, f"Missing keys in 'created' JSON: {missing}"
518
519 def test_switch_has_all_keys(self, two_branch_repo: pathlib.Path) -> None:
520 result = _checkout(two_branch_repo, "feat", "--json")
521 data = json.loads(result.output)
522 missing = self.REQUIRED_KEYS - set(data)
523 assert not missing, f"Missing keys in 'switched' JSON: {missing}"
524
525 def test_already_on_has_all_keys(self, repo: pathlib.Path) -> None:
526 result = _checkout(repo, "main", "--json")
527 data = json.loads(result.output)
528 missing = self.REQUIRED_KEYS - set(data)
529 assert not missing, f"Missing keys in 'already_on' JSON: {missing}"
530
531 def test_detach_has_all_keys(self, repo: pathlib.Path) -> None:
532 sha = get_head_commit_id(repo, "main")
533 assert sha is not None
534 result = _checkout(repo, sha, "--json")
535 data = json.loads(result.output)
536 missing = self.REQUIRED_KEYS - set(data)
537 assert not missing, f"Missing keys in 'detached' JSON: {missing}"
538
539 def test_detach_branch_is_null(self, repo: pathlib.Path) -> None:
540 sha = get_head_commit_id(repo, "main")
541 assert sha is not None
542 result = _checkout(repo, sha, "--json")
543 data = json.loads(result.output)
544 assert data["branch"] is None
545
546 def test_from_branch_reflects_previous(self, two_branch_repo: pathlib.Path) -> None:
547 _checkout(two_branch_repo, "feat")
548 result = _checkout(two_branch_repo, "main", "--json")
549 data = json.loads(result.output)
550 assert data["from_branch"] == "feat"
551
552
553 # ──────────────────────────────────────────────────────────────────────────────
554 # Integration — validation
555 # ──────────────────────────────────────────────────────────────────────────────
556
557
558 class TestValidation:
559 def test_no_target_exits_1(self, repo: pathlib.Path) -> None:
560 result = _checkout(repo)
561 assert result.exit_code == 1
562
563 def test_no_target_error_to_stderr(self, repo: pathlib.Path) -> None:
564 result = _checkout(repo)
565 assert "Specify" in (result.stderr or "")
566
567 def test_unknown_format_exits_1(self, repo: pathlib.Path) -> None:
568 result = _checkout(repo, "main", "--format", "xml")
569 assert result.exit_code == 1
570
571 def test_unknown_format_error_to_stderr(self, repo: pathlib.Path) -> None:
572 result = _checkout(repo, "main", "--format", "xml")
573 assert "Unknown" in (result.stderr or "")
574
575 def test_ours_without_theirs_context_exits_1(self, repo: pathlib.Path) -> None:
576 result = _checkout(repo, "--ours", "file.py")
577 assert result.exit_code == 1
578
579 def test_ours_and_theirs_together_exits_1(self, repo: pathlib.Path) -> None:
580 result = _checkout(repo, "--ours", "--theirs", "--all")
581 assert result.exit_code == 1
582
583
584 # ──────────────────────────────────────────────────────────────────────────────
585 # Security — ANSI injection
586 # ──────────────────────────────────────────────────────────────────────────────
587
588
589 class TestSecurityAnsi:
590 def _has_ansi(self, s: str) -> bool:
591 return "\x1b[" in s
592
593 def test_ansi_in_target_sanitized(self, repo: pathlib.Path) -> None:
594 result = _checkout(repo, "\x1b[31mevil\x1b[0m")
595 assert not self._has_ansi(result.output)
596
597 def test_ansi_in_create_name_sanitized(self, repo: pathlib.Path) -> None:
598 result = _checkout(repo, "-b", "\x1b[31mevil\x1b[0m")
599 assert not self._has_ansi(result.output)
600
601 def test_ansi_in_format_sanitized(self, repo: pathlib.Path) -> None:
602 result = _checkout(repo, "main", "--format", "\x1b[31mxml\x1b[0m")
603 assert not self._has_ansi(result.output)
604
605 def test_error_not_a_branch_sanitized(self, repo: pathlib.Path) -> None:
606 """The 'not a branch' error message must not echo raw ANSI from target."""
607 result = _checkout(repo, "\x1b[31mnotabranch\x1b[0m")
608 assert not self._has_ansi(result.output)
609 assert not self._has_ansi(result.stderr or "")
610
611 def test_all_errors_to_stderr(self, repo: pathlib.Path) -> None:
612 """Every ❌ error must go to stderr; stderr must contain the error."""
613 error_cases = [
614 ["ghost"],
615 ["-b", "bad..name"],
616 ["--format", "xml"],
617 ]
618 for case in error_cases:
619 result = _checkout(repo, *case)
620 assert result.exit_code != 0, f"Expected failure for args {case}"
621 assert "❌" in (result.stderr or ""), (
622 f"Error not in stderr for args {case}: stderr={result.stderr!r}"
623 )
624
625
626 # ──────────────────────────────────────────────────────────────────────────────
627 # Integration — conflict resolution
628 # ──────────────────────────────────────────────────────────────────────────────
629
630
631 class TestConflictResolution:
632 def _setup_merge_conflict(
633 self, repo: pathlib.Path
634 ) -> tuple[str, str]:
635 """Create a merge conflict on ``repo``. Returns (ours_commit, theirs_commit)."""
636 # ours: commit on main
637 (repo / "shared.py").write_text("x = 1\n")
638 _commit(repo, "-m", "main: set x=1")
639 ours_cid = get_head_commit_id(repo, "main") or ""
640
641 # theirs: commit on feature branch
642 _branch(repo, "feat2")
643 _invoke(repo, ["checkout", "feat2"])
644 (repo / "shared.py").write_text("x = 2\n")
645 _commit(repo, "-m", "feat: set x=2")
646 theirs_cid = get_head_commit_id(repo, "feat2") or ""
647
648 _invoke(repo, ["checkout", "main"])
649 # Force a merge conflict via merge_engine plumbing
650 from muse.core.merge_engine import write_merge_state
651
652 write_merge_state(
653 repo,
654 base_commit="",
655 ours_commit=ours_cid,
656 theirs_commit=theirs_cid,
657 conflict_paths=["shared.py"],
658 other_branch="feat2",
659 )
660 return ours_cid, theirs_cid
661
662 def test_ours_no_merge_state_exits_1(self, repo: pathlib.Path) -> None:
663 result = _checkout(repo, "--ours", "file.py")
664 assert result.exit_code == 1
665
666 def test_theirs_no_merge_state_exits_1(self, repo: pathlib.Path) -> None:
667 result = _checkout(repo, "--theirs", "file.py")
668 assert result.exit_code == 1
669
670 def test_ours_and_theirs_both_exits_1(self, repo: pathlib.Path) -> None:
671 result = _checkout(repo, "--ours", "--theirs", "--all")
672 assert result.exit_code == 1
673
674 def test_ours_resolves_conflict(self, repo: pathlib.Path) -> None:
675 self._setup_merge_conflict(repo)
676 result = _checkout(repo, "--ours", "shared.py")
677 assert result.exit_code == 0
678
679 def test_theirs_resolves_conflict(self, repo: pathlib.Path) -> None:
680 self._setup_merge_conflict(repo)
681 result = _checkout(repo, "--theirs", "shared.py")
682 assert result.exit_code == 0
683
684 def test_resolve_all_ours_json(self, repo: pathlib.Path) -> None:
685 self._setup_merge_conflict(repo)
686 result = _checkout(repo, "--ours", "--all", "--json")
687 assert result.exit_code == 0
688 data = json.loads(result.output)
689 assert data["action"] == "conflict_resolved_all"
690 assert data["side"] == "ours"
691 assert "resolved_count" in data
692 assert "remaining_conflicts" in data
693
694 def test_resolve_all_theirs_json(self, repo: pathlib.Path) -> None:
695 self._setup_merge_conflict(repo)
696 result = _checkout(repo, "--theirs", "--all", "--json")
697 assert result.exit_code == 0
698 data = json.loads(result.output)
699 assert data["action"] == "conflict_resolved_all"
700 assert data["side"] == "theirs"
701
702 def test_resolve_single_file_json(self, repo: pathlib.Path) -> None:
703 self._setup_merge_conflict(repo)
704 result = _checkout(repo, "--ours", "shared.py", "--json")
705 assert result.exit_code == 0
706 data = json.loads(result.output)
707 assert data["action"] == "conflict_resolved"
708 assert data["file"] == "shared.py"
709 assert data["side"] == "ours"
710 assert "remaining_conflicts" in data
711
712 def test_resolve_all_empty_conflicts_exits_0(self, repo: pathlib.Path) -> None:
713 """--ours --all when no conflicts exist still exits 0."""
714 from muse.core.merge_engine import write_merge_state
715
716 ours = get_head_commit_id(repo, "main") or ""
717 write_merge_state(
718 repo,
719 base_commit="",
720 ours_commit=ours,
721 theirs_commit=ours,
722 conflict_paths=[],
723 other_branch="feat",
724 )
725 result = _checkout(repo, "--ours", "--all")
726 assert result.exit_code == 0
727
728 def test_resolve_nonexistent_path_exits_0(self, repo: pathlib.Path) -> None:
729 """A path not in the conflict list is informational, not an error."""
730 self._setup_merge_conflict(repo)
731 result = _checkout(repo, "--ours", "not_conflicted.py")
732 assert result.exit_code == 0
733
734 def test_missing_ours_theirs_without_all_exits_1(self, repo: pathlib.Path) -> None:
735 result = _checkout(repo, "--ours")
736 assert result.exit_code == 1
737
738
739 # ──────────────────────────────────────────────────────────────────────────────
740 # Stress
741 # ──────────────────────────────────────────────────────────────────────────────
742
743
744 @pytest.mark.slow
745 class TestStress:
746 def test_checkout_100_file_branch_fast(self, repo: pathlib.Path) -> None:
747 """Switching between branches with 100 modified files under 2s."""
748 for i in range(100):
749 (repo / f"f{i:03d}.py").write_text(f"x={i}\n")
750 _commit(repo, "-m", "big main")
751 _branch(repo, "big-alt")
752 _checkout(repo, "big-alt")
753 for i in range(100):
754 (repo / f"f{i:03d}.py").write_text(f"y={i}\n")
755 _commit(repo, "-m", "big alt")
756 _checkout(repo, "main")
757
758 t0 = time.perf_counter()
759 result = _checkout(repo, "big-alt")
760 elapsed = (time.perf_counter() - t0) * 1000
761 assert result.exit_code == 0
762 assert elapsed < 2000, f"checkout 100-file branch took {elapsed:.0f}ms (limit 2s)"
763
764 def test_dry_run_100_file_branch_fast(self, repo: pathlib.Path) -> None:
765 """dry-run on 100-file branch should be very fast (no restore)."""
766 for i in range(100):
767 (repo / f"g{i:03d}.py").write_text(f"x={i}\n")
768 _commit(repo, "-m", "big2")
769 _branch(repo, "big2-alt")
770
771 t0 = time.perf_counter()
772 result = _checkout(repo, "--dry-run", "big2-alt")
773 elapsed = (time.perf_counter() - t0) * 1000
774 assert result.exit_code == 0
775 assert elapsed < 500, f"dry-run took {elapsed:.0f}ms (limit 500ms)"
776
777 def test_concurrent_checkouts_separate_repos(self, tmp_path: pathlib.Path) -> None:
778 """Multiple threads checking out branches in separate repos must not interfere."""
779 errors: list[str] = []
780
781 def do_checkout(idx: int) -> None:
782 repo_dir = tmp_path / f"repo_{idx}"
783 repo_dir.mkdir()
784 subprocess.run(["muse", "init"], cwd=str(repo_dir), capture_output=True)
785 (repo_dir / "x.py").write_text(f"x={idx}\n")
786 subprocess.run(
787 ["muse", "commit", "-m", f"base{idx}"],
788 cwd=str(repo_dir), capture_output=True,
789 )
790 subprocess.run(
791 ["muse", "branch", "alt"], cwd=str(repo_dir), capture_output=True
792 )
793 subprocess.run(
794 ["muse", "checkout", "alt"], cwd=str(repo_dir), capture_output=True
795 )
796 (repo_dir / "y.py").write_text(f"y={idx}\n")
797 subprocess.run(
798 ["muse", "commit", "-m", f"alt{idx}"],
799 cwd=str(repo_dir), capture_output=True,
800 )
801 r = subprocess.run(
802 ["muse", "checkout", "main", "--json"],
803 cwd=str(repo_dir), capture_output=True, text=True,
804 )
805 if r.returncode != 0:
806 errors.append(f"repo_{idx}: checkout failed")
807 return
808 data = json.loads(r.stdout)
809 if data.get("action") != "switched":
810 errors.append(f"repo_{idx}: expected switched, got {data.get('action')}")
811
812 threads = [threading.Thread(target=do_checkout, args=(i,)) for i in range(6)]
813 for t in threads:
814 t.start()
815 for t in threads:
816 t.join()
817 assert not errors, "Concurrent checkout errors:\n" + "\n".join(errors)
818
819 def test_repeated_back_and_forth_100_times(self, two_branch_repo: pathlib.Path) -> None:
820 """Switching back and forth 100 times must not corrupt the working tree."""
821 for i in range(50):
822 r1 = _checkout(two_branch_repo, "feat")
823 assert r1.exit_code == 0, f"Iteration {i}: switch to feat failed"
824 assert (two_branch_repo / "feat.py").exists()
825 r2 = _checkout(two_branch_repo, "main")
826 assert r2.exit_code == 0, f"Iteration {i}: switch to main failed"
827 assert not (two_branch_repo / "feat.py").exists()
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 153 days ago