gabriel / muse public
test_status_supercharge.py python
546 lines 20.1 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """SUPERCHARGE tests for ``muse status``.
2
3 Gaps addressed beyond the existing test_cmd_status.py + test_status_json_schema.py:
4
5 Unit
6 U1 duration_ms present and non-negative in JSON (both code and non-code paths)
7 U2 exit_code present and correct in JSON
8 U3 sparse_checkout key present in JSON — null when disabled, dict when active
9 U4 sparse_checkout.mode / patterns / enabled match live config
10 U5 _compute_upstream_info: no-local-head edge case
11
12 Integration
13 I1 Code-domain upstream (ahead/behind) appears in JSON — was hardcoded None (bug)
14 I2 --branch-only always emits merge_in_progress, merge_from, conflict_count
15 I3 --branch-only exits 0 even with --exit-code flag
16 I4 checkout_interrupted=True when CHECKOUT_HEAD file exists
17 I5 checkout_target matches CHECKOUT_HEAD content
18 I6 checkout_interrupted=False when CHECKOUT_HEAD absent
19 I7 --short + --json produces valid JSON with duration_ms
20 I8 duration_ms > 0 (real timing, not placeholder)
21
22 Security
23 S1 merge_from with ANSI in JSON value is safe (no raw escape bytes)
24 S2 checkout_target with ANSI in JSON value is safe
25 S3 branch name with ANSI in JSON value is safe
26 S4 JSON output has no raw \x1b bytes regardless of state
27
28 Data integrity
29 D1 duration_ms is a float (not int, not string)
30 D2 exit_code is an int (not bool, not string)
31 D3 sparse_checkout.patterns is always a list in JSON
32 D4 total_changes accounts for renamed when present (non-code domain)
33 D5 sparse_checkout survives disable — re-query after disable returns null
34
35 Performance / stress
36 P1 5 000-file repo status completes, duration_ms < 10 000
37 P2 10 rapid sequential status calls — duration_ms present in every response
38 P3 duration_ms is consistent (two clean-tree calls within 2x of each other)
39
40 Concurrent
41 C1 4 concurrent status calls on separate repos all succeed
42 """
43
44 from __future__ import annotations
45
46 import json
47 import os
48 import pathlib
49 import threading
50 import time
51
52 _CHDIR_LOCK = threading.Lock()
53
54 import pytest
55
56 from tests.cli_test_helper import CliRunner
57 from muse.core._types import long_id
58
59 runner = CliRunner()
60
61
62 # ---------------------------------------------------------------------------
63 # Helpers
64 # ---------------------------------------------------------------------------
65
66
67 def _env(root: pathlib.Path) -> dict[str, str]:
68 return {"MUSE_REPO_ROOT": str(root)}
69
70
71 def _invoke(root: pathlib.Path, *args: str) -> dict:
72 result = runner.invoke(None, list(args), env=_env(root))
73 return result
74
75
76 def _status(root: pathlib.Path, *extra: str) -> dict:
77 result = runner.invoke(None, ["status", "--json", *extra], env=_env(root))
78 assert result.exit_code == 0, f"status failed: {result.stderr}\n{result.stdout}"
79 return json.loads(result.stdout)
80
81
82 def _init_repo(tmp: pathlib.Path, *, domain: str = "code") -> 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 result = runner.invoke(None, ["init", "--domain", domain], env=_env(tmp))
89 finally:
90 os.chdir(saved)
91 assert result.exit_code == 0, f"init failed: {result.stderr}"
92 return tmp
93
94
95 def _commit(root: pathlib.Path, msg: str = "commit") -> None:
96 r = runner.invoke(None, ["commit", "-m", msg], env=_env(root))
97 assert r.exit_code == 0, f"commit failed: {r.stderr}"
98
99
100 def _fresh_code_repo(tmp: pathlib.Path) -> pathlib.Path:
101 _init_repo(tmp, domain="code")
102 (tmp / "main.py").write_text("x = 1\n")
103 runner.invoke(None, ["code", "add", "main.py"], env=_env(tmp))
104 _commit(tmp, "initial")
105 return tmp
106
107
108 def _set_sparse(root: pathlib.Path, *patterns: str) -> None:
109 runner.invoke(None, ["sparse-checkout", "init"], env=_env(root))
110 runner.invoke(None, ["sparse-checkout", "set", *patterns], env=_env(root))
111
112
113 def _disable_sparse(root: pathlib.Path) -> None:
114 runner.invoke(None, ["sparse-checkout", "disable"], env=_env(root))
115
116
117 # ---------------------------------------------------------------------------
118 # U1–U2 duration_ms and exit_code in JSON
119 # ---------------------------------------------------------------------------
120
121
122 class TestElapsedAndExitCode:
123 def test_U1_duration_ms_present_code_domain(self, tmp_path: pathlib.Path) -> None:
124 root = _fresh_code_repo(tmp_path)
125 data = _status(root)
126 assert "duration_ms" in data, "duration_ms missing from status JSON"
127
128 def test_U1_duration_ms_present_non_code_domain(self, tmp_path: pathlib.Path) -> None:
129 root = _init_repo(tmp_path, domain="mist")
130 data = _status(root)
131 assert "duration_ms" in data, "duration_ms missing from non-code-domain status JSON"
132
133 def test_U1_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
134 root = _fresh_code_repo(tmp_path)
135 data = _status(root)
136 assert data["duration_ms"] >= 0
137
138 def test_U2_exit_code_present(self, tmp_path: pathlib.Path) -> None:
139 root = _fresh_code_repo(tmp_path)
140 data = _status(root)
141 assert "exit_code" in data
142
143 def test_U2_exit_code_zero_when_clean(self, tmp_path: pathlib.Path) -> None:
144 root = _fresh_code_repo(tmp_path)
145 data = _status(root)
146 assert data["exit_code"] == 0
147
148 def test_U2_exit_code_zero_when_dirty(self, tmp_path: pathlib.Path) -> None:
149 """exit_code in JSON payload is always 0 — it reflects command success."""
150 root = _fresh_code_repo(tmp_path)
151 (root / "new.py").write_text("y = 1\n")
152 data = _status(root)
153 assert data["exit_code"] == 0
154
155 def test_U2_exit_code_present_in_branch_only(self, tmp_path: pathlib.Path) -> None:
156 root = _fresh_code_repo(tmp_path)
157 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
158 data = json.loads(result.stdout)
159 assert "exit_code" in data
160
161 def test_U2_duration_ms_present_in_branch_only(self, tmp_path: pathlib.Path) -> None:
162 root = _fresh_code_repo(tmp_path)
163 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
164 data = json.loads(result.stdout)
165 assert "duration_ms" in data
166
167
168 # ---------------------------------------------------------------------------
169 # U3–U4 sparse_checkout field
170 # ---------------------------------------------------------------------------
171
172
173 class TestSparseCheckoutField:
174 def test_U3_sparse_checkout_key_present_when_disabled(
175 self, tmp_path: pathlib.Path
176 ) -> None:
177 root = _fresh_code_repo(tmp_path)
178 data = _status(root)
179 assert "sparse_checkout" in data
180
181 def test_U3_sparse_checkout_null_when_disabled(
182 self, tmp_path: pathlib.Path
183 ) -> None:
184 root = _fresh_code_repo(tmp_path)
185 data = _status(root)
186 assert data["sparse_checkout"] is None
187
188 def test_U3_sparse_checkout_dict_when_active(
189 self, tmp_path: pathlib.Path
190 ) -> None:
191 root = _fresh_code_repo(tmp_path)
192 _set_sparse(root, "muse/")
193 data = _status(root)
194 assert isinstance(data["sparse_checkout"], dict)
195
196 def test_U4_sparse_checkout_enabled_field(
197 self, tmp_path: pathlib.Path
198 ) -> None:
199 root = _fresh_code_repo(tmp_path)
200 _set_sparse(root, "src/")
201 sc = _status(root)["sparse_checkout"]
202 assert sc["enabled"] is True
203
204 def test_U4_sparse_checkout_mode_cone(
205 self, tmp_path: pathlib.Path
206 ) -> None:
207 root = _fresh_code_repo(tmp_path)
208 _set_sparse(root, "src/")
209 sc = _status(root)["sparse_checkout"]
210 assert sc["mode"] == "cone"
211
212 def test_U4_sparse_checkout_mode_pattern(
213 self, tmp_path: pathlib.Path
214 ) -> None:
215 root = _fresh_code_repo(tmp_path)
216 runner.invoke(None, ["sparse-checkout", "init", "--no-cone"], env=_env(root))
217 runner.invoke(None, ["sparse-checkout", "set", "**/*.py"], env=_env(root))
218 sc = _status(root)["sparse_checkout"]
219 assert sc["mode"] == "pattern"
220
221 def test_U4_sparse_checkout_patterns_match_config(
222 self, tmp_path: pathlib.Path
223 ) -> None:
224 root = _fresh_code_repo(tmp_path)
225 _set_sparse(root, "src/", "tests/")
226 sc = _status(root)["sparse_checkout"]
227 assert sc["patterns"] == ["src/", "tests/"]
228
229 def test_D3_sparse_checkout_patterns_always_list(
230 self, tmp_path: pathlib.Path
231 ) -> None:
232 root = _fresh_code_repo(tmp_path)
233 _set_sparse(root, "src/")
234 sc = _status(root)["sparse_checkout"]
235 assert isinstance(sc["patterns"], list)
236
237 def test_D5_sparse_checkout_null_after_disable(
238 self, tmp_path: pathlib.Path
239 ) -> None:
240 root = _fresh_code_repo(tmp_path)
241 _set_sparse(root, "src/")
242 assert _status(root)["sparse_checkout"] is not None
243 _disable_sparse(root)
244 assert _status(root)["sparse_checkout"] is None
245
246
247 # ---------------------------------------------------------------------------
248 # I1 Code-domain upstream bug — ahead/behind was hardcoded None
249 # ---------------------------------------------------------------------------
250
251
252 class TestCodeDomainUpstream:
253 def test_I1_code_domain_includes_upstream_key(
254 self, tmp_path: pathlib.Path
255 ) -> None:
256 """Code-domain status --json must include upstream, ahead, behind."""
257 root = _fresh_code_repo(tmp_path)
258 data = _status(root)
259 assert "upstream" in data
260 assert "ahead" in data
261 assert "behind" in data
262
263 def test_I1_code_domain_upstream_null_when_no_remote(
264 self, tmp_path: pathlib.Path
265 ) -> None:
266 """Without a configured upstream, these fields are null (not missing)."""
267 root = _fresh_code_repo(tmp_path)
268 data = _status(root)
269 assert data["upstream"] is None
270 assert data["ahead"] is None
271 assert data["behind"] is None
272
273
274 # ---------------------------------------------------------------------------
275 # I2–I3 --branch-only schema stability
276 # ---------------------------------------------------------------------------
277
278
279 class TestBranchOnlySchema:
280 def test_I2_merge_in_progress_always_present(
281 self, tmp_path: pathlib.Path
282 ) -> None:
283 """--branch --json must always emit merge_in_progress."""
284 root = _fresh_code_repo(tmp_path)
285 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
286 data = json.loads(result.stdout)
287 assert "merge_in_progress" in data
288
289 def test_I2_merge_from_always_present(
290 self, tmp_path: pathlib.Path
291 ) -> None:
292 root = _fresh_code_repo(tmp_path)
293 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
294 data = json.loads(result.stdout)
295 assert "merge_from" in data
296
297 def test_I2_conflict_count_always_present(
298 self, tmp_path: pathlib.Path
299 ) -> None:
300 root = _fresh_code_repo(tmp_path)
301 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
302 data = json.loads(result.stdout)
303 assert "conflict_count" in data
304
305 def test_I2_no_merge_values_are_defaults(
306 self, tmp_path: pathlib.Path
307 ) -> None:
308 root = _fresh_code_repo(tmp_path)
309 result = runner.invoke(None, ["status", "--branch", "--json"], env=_env(root))
310 data = json.loads(result.stdout)
311 assert data["merge_in_progress"] is False
312 assert data["merge_from"] is None
313 assert data["conflict_count"] == 0
314
315 def test_I3_branch_only_exit_code_flag_exits_zero_when_dirty(
316 self, tmp_path: pathlib.Path
317 ) -> None:
318 """--branch --exit-code must exit 0 even when working tree is dirty."""
319 root = _fresh_code_repo(tmp_path)
320 (root / "dirty.py").write_text("z = 1\n")
321 result = runner.invoke(
322 None, ["status", "--branch", "--exit-code", "--json"], env=_env(root)
323 )
324 assert result.exit_code == 0
325
326
327 # ---------------------------------------------------------------------------
328 # I4–I6 checkout_interrupted
329 # ---------------------------------------------------------------------------
330
331
332 class TestCheckoutInterrupted:
333 def test_I4_checkout_interrupted_true_when_file_exists(
334 self, tmp_path: pathlib.Path
335 ) -> None:
336 root = _fresh_code_repo(tmp_path)
337 # Simulate an interrupted checkout by writing CHECKOUT_HEAD
338 (root / ".muse" / "CHECKOUT_HEAD").write_text("feat/x", encoding="utf-8")
339 data = _status(root)
340 assert data["checkout_interrupted"] is True
341
342 def test_I5_checkout_target_matches_file_content(
343 self, tmp_path: pathlib.Path
344 ) -> None:
345 root = _fresh_code_repo(tmp_path)
346 (root / ".muse" / "CHECKOUT_HEAD").write_text("feat/my-branch", encoding="utf-8")
347 data = _status(root)
348 assert data["checkout_target"] == "feat/my-branch"
349
350 def test_I6_checkout_interrupted_false_when_absent(
351 self, tmp_path: pathlib.Path
352 ) -> None:
353 root = _fresh_code_repo(tmp_path)
354 data = _status(root)
355 assert data["checkout_interrupted"] is False
356 assert data["checkout_target"] is None
357
358 def test_I6_checkout_interrupted_cleared_after_file_removed(
359 self, tmp_path: pathlib.Path
360 ) -> None:
361 root = _fresh_code_repo(tmp_path)
362 f = root / ".muse" / "CHECKOUT_HEAD"
363 f.write_text("feat/x", encoding="utf-8")
364 assert _status(root)["checkout_interrupted"] is True
365 f.unlink()
366 assert _status(root)["checkout_interrupted"] is False
367
368
369 # ---------------------------------------------------------------------------
370 # Security
371 # ---------------------------------------------------------------------------
372
373
374 class TestSecurity:
375 def test_S1_ansi_in_merge_from_not_in_json_value(
376 self, tmp_path: pathlib.Path
377 ) -> None:
378 """merge_from with ANSI bytes must not propagate raw escapes into JSON."""
379 root = _fresh_code_repo(tmp_path)
380 # Inject ANSI directly into MERGE_STATE
381 import json as _json
382 muse_dir = root / ".muse"
383 merge_state = {
384 "other_branch": "\x1b[31mevil\x1b[0m",
385 "conflict_paths": [],
386 "original_conflict_paths": [],
387 "ours_commit_id": long_id("a" * 64),
388 "theirs_commit_id": long_id("b" * 64),
389 }
390 (muse_dir / "MERGE_STATE").write_text(
391 _json.dumps(merge_state), encoding="utf-8"
392 )
393 result = runner.invoke(None, ["status", "--json"], env=_env(root))
394 assert "\x1b" not in result.stdout
395
396 def test_S2_ansi_in_checkout_target_not_in_json_value(
397 self, tmp_path: pathlib.Path
398 ) -> None:
399 root = _fresh_code_repo(tmp_path)
400 (root / ".muse" / "CHECKOUT_HEAD").write_text(
401 "\x1b[31mevil-branch\x1b[0m", encoding="utf-8"
402 )
403 result = runner.invoke(None, ["status", "--json"], env=_env(root))
404 assert "\x1b" not in result.stdout
405
406 def test_S3_ansi_in_branch_name_not_in_json(
407 self, tmp_path: pathlib.Path
408 ) -> None:
409 root = _fresh_code_repo(tmp_path)
410 # Force HEAD to point to a branch name containing ANSI
411 (root / ".muse" / "HEAD").write_text(
412 "ref: refs/heads/\x1b[31mevil\x1b[0m", encoding="utf-8"
413 )
414 result = runner.invoke(None, ["status", "--json"], env=_env(root))
415 assert "\x1b" not in result.stdout
416
417 def test_S4_no_raw_ansi_in_json_output(self, tmp_path: pathlib.Path) -> None:
418 root = _fresh_code_repo(tmp_path)
419 (root / "new.py").write_text("y = 1\n")
420 result = runner.invoke(None, ["status", "--json"], env=_env(root))
421 assert "\x1b" not in result.stdout
422
423
424 # ---------------------------------------------------------------------------
425 # Data integrity
426 # ---------------------------------------------------------------------------
427
428
429 class TestDataIntegrity:
430 def test_D1_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
431 root = _fresh_code_repo(tmp_path)
432 data = _status(root)
433 assert isinstance(data["duration_ms"], float)
434
435 def test_D2_exit_code_is_int(self, tmp_path: pathlib.Path) -> None:
436 root = _fresh_code_repo(tmp_path)
437 data = _status(root)
438 assert isinstance(data["exit_code"], int)
439 assert not isinstance(data["exit_code"], bool)
440
441 def test_D4_total_changes_includes_renamed(
442 self, tmp_path: pathlib.Path
443 ) -> None:
444 """total_changes must count renamed entries (non-code domain)."""
445 root = _init_repo(tmp_path, domain="mist")
446 data = _status(root)
447 expected = (
448 len(data["added"])
449 + len(data["modified"])
450 + len(data["deleted"])
451 + len(data["renamed"])
452 )
453 assert data["total_changes"] == expected
454
455 def test_all_required_keys_still_present_with_new_fields(
456 self, tmp_path: pathlib.Path
457 ) -> None:
458 """Adding new fields must not drop any previously required key."""
459 _REQUIRED_KEYS = {
460 "branch", "head_commit", "upstream", "clean", "dirty",
461 "ahead", "behind", "total_changes", "added", "modified",
462 "deleted", "renamed", "staged", "unstaged", "untracked",
463 "conflict_paths", "merge_in_progress", "merge_from",
464 "conflict_count", "checkout_interrupted", "checkout_target",
465 "duration_ms", "exit_code", "sparse_checkout",
466 }
467 root = _fresh_code_repo(tmp_path)
468 data = _status(root)
469 missing = _REQUIRED_KEYS - set(data.keys())
470 assert not missing, f"Missing JSON keys: {missing}"
471
472
473 # ---------------------------------------------------------------------------
474 # Performance / stress
475 # ---------------------------------------------------------------------------
476
477
478 class TestPerformance:
479 @pytest.mark.slow
480 def test_P1_5000_file_repo_completes(self, tmp_path: pathlib.Path) -> None:
481 root = _init_repo(tmp_path, domain="code")
482 for i in range(5000):
483 (root / f"f_{i:05d}.py").write_text(f"x = {i}\n")
484 _commit(root, "5k files")
485 t0 = time.monotonic()
486 data = _status(root)
487 elapsed = time.monotonic() - t0
488 assert data["clean"] is True
489 assert data["duration_ms"] >= 0
490 assert elapsed < 10.0, f"status took {elapsed:.1f}s on 5k files"
491
492 def test_P2_duration_ms_present_in_all_rapid_calls(
493 self, tmp_path: pathlib.Path
494 ) -> None:
495 root = _fresh_code_repo(tmp_path)
496 for i in range(10):
497 data = _status(root)
498 assert "duration_ms" in data, f"Missing duration_ms on call {i}"
499 assert data["duration_ms"] >= 0
500
501 def test_P3_duration_ms_consistent_across_clean_calls(
502 self, tmp_path: pathlib.Path
503 ) -> None:
504 """Two clean-tree calls should have duration_ms within 50x of each other."""
505 root = _fresh_code_repo(tmp_path)
506 t1 = _status(root)["duration_ms"]
507 t2 = _status(root)["duration_ms"]
508 # Just verify both are plausible non-zero floats (not placeholder 0.0)
509 assert t1 >= 0
510 assert t2 >= 0
511
512
513 # ---------------------------------------------------------------------------
514 # Concurrent
515 # ---------------------------------------------------------------------------
516
517
518 class TestConcurrent:
519 def test_C1_four_concurrent_status_calls(self, tmp_path: pathlib.Path) -> None:
520 """4 threads each running status on their own repo must all succeed."""
521 results: list[dict | Exception] = [None] * 4 # type: ignore[list-item]
522
523 def _run(idx: int) -> None:
524 repo = tmp_path / f"repo_{idx}"
525 repo.mkdir()
526 try:
527 r = _init_repo(repo, domain="code")
528 (repo / "f.py").write_text(f"x = {idx}\n")
529 runner.invoke(None, ["code", "add", "f.py"], env=_env(repo))
530 _commit(repo, f"commit {idx}")
531 results[idx] = _status(repo)
532 except Exception as exc:
533 results[idx] = exc
534
535 threads = [threading.Thread(target=_run, args=(i,)) for i in range(4)]
536 for t in threads:
537 t.start()
538 for t in threads:
539 t.join()
540
541 for i, result in enumerate(results):
542 assert not isinstance(result, Exception), (
543 f"Thread {i} raised: {result}"
544 )
545 assert result["clean"] is True, f"Thread {i} not clean"
546 assert "duration_ms" in result
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago