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