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