gabriel / muse public
test_switch_supercharge.py python
510 lines 19.1 KB
Raw
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
1 """SUPERCHARGE tests for ``muse switch``.
2
3 Gaps addressed beyond the existing test_cmd_switch.py:
4
5 Unit
6 U1 duration_ms present and float in all JSON success paths
7 U2 exit_code present and 0 in all JSON success paths
8 U3 JSON error emitted to stdout when --json + branch not found
9 U4 JSON error emitted to stdout when --json + no PREV_BRANCH
10 U5 JSON error emitted to stdout when --json + mutual-exclusion violation
11 U6 commit_id is sha256:-prefixed in JSON output
12 U7 from_branch always present and correct in JSON
13
14 Integration
15 I1 switch - + --json emits valid JSON with correct action
16 I2 switch -C existing branch + --json → action == "reset"
17 I3 switch -C new branch + --json → action == "created"
18 I4 switch -c + --intent stores intent on new branch
19 I5 switch -c + --resumable marks branch resumable
20 I6 --autoshelf + --json emits valid JSON
21 I7 --merge + --json emits valid JSON (smoke — merge may yield conflict)
22 I8 dry-run -C + --json emits JSON without creating branch
23 I9 --json + --dry-run switch - emits JSON (prev branch preview)
24 I10 switch -c on existing branch + --json → JSON error (not text traceback)
25
26 Security
27 S1 null byte in branch name rejected (exit non-zero)
28 S2 path traversal (../) in branch name rejected
29 S3 JSON error output contains no raw ANSI bytes
30 S4 branch name sanitized in JSON output values
31
32 Data integrity
33 D1 PREV_BRANCH correct after 10 alternating switches
34 D2 commit_id in JSON matches actual HEAD after switch
35 D3 duration_ms is float not int
36 D4 exit_code is int not bool
37 D5 force-create action field: "reset" when branch existed, "created" when new
38
39 Stress
40 P1 50 rapid switches among 3 branches — final state consistent
41 P2 switch with 30 branches in the repo completes
42 P3 duration_ms present in all 10 rapid JSON calls
43
44 Concurrent
45 C1 4 threads switching in separate repos — all succeed
46 """
47
48 from __future__ import annotations
49 from collections.abc import Mapping
50
51 import json
52 import os
53 import pathlib
54 import threading
55
56 import pytest
57
58 from tests.cli_test_helper import CliRunner, InvokeResult
59 from muse.core.store import get_head_commit_id, read_current_branch
60 from muse.core.paths import muse_dir
61
62 runner = CliRunner()
63
64 # ---------------------------------------------------------------------------
65 # Helpers
66 # ---------------------------------------------------------------------------
67
68
69 _CHDIR_LOCK = threading.Lock()
70
71
72 def _env(repo: pathlib.Path) -> Mapping[str, str]:
73 return {"MUSE_REPO_ROOT": str(repo)}
74
75
76 def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult:
77 return runner.invoke(None, ["switch", *args], env=_env(repo))
78
79
80 def _run(repo: pathlib.Path, *args: str) -> InvokeResult:
81 return runner.invoke(None, list(args), env=_env(repo))
82
83
84 def _init_repo(tmp: pathlib.Path) -> pathlib.Path:
85 tmp.mkdir(parents=True, exist_ok=True)
86 with _CHDIR_LOCK:
87 saved = os.getcwd()
88 try:
89 os.chdir(tmp)
90 runner.invoke(None, ["init"])
91 finally:
92 os.chdir(saved)
93 (tmp / "a.py").write_text("x = 1\n")
94 _run(tmp, "commit", "-m", "initial")
95 return tmp
96
97
98 def _add_branch(repo: pathlib.Path, name: str) -> None:
99 _run(repo, "branch", name)
100
101
102 def _switch(repo: pathlib.Path, *args: str) -> InvokeResult:
103 return _invoke(repo, *args)
104
105
106 def _json_switch(repo: pathlib.Path, *args: str) -> Mapping[str, object]:
107 result = _invoke(repo, "--json", *args)
108 return result, json.loads(result.stdout) if result.stdout.strip() else {}
109
110
111 @pytest.fixture()
112 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
113 return _init_repo(tmp_path)
114
115
116 @pytest.fixture()
117 def two_branch_repo(repo: pathlib.Path) -> pathlib.Path:
118 _add_branch(repo, "feat")
119 _run(repo, "checkout", "feat")
120 (repo / "feat.py").write_text("f = 1\n")
121 _run(repo, "commit", "-m", "feat commit")
122 _run(repo, "checkout", "main")
123 return repo
124
125
126 # ---------------------------------------------------------------------------
127 # U1–U2 duration_ms and exit_code in all JSON success paths
128 # ---------------------------------------------------------------------------
129
130
131 class TestElapsedAndExitCode:
132 def test_U1_duration_ms_switched(self, two_branch_repo: pathlib.Path) -> None:
133 result, data = _json_switch(two_branch_repo, "feat")
134 assert result.exit_code == 0
135 assert "duration_ms" in data, f"duration_ms missing; got keys: {list(data)}"
136
137 def test_U1_duration_ms_created(self, repo: pathlib.Path) -> None:
138 result, data = _json_switch(repo, "-c", "new-feat")
139 assert result.exit_code == 0
140 assert "duration_ms" in data
141
142 def test_U1_duration_ms_force_create_new(self, repo: pathlib.Path) -> None:
143 result, data = _json_switch(repo, "-C", "brand-new")
144 assert result.exit_code == 0
145 assert "duration_ms" in data
146
147 def test_U1_duration_ms_force_create_existing(
148 self, two_branch_repo: pathlib.Path
149 ) -> None:
150 result, data = _json_switch(two_branch_repo, "-C", "feat")
151 assert result.exit_code == 0
152 assert "duration_ms" in data
153
154 def test_U1_duration_ms_dry_run(self, two_branch_repo: pathlib.Path) -> None:
155 result, data = _json_switch(two_branch_repo, "--dry-run", "feat")
156 assert result.exit_code == 0
157 assert "duration_ms" in data
158
159 def test_U2_exit_code_in_json(self, two_branch_repo: pathlib.Path) -> None:
160 result, data = _json_switch(two_branch_repo, "feat")
161 assert result.exit_code == 0
162 assert "exit_code" in data
163 assert data["exit_code"] == 0
164
165 def test_D3_duration_ms_is_float(self, two_branch_repo: pathlib.Path) -> None:
166 _, data = _json_switch(two_branch_repo, "feat")
167 assert isinstance(data["duration_ms"], float)
168
169 def test_D4_exit_code_is_int_not_bool(self, two_branch_repo: pathlib.Path) -> None:
170 _, data = _json_switch(two_branch_repo, "feat")
171 assert isinstance(data["exit_code"], int)
172 assert not isinstance(data["exit_code"], bool)
173
174
175 # ---------------------------------------------------------------------------
176 # U3–U5 JSON error output on failure
177 # ---------------------------------------------------------------------------
178
179
180 class TestJsonErrors:
181 def test_U3_branch_not_found_json_error(self, repo: pathlib.Path) -> None:
182 result = _invoke(repo, "--json", "no-such-branch")
183 assert result.exit_code != 0
184 data = json.loads(result.stdout)
185 assert "error" in data
186 assert data["exit_code"] != 0
187
188 def test_U4_no_prev_branch_json_error(self, repo: pathlib.Path) -> None:
189 result = _invoke(repo, "--json", "-")
190 assert result.exit_code != 0
191 data = json.loads(result.stdout)
192 assert "error" in data
193
194 def test_U5_mutual_exclusion_c_and_C_json_error(self, repo: pathlib.Path) -> None:
195 result = _invoke(repo, "--json", "-c", "-C", "foo")
196 assert result.exit_code != 0
197 data = json.loads(result.stdout)
198 assert "error" in data
199
200 def test_U5_mutual_exclusion_discard_merge_json_error(
201 self, repo: pathlib.Path
202 ) -> None:
203 result = _invoke(repo, "--json", "--discard-changes", "--merge", "feat")
204 assert result.exit_code != 0
205 data = json.loads(result.stdout)
206 assert "error" in data
207
208 def test_U5_mutual_exclusion_discard_autoshelf_json_error(
209 self, repo: pathlib.Path
210 ) -> None:
211 result = _invoke(repo, "--json", "--discard-changes", "--autoshelf", "feat")
212 assert result.exit_code != 0
213 data = json.loads(result.stdout)
214 assert "error" in data
215
216 def test_U5_mutual_exclusion_merge_autoshelf_json_error(
217 self, repo: pathlib.Path
218 ) -> None:
219 result = _invoke(repo, "--json", "--merge", "--autoshelf", "feat")
220 assert result.exit_code != 0
221 data = json.loads(result.stdout)
222 assert "error" in data
223
224 def test_json_error_has_duration_ms(self, repo: pathlib.Path) -> None:
225 result = _invoke(repo, "--json", "no-such-branch")
226 data = json.loads(result.stdout)
227 assert "duration_ms" in data
228
229 def test_json_error_has_exit_code(self, repo: pathlib.Path) -> None:
230 result = _invoke(repo, "--json", "no-such-branch")
231 data = json.loads(result.stdout)
232 assert "exit_code" in data
233 assert data["exit_code"] != 0
234
235 def test_create_existing_branch_json_error(
236 self, two_branch_repo: pathlib.Path
237 ) -> None:
238 """switch -c on an existing branch must emit a JSON error, not a traceback."""
239 result = _invoke(two_branch_repo, "--json", "-c", "feat")
240 assert result.exit_code != 0
241 data = json.loads(result.stdout)
242 assert "error" in data
243 assert "Traceback" not in result.stdout
244
245
246 # ---------------------------------------------------------------------------
247 # U6–U7 commit_id and from_branch
248 # ---------------------------------------------------------------------------
249
250
251 class TestCommitIdAndFromBranch:
252 def test_U6_commit_id_sha256_prefixed(self, two_branch_repo: pathlib.Path) -> None:
253 _, data = _json_switch(two_branch_repo, "feat")
254 assert data["commit_id"].startswith("sha256:")
255
256 def test_U7_from_branch_correct(self, two_branch_repo: pathlib.Path) -> None:
257 _, data = _json_switch(two_branch_repo, "feat")
258 assert data["from_branch"] == "main"
259
260 def test_U7_from_branch_after_create(self, repo: pathlib.Path) -> None:
261 _, data = _json_switch(repo, "-c", "new-feat")
262 assert data["from_branch"] == "main"
263
264 def test_D2_commit_id_matches_head(self, two_branch_repo: pathlib.Path) -> None:
265 result, data = _json_switch(two_branch_repo, "feat")
266 assert result.exit_code == 0
267 actual = get_head_commit_id(two_branch_repo, "feat")
268 assert data["commit_id"] == actual
269
270
271 # ---------------------------------------------------------------------------
272 # I1–I10 Integration — uncovered paths
273 # ---------------------------------------------------------------------------
274
275
276 class TestIntegration:
277 def test_I1_switch_dash_json(self, two_branch_repo: pathlib.Path) -> None:
278 """switch - + --json must emit valid JSON."""
279 _switch(two_branch_repo, "feat") # go to feat first
280 result = _invoke(two_branch_repo, "--json", "-")
281 assert result.exit_code == 0
282 data = json.loads(result.stdout)
283 assert "action" in data
284 assert data["branch"] == "main"
285 assert "duration_ms" in data
286
287 def test_I2_force_create_existing_action_reset(
288 self, two_branch_repo: pathlib.Path
289 ) -> None:
290 result, data = _json_switch(two_branch_repo, "-C", "feat")
291 assert result.exit_code == 0
292 assert data["action"] == "reset"
293
294 def test_I3_force_create_new_action_created(self, repo: pathlib.Path) -> None:
295 result, data = _json_switch(repo, "-C", "brand-new")
296 assert result.exit_code == 0
297 assert data["action"] == "created"
298
299 def test_I4_create_with_intent(self, repo: pathlib.Path) -> None:
300 """switch -c + --intent stores intent metadata on the new branch."""
301 result = _invoke(repo, "-c", "task/foo", "--intent", "implement foo feature")
302 assert result.exit_code == 0
303 # Verify branch was created
304 assert read_current_branch(repo) == "task/foo"
305 # Verify intent is queryable
306 branch_data = json.loads(
307 _run(repo, "branch", "--json").stdout
308 )
309 foo = next((b for b in branch_data if b["name"] == "task/foo"), None)
310 assert foo is not None
311 assert foo.get("intent") == "implement foo feature"
312
313 def test_I5_create_with_resumable(self, repo: pathlib.Path) -> None:
314 """switch -c + --resumable marks the branch as a resumable checkpoint."""
315 result = _invoke(repo, "-c", "task/bar", "--resumable")
316 assert result.exit_code == 0
317 branch_data = json.loads(_run(repo, "branch", "--json").stdout)
318 bar = next((b for b in branch_data if b["name"] == "task/bar"), None)
319 assert bar is not None
320 assert bar.get("resumable") is True
321
322 def test_I6_autoshelf_json(self, two_branch_repo: pathlib.Path) -> None:
323 result, data = _json_switch(two_branch_repo, "--autoshelf", "feat")
324 assert result.exit_code == 0
325 assert "action" in data
326 assert "duration_ms" in data
327
328 def test_I8_dry_run_force_create_json(self, two_branch_repo: pathlib.Path) -> None:
329 result, data = _json_switch(two_branch_repo, "--dry-run", "-C", "feat")
330 assert result.exit_code == 0
331 assert data["dry_run"] is True
332 assert "action" in data
333 assert "duration_ms" in data
334 # Branch ref must not be modified
335 orig_tip = get_head_commit_id(two_branch_repo, "feat")
336 main_tip = get_head_commit_id(two_branch_repo, "main")
337 assert get_head_commit_id(two_branch_repo, "feat") == orig_tip
338
339 def test_I9_dry_run_switch_dash_json(self, two_branch_repo: pathlib.Path) -> None:
340 _switch(two_branch_repo, "feat") # record prev
341 result, data = _json_switch(two_branch_repo, "--dry-run", "-")
342 assert result.exit_code == 0
343 assert data["dry_run"] is True
344 assert "duration_ms" in data
345
346 def test_I10_create_existing_json_error_not_traceback(
347 self, two_branch_repo: pathlib.Path
348 ) -> None:
349 result = _invoke(two_branch_repo, "--json", "-c", "feat")
350 assert result.exit_code != 0
351 assert "Traceback" not in result.stdout
352 data = json.loads(result.stdout)
353 assert "error" in data
354
355
356 # ---------------------------------------------------------------------------
357 # Security
358 # ---------------------------------------------------------------------------
359
360
361 class TestSecurity:
362 def test_S1_null_byte_in_branch_name_rejected(self, repo: pathlib.Path) -> None:
363 result = _invoke(repo, "feat\x00malicious")
364 assert result.exit_code != 0
365
366 def test_S2_path_traversal_in_branch_name_rejected(
367 self, repo: pathlib.Path
368 ) -> None:
369 result = _invoke(repo, "../traversal")
370 assert result.exit_code != 0
371
372 def test_S3_json_error_no_ansi(self, repo: pathlib.Path) -> None:
373 result = _invoke(repo, "--json", "no-such-branch")
374 assert "\x1b" not in result.stdout
375
376 def test_S4_branch_name_sanitized_in_json(
377 self, two_branch_repo: pathlib.Path
378 ) -> None:
379 _, data = _json_switch(two_branch_repo, "feat")
380 assert "\x1b" not in json.dumps(data)
381
382 def test_ansi_in_branch_name_rejected_with_json(
383 self, repo: pathlib.Path
384 ) -> None:
385 result = _invoke(repo, "--json", "\x1b[31mbad\x1b[0m")
386 assert result.exit_code != 0
387 # Must be parseable JSON, not raw error text
388 data = json.loads(result.stdout)
389 assert "error" in data
390
391
392 # ---------------------------------------------------------------------------
393 # Data integrity
394 # ---------------------------------------------------------------------------
395
396
397 class TestDataIntegrity:
398 def test_D1_prev_branch_correct_after_10_toggles(
399 self, two_branch_repo: pathlib.Path
400 ) -> None:
401 branches = ["main", "feat"]
402 for i in range(10):
403 _switch(two_branch_repo, branches[i % 2])
404 # After 10 switches index 9 → branches[1] = feat
405 assert read_current_branch(two_branch_repo) == "feat"
406 # PREV_BRANCH should be main (the branch we came from)
407 prev = (muse_dir(two_branch_repo) / "PREV_BRANCH").read_text().strip()
408 assert prev == "main"
409
410 def test_D5_force_create_action_reset_when_existed(
411 self, two_branch_repo: pathlib.Path
412 ) -> None:
413 _, data = _json_switch(two_branch_repo, "-C", "feat")
414 assert data["action"] == "reset"
415
416 def test_D5_force_create_action_created_when_new(
417 self, repo: pathlib.Path
418 ) -> None:
419 _, data = _json_switch(repo, "-C", "new-branch")
420 assert data["action"] == "created"
421
422 def test_all_success_json_keys_present(
423 self, two_branch_repo: pathlib.Path
424 ) -> None:
425 """Every successful switch --json must have the full schema."""
426 _, data = _json_switch(two_branch_repo, "feat")
427 required = {"action", "branch", "from_branch", "commit_id", "dry_run",
428 "duration_ms", "exit_code"}
429 missing = required - set(data.keys())
430 assert not missing, f"Missing JSON keys: {missing}"
431
432 def test_all_error_json_keys_present(self, repo: pathlib.Path) -> None:
433 result = _invoke(repo, "--json", "ghost")
434 data = json.loads(result.stdout)
435 required = {"error", "duration_ms", "exit_code"}
436 missing = required - set(data.keys())
437 assert not missing, f"Missing error JSON keys: {missing}"
438
439
440 # ---------------------------------------------------------------------------
441 # Stress
442 # ---------------------------------------------------------------------------
443
444
445 class TestStress:
446 def test_P1_50_rapid_switches_three_branches(
447 self, tmp_path: pathlib.Path
448 ) -> None:
449 repo = _init_repo(tmp_path)
450 for name in ("feat-a", "feat-b"):
451 _add_branch(repo, name)
452 branches = ["main", "feat-a", "feat-b"]
453 for i in range(50):
454 result = _switch(repo, branches[i % 3])
455 assert result.exit_code == 0, f"Switch {i} failed"
456 expected = branches[49 % 3]
457 assert read_current_branch(repo) == expected
458
459 def test_P2_switch_with_many_branches(self, tmp_path: pathlib.Path) -> None:
460 """30 branches in the repo; switching still completes."""
461 repo = _init_repo(tmp_path)
462 for i in range(30):
463 _run(repo, "branch", f"branch-{i:02d}")
464 result = _switch(repo, "branch-00")
465 assert result.exit_code == 0
466 assert read_current_branch(repo) == "branch-00"
467
468 def test_P3_duration_ms_in_all_rapid_json_calls(
469 self, two_branch_repo: pathlib.Path
470 ) -> None:
471 branches = ["main", "feat"]
472 for i in range(10):
473 result = _invoke(two_branch_repo, "--json", branches[i % 2])
474 data = json.loads(result.stdout)
475 assert "duration_ms" in data, f"Missing duration_ms on call {i}"
476 assert isinstance(data["duration_ms"], float)
477
478
479 # ---------------------------------------------------------------------------
480 # Concurrent
481 # ---------------------------------------------------------------------------
482
483
484 _INIT_LOCK = threading.Lock()
485
486
487 class TestConcurrent:
488 def test_C1_four_concurrent_switches(self, tmp_path: pathlib.Path) -> None:
489 results = [None] * 4
490
491 def _work(idx: int) -> None:
492 repo = tmp_path / f"repo_{idx}"
493 with _INIT_LOCK:
494 r = _init_repo(repo)
495 _add_branch(r, "feat")
496 try:
497 res = _switch(r, "feat")
498 results[idx] = res.exit_code
499 except Exception as exc:
500 results[idx] = exc
501
502 threads = [threading.Thread(target=_work, args=(i,)) for i in range(4)]
503 for t in threads:
504 t.start()
505 for t in threads:
506 t.join()
507
508 for i, result in enumerate(results):
509 assert not isinstance(result, Exception), f"Thread {i}: {result}"
510 assert result == 0, f"Thread {i} exit code: {result}"
File History 2 commits
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago