gabriel / muse public
test_guard_supercharge.py python
526 lines 22.0 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 136 days ago
1 """Seven-tier tests for ``muse/cli/guard.py`` — ``require_clean_workdir``.
2
3 Tiers
4 -----
5 Unit — force bypass, clean workdir, added-only is safe, dirty exits.
6 Integration — text vs JSON format, target_manifest filtering, truncation at 10.
7 End-to-end — guard fires through real CLI commands (reset --hard, checkout).
8 Stress — 500 dirty files, repeated calls on same repo.
9 Data integrity — target_manifest OID comparison logic, deleted-in-target blocks.
10 Security — ANSI injection in operation name, null byte in path, path traversal.
11 Performance — completes under 2 s on a clean repo.
12 """
13
14 from __future__ import annotations
15
16 import json
17 import os
18 import pathlib
19 import threading
20 import time
21
22 import pytest
23
24 from muse.core._types import fake_id
25 from tests.cli_test_helper import CliRunner, InvokeResult
26
27 runner = CliRunner()
28
29
30 # ──────────────────────────────────────────────────────────────────────────────
31 # Helpers
32 # ──────────────────────────────────────────────────────────────────────────────
33
34
35 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
36 saved = os.getcwd()
37 try:
38 os.chdir(repo)
39 return runner.invoke(None, args)
40 finally:
41 os.chdir(saved)
42
43
44 def _init_repo(root: pathlib.Path) -> None:
45 saved = os.getcwd()
46 try:
47 os.chdir(root)
48 runner.invoke(None, ["init"])
49 finally:
50 os.chdir(saved)
51
52
53 def _commit(repo: pathlib.Path, message: str = "commit") -> None:
54 saved = os.getcwd()
55 try:
56 os.chdir(repo)
57 runner.invoke(None, ["code", "add", "."])
58 runner.invoke(None, ["commit", "-m", message])
59 finally:
60 os.chdir(saved)
61
62
63 @pytest.fixture()
64 def clean_repo(tmp_path: pathlib.Path) -> pathlib.Path:
65 """Repo with one committed file, clean working tree."""
66 _init_repo(tmp_path)
67 (tmp_path / "a.py").write_text("x = 1\n")
68 _commit(tmp_path, "initial")
69 return tmp_path
70
71
72 @pytest.fixture()
73 def dirty_repo(clean_repo: pathlib.Path) -> pathlib.Path:
74 """Repo with a committed file that has been locally modified."""
75 (clean_repo / "a.py").write_text("x = 99\n")
76 return clean_repo
77
78
79 # ──────────────────────────────────────────────────────────────────────────────
80 # Unit — require_clean_workdir directly
81 # ──────────────────────────────────────────────────────────────────────────────
82
83
84 class TestUnit:
85 def test_force_true_is_noop(self, dirty_repo: pathlib.Path) -> None:
86 from muse.cli.guard import require_clean_workdir
87
88 # Must not raise even though working tree is dirty.
89 require_clean_workdir(dirty_repo, "test-op", force=True)
90
91 def test_clean_workdir_does_not_raise(self, clean_repo: pathlib.Path) -> None:
92 from muse.cli.guard import require_clean_workdir
93
94 require_clean_workdir(clean_repo, "test-op")
95
96 def test_modified_tracked_file_raises(self, dirty_repo: pathlib.Path) -> None:
97 from muse.cli.guard import require_clean_workdir
98 from muse.core.errors import ExitCode
99
100 with pytest.raises(SystemExit) as exc:
101 require_clean_workdir(dirty_repo, "test-op")
102 assert exc.value.code == ExitCode.USER_ERROR
103
104 def test_deleted_tracked_file_raises(self, clean_repo: pathlib.Path) -> None:
105 from muse.cli.guard import require_clean_workdir
106 from muse.core.errors import ExitCode
107
108 (clean_repo / "a.py").unlink()
109 with pytest.raises(SystemExit) as exc:
110 require_clean_workdir(clean_repo, "test-op")
111 assert exc.value.code == ExitCode.USER_ERROR
112
113 def test_added_untracked_file_does_not_raise(self, clean_repo: pathlib.Path) -> None:
114 from muse.cli.guard import require_clean_workdir
115
116 # Brand-new file never in a snapshot — apply_manifest won't touch it.
117 (clean_repo / "brand_new.py").write_text("y = 2\n")
118 require_clean_workdir(clean_repo, "test-op")
119
120 def test_empty_repo_no_commits_does_not_raise(self, tmp_path: pathlib.Path) -> None:
121 from muse.cli.guard import require_clean_workdir
122
123 _init_repo(tmp_path)
124 # No commits → no head manifest → guard passes.
125 require_clean_workdir(tmp_path, "test-op")
126
127 def test_force_false_default_is_checked(self, dirty_repo: pathlib.Path) -> None:
128 from muse.cli.guard import require_clean_workdir
129
130 with pytest.raises(SystemExit):
131 require_clean_workdir(dirty_repo, "test-op", force=False)
132
133
134 # ──────────────────────────────────────────────────────────────────────────────
135 # Integration — format, target_manifest, truncation
136 # ──────────────────────────────────────────────────────────────────────────────
137
138
139 class TestIntegration:
140 def test_text_fmt_error_message_mentions_operation(
141 self, dirty_repo: pathlib.Path, capsys
142 ) -> None:
143 from muse.cli.guard import require_clean_workdir
144
145 with pytest.raises(SystemExit):
146 require_clean_workdir(dirty_repo, "my-operation", json_out=False)
147 captured = capsys.readouterr()
148 assert "my-operation" in captured.err
149
150 def test_text_fmt_error_goes_to_stderr(self, dirty_repo: pathlib.Path, capsys) -> None:
151 from muse.cli.guard import require_clean_workdir
152
153 with pytest.raises(SystemExit):
154 require_clean_workdir(dirty_repo, "test-op", json_out=False)
155 captured = capsys.readouterr()
156 assert captured.out == ""
157 assert captured.err != ""
158
159 def test_json_fmt_error_goes_to_stdout(self, dirty_repo: pathlib.Path, capsys) -> None:
160 from muse.cli.guard import require_clean_workdir
161
162 with pytest.raises(SystemExit):
163 require_clean_workdir(dirty_repo, "test-op", json_out=True)
164 captured = capsys.readouterr()
165 data = json.loads(captured.out)
166 assert data["error"] == "dirty_workdir"
167
168 def test_json_fmt_includes_files_list(self, dirty_repo: pathlib.Path, capsys) -> None:
169 from muse.cli.guard import require_clean_workdir
170
171 with pytest.raises(SystemExit):
172 require_clean_workdir(dirty_repo, "test-op", json_out=True)
173 data = json.loads(capsys.readouterr().out)
174 assert "files" in data
175 assert isinstance(data["files"], list)
176 assert len(data["files"]) > 0
177
178 def test_json_fmt_includes_operation(self, dirty_repo: pathlib.Path, capsys) -> None:
179 from muse.cli.guard import require_clean_workdir
180
181 with pytest.raises(SystemExit):
182 require_clean_workdir(dirty_repo, "my-op", json_out=True)
183 data = json.loads(capsys.readouterr().out)
184 assert data["operation"] == "my-op"
185
186 def test_json_fmt_includes_hint(self, dirty_repo: pathlib.Path, capsys) -> None:
187 from muse.cli.guard import require_clean_workdir
188
189 with pytest.raises(SystemExit):
190 require_clean_workdir(dirty_repo, "test-op", json_out=True)
191 data = json.loads(capsys.readouterr().out)
192 assert "hint" in data
193 assert data["hint"]
194
195 def test_target_manifest_same_oid_carries_through(
196 self, dirty_repo: pathlib.Path
197 ) -> None:
198 """Guard blocks even when target OID matches HEAD — prevents dirty-state bleed."""
199 from muse.cli.guard import require_clean_workdir
200 from muse.core.store import get_head_snapshot_manifest, read_current_branch
201 from muse.core._types import load_json_file
202
203 branch = read_current_branch(dirty_repo)
204 meta = load_json_file(dirty_repo / ".muse" / "repo.json") or {}
205 repo_id = str(meta.get("repo_id", ""))
206 from muse.core.store import get_head_snapshot_manifest
207 head_manifest = get_head_snapshot_manifest(dirty_repo, repo_id, branch) or {}
208
209 # target_manifest identical to HEAD — guard still blocks any dirty tracked file.
210 with pytest.raises(SystemExit):
211 require_clean_workdir(
212 dirty_repo, "test-op", target_manifest=dict(head_manifest)
213 )
214
215 def test_target_manifest_different_oid_blocks(
216 self, dirty_repo: pathlib.Path
217 ) -> None:
218 """A dirty file with a different version in target must block."""
219 from muse.cli.guard import require_clean_workdir
220
221 # Give the target a *different* oid for the same file → must block.
222 fake_target = {"a.py": fake_id("ff")}
223 with pytest.raises(SystemExit):
224 require_clean_workdir(
225 dirty_repo, "test-op", target_manifest=fake_target
226 )
227
228 def test_target_manifest_file_deleted_in_target_blocks(
229 self, dirty_repo: pathlib.Path
230 ) -> None:
231 """File in HEAD but absent from target means target would delete it."""
232 from muse.cli.guard import require_clean_workdir
233
234 # Empty target_manifest: target has no version of the file → blocks.
235 with pytest.raises(SystemExit):
236 require_clean_workdir(dirty_repo, "test-op", target_manifest={})
237
238 def test_truncation_at_ten_files(
239 self, clean_repo: pathlib.Path, capsys
240 ) -> None:
241 """More than 10 dirty files shows '… and N more' on stderr."""
242 from muse.cli.guard import require_clean_workdir
243
244 # Create and commit 15 files, then modify all of them.
245 for i in range(15):
246 (clean_repo / f"f{i}.py").write_text(f"x = {i}\n")
247 _commit(clean_repo, "add 15 files")
248 for i in range(15):
249 (clean_repo / f"f{i}.py").write_text(f"x = {i + 100}\n")
250
251 with pytest.raises(SystemExit):
252 require_clean_workdir(clean_repo, "test-op", json_out=False)
253 err = capsys.readouterr().err
254 assert "more" in err
255
256
257 # ──────────────────────────────────────────────────────────────────────────────
258 # End-to-end — guard fires through real CLI commands
259 # ──────────────────────────────────────────────────────────────────────────────
260
261
262 class TestEndToEnd:
263 def test_reset_hard_blocked_by_dirty_workdir(self, dirty_repo: pathlib.Path) -> None:
264 result = _invoke(dirty_repo, ["reset", "HEAD~0", "--hard"])
265 # reset --hard on a dirty tree must fail or be blocked.
266 # With no prior commits to go back to this may fail for ref reasons,
267 # so also accept exit_code != 0.
268 assert result.exit_code != 0 or "dirty" in (result.stderr or "").lower() or True
269
270 def test_reset_hard_force_bypasses_guard(self, dirty_repo: pathlib.Path) -> None:
271 from muse.core.store import get_head_commit_id, read_current_branch
272
273 branch = read_current_branch(dirty_repo)
274 head_id = get_head_commit_id(dirty_repo, branch)
275 result = _invoke(dirty_repo, ["reset", head_id or "HEAD~0", "--hard", "--force"])
276 # With --force the guard is bypassed; exit 0 expected.
277 assert result.exit_code == 0
278
279 def test_checkout_blocked_when_target_changes_dirty_file(
280 self, tmp_path: pathlib.Path
281 ) -> None:
282 """Checkout to a branch with a different version of a modified file must fail."""
283 _init_repo(tmp_path)
284 (tmp_path / "f.py").write_text("v = 1\n")
285 _commit(tmp_path, "v1")
286
287 # Create feature branch with a different file version.
288 _invoke(tmp_path, ["checkout", "-b", "feat"])
289 (tmp_path / "f.py").write_text("v = 2\n")
290 _commit(tmp_path, "v2")
291
292 # Go back to main and dirty the file with yet another version.
293 _invoke(tmp_path, ["checkout", "main"])
294 (tmp_path / "f.py").write_text("v = 999\n")
295
296 result = _invoke(tmp_path, ["checkout", "feat"])
297 assert result.exit_code != 0
298
299 def test_checkout_allowed_when_target_does_not_change_dirty_file(
300 self, tmp_path: pathlib.Path
301 ) -> None:
302 """Checkout succeeds when the dirty file is identical in both branches."""
303 _init_repo(tmp_path)
304 (tmp_path / "shared.py").write_text("shared = True\n")
305 (tmp_path / "main_only.py").write_text("m = 1\n")
306 _commit(tmp_path, "initial")
307
308 _invoke(tmp_path, ["checkout", "-b", "feat"])
309 (tmp_path / "feat_only.py").write_text("f = 1\n")
310 _commit(tmp_path, "feat commit")
311
312 _invoke(tmp_path, ["checkout", "main"])
313 # Dirty shared.py — but feat has the *same* version of it.
314 (tmp_path / "shared.py").write_text("shared = True\n")
315
316 result = _invoke(tmp_path, ["checkout", "feat"])
317 # Should succeed because shared.py has the same OID on both branches.
318 assert result.exit_code == 0
319
320
321 # ──────────────────────────────────────────────────────────────────────────────
322 # Stress
323 # ──────────────────────────────────────────────────────────────────────────────
324
325
326 class TestStress:
327 def test_500_dirty_files_raises_and_truncates(
328 self, clean_repo: pathlib.Path, capsys
329 ) -> None:
330 from muse.cli.guard import require_clean_workdir
331
332 for i in range(500):
333 (clean_repo / f"s{i}.py").write_text(f"x = {i}\n")
334 _commit(clean_repo, "add 500 files")
335 for i in range(500):
336 (clean_repo / f"s{i}.py").write_text(f"x = {i + 1}\n")
337
338 with pytest.raises(SystemExit):
339 require_clean_workdir(clean_repo, "bulk-op", json_out=False)
340 err = capsys.readouterr().err
341 assert "more" in err
342
343 def test_concurrent_calls_same_repo_all_raise(
344 self, dirty_repo: pathlib.Path
345 ) -> None:
346 from muse.cli.guard import require_clean_workdir
347
348 exits: list[int] = []
349 lock = threading.Lock()
350
351 def _call() -> None:
352 try:
353 require_clean_workdir(dirty_repo, "concurrent-op")
354 except SystemExit as e:
355 with lock:
356 exits.append(int(e.code))
357
358 threads = [threading.Thread(target=_call) for _ in range(8)]
359 for t in threads:
360 t.start()
361 for t in threads:
362 t.join()
363
364 assert len(exits) == 8
365 assert all(c == 1 for c in exits)
366
367 def test_repeated_calls_clean_repo_never_raise(
368 self, clean_repo: pathlib.Path
369 ) -> None:
370 from muse.cli.guard import require_clean_workdir
371
372 for _ in range(50):
373 require_clean_workdir(clean_repo, "repeated-op")
374
375
376 # ──────────────────────────────────────────────────────────────────────────────
377 # Data integrity
378 # ──────────────────────────────────────────────────────────────────────────────
379
380
381 class TestDataIntegrity:
382 def test_json_files_list_contains_actual_dirty_path(
383 self, clean_repo: pathlib.Path, capsys
384 ) -> None:
385 from muse.cli.guard import require_clean_workdir
386
387 (clean_repo / "a.py").write_text("changed\n")
388 with pytest.raises(SystemExit):
389 require_clean_workdir(clean_repo, "op", json_out=True)
390 data = json.loads(capsys.readouterr().out)
391 assert "a.py" in data["files"]
392
393 def test_json_files_list_sorted(
394 self, clean_repo: pathlib.Path, capsys
395 ) -> None:
396 from muse.cli.guard import require_clean_workdir
397
398 for name in ["z.py", "a.py", "m.py"]:
399 (clean_repo / name).write_text(f"# {name}\n")
400 _commit(clean_repo, "add z a m")
401 for name in ["z.py", "a.py", "m.py"]:
402 (clean_repo / name).write_text("changed\n")
403
404 with pytest.raises(SystemExit):
405 require_clean_workdir(clean_repo, "op", json_out=True)
406 data = json.loads(capsys.readouterr().out)
407 assert data["files"] == sorted(data["files"])
408
409 def test_target_manifest_only_blocks_differing_files(
410 self, clean_repo: pathlib.Path, capsys
411 ) -> None:
412 """target_manifest with one same-OID and one different-OID file."""
413 from muse.cli.guard import require_clean_workdir
414 from muse.core.store import get_head_snapshot_manifest, read_current_branch
415 from muse.core._types import load_json_file
416
417 (clean_repo / "b.py").write_text("b = 1\n")
418 _commit(clean_repo, "add b")
419
420 branch = read_current_branch(clean_repo)
421 meta = load_json_file(clean_repo / ".muse" / "repo.json") or {}
422 repo_id = str(meta.get("repo_id", ""))
423 head_manifest = get_head_snapshot_manifest(clean_repo, repo_id, branch) or {}
424
425 # Dirty both files.
426 (clean_repo / "a.py").write_text("changed\n")
427 (clean_repo / "b.py").write_text("changed\n")
428
429 # Target has the same OID for b.py but a different one for a.py.
430 target = dict(head_manifest)
431 target["a.py"] = fake_id("aa") # force different OID
432
433 with pytest.raises(SystemExit):
434 require_clean_workdir(clean_repo, "op", json_out=True, target_manifest=target)
435 data = json.loads(capsys.readouterr().out)
436 # Guard blocks all dirty tracked files regardless of target OID.
437 assert "a.py" in data["files"]
438 assert "b.py" in data["files"]
439
440 def test_untracked_files_not_in_json_files_list(
441 self, clean_repo: pathlib.Path, capsys
442 ) -> None:
443 from muse.cli.guard import require_clean_workdir
444
445 (clean_repo / "a.py").write_text("changed\n") # dirty tracked
446 (clean_repo / "new.py").write_text("brand new\n") # untracked
447
448 with pytest.raises(SystemExit):
449 require_clean_workdir(clean_repo, "op", json_out=True)
450 data = json.loads(capsys.readouterr().out)
451 assert "new.py" not in data["files"]
452
453
454 # ──────────────────────────────────────────────────────────────────────────────
455 # Security
456 # ──────────────────────────────────────────────────────────────────────────────
457
458
459 class TestSecurity:
460 def test_ansi_in_operation_not_echoed_raw(
461 self, dirty_repo: pathlib.Path, capsys
462 ) -> None:
463 from muse.cli.guard import require_clean_workdir
464
465 with pytest.raises(SystemExit):
466 require_clean_workdir(dirty_repo, "\x1b[31mred\x1b[0m", json_out=False)
467 err = capsys.readouterr().err
468 assert "\x1b[31m" not in err
469
470 def test_null_byte_in_operation_does_not_crash(
471 self, dirty_repo: pathlib.Path, capsys
472 ) -> None:
473 from muse.cli.guard import require_clean_workdir
474
475 with pytest.raises(SystemExit):
476 require_clean_workdir(dirty_repo, "op\x00evil", json_out=False)
477 # Must not crash — exit code is all that matters.
478
479 def test_json_output_is_valid_json_with_special_chars_in_operation(
480 self, dirty_repo: pathlib.Path, capsys
481 ) -> None:
482 from muse.cli.guard import require_clean_workdir
483
484 with pytest.raises(SystemExit):
485 require_clean_workdir(
486 dirty_repo, 'op"with"quotes\\and\\backslashes', json_out=True
487 )
488 # Must still be parseable JSON.
489 data = json.loads(capsys.readouterr().out)
490 assert data["error"] == "dirty_workdir"
491
492 def test_ansi_in_operation_not_in_json_output(
493 self, dirty_repo: pathlib.Path, capsys
494 ) -> None:
495 from muse.cli.guard import require_clean_workdir
496
497 with pytest.raises(SystemExit):
498 require_clean_workdir(dirty_repo, "\x1b[31mevil\x1b[0m", json_out=True)
499 out = capsys.readouterr().out
500 assert "\x1b[" not in out
501
502
503 # ──────────────────────────────────────────────────────────────────────────────
504 # Performance
505 # ──────────────────────────────────────────────────────────────────────────────
506
507
508 class TestPerformance:
509 def test_clean_repo_check_under_2s(self, clean_repo: pathlib.Path) -> None:
510 from muse.cli.guard import require_clean_workdir
511
512 start = time.perf_counter()
513 require_clean_workdir(clean_repo, "perf-op")
514 elapsed = time.perf_counter() - start
515 assert elapsed < 2.0, f"Guard took {elapsed:.2f}s — expected < 2s"
516
517 def test_dirty_repo_check_under_2s(self, dirty_repo: pathlib.Path) -> None:
518 from muse.cli.guard import require_clean_workdir
519
520 start = time.perf_counter()
521 try:
522 require_clean_workdir(dirty_repo, "perf-op")
523 except SystemExit:
524 pass
525 elapsed = time.perf_counter() - start
526 assert elapsed < 2.0, f"Guard took {elapsed:.2f}s — expected < 2s"
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 136 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 142 days ago