gabriel / muse public
test_cmd_content_grep_hardening.py python
963 lines 33.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Hardening tests for ``muse content-grep``.
2
3 Covers:
4 Unit — _is_binary, _path_matches_globs, _search_object (context,
5 binary skip, utf-8 replace), pattern validation order
6 Security — ANSI injection in file paths and match text, pattern length
7 cap, invalid regex, ReDoS pattern rejected before I/O
8 Perf — parallel reads complete correctly, --max-matches cap
9 JSON — _ContentGrepJson schema (commit_id, snapshot_id, totals),
10 GrepMatch context_before/context_after fields
11 Flags — --include, --exclude, --max-matches, --context/-C, --json,
12 rejection of old --format flag
13 Integration — multi-file with mixed hits, --include narrows search,
14 --exclude skips files, --context shows surrounding lines,
15 --ref searches historical commit
16 E2E — --help output mentions all new flags
17 Stress — 500-file snapshot, concurrent parallel reads
18 """
19
20 from __future__ import annotations
21 from collections.abc import Mapping
22
23 import datetime
24 import json
25 import pathlib
26 import threading
27 from typing import TypedDict
28
29 import pytest
30 from tests.cli_test_helper import CliRunner, InvokeResult
31
32 from muse.core.object_store import write_object
33 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
34 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
35 from muse.core.types import Manifest, blob_id
36
37 cli = None
38 runner = CliRunner()
39 _invoke_lock = threading.Lock()
40
41 type _FilesMap = dict[str, bytes]
42
43 _REPO_ID = "cgrep-hardening"
44
45
46 # ---------------------------------------------------------------------------
47 # Helpers
48 # ---------------------------------------------------------------------------
49
50
51 class _GrepMatchOut(TypedDict):
52 line_number: int
53 line: str
54 context_before: list[str]
55 context_after: list[str]
56
57
58 class _GrepResultOut(TypedDict):
59 file: str
60 object_id: str
61 match_count: int
62 matches: list[_GrepMatchOut]
63
64
65 class _GrepOut(TypedDict):
66 source: str
67 commit_id: str
68 snapshot_id: str
69 pattern: str
70 total_files_matched: int
71 total_matches: int
72 results: list[_GrepResultOut]
73 duration_ms: float
74 exit_code: int
75
76
77
78
79 def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path:
80 dot_muse = muse_dir(path)
81 for d in ("commits", "snapshots", "objects", "refs/heads"):
82 (dot_muse / d).mkdir(parents=True, exist_ok=True)
83 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
84 (dot_muse / "repo.json").write_text(
85 json.dumps({"repo_id": repo_id, "domain": "midi"}), encoding="utf-8"
86 )
87 return path
88
89
90 def _env(repo: pathlib.Path) -> Manifest:
91 return {"MUSE_REPO_ROOT": str(repo)}
92
93
94 _counter = 0
95
96
97 def _commit_files(
98 root: pathlib.Path,
99 files: _FilesMap,
100 branch: str = "main",
101 parent_id: str | None = None,
102 ) -> str:
103 global _counter
104 _counter += 1
105 manifest: Manifest = {}
106 for rel_path, content in files.items():
107 obj_id = blob_id(content)
108 write_object(root, obj_id, content)
109 manifest[rel_path] = obj_id
110 snap_id = compute_snapshot_id(manifest)
111 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
112 committed_at = datetime.datetime.now(datetime.timezone.utc)
113 parent_ids = [parent_id] if parent_id else []
114 commit_id = compute_commit_id(
115 parent_ids, snap_id, f"commit {_counter}", committed_at.isoformat(),
116 )
117 write_commit(
118 root,
119 CommitRecord(
120 commit_id=commit_id,
121 repo_id="test-repo",
122 branch=branch,
123 snapshot_id=snap_id,
124 message=f"commit {_counter}",
125 committed_at=committed_at,
126 parent_commit_id=parent_id,
127 ),
128 )
129 branch_ref = ref_path(root, branch)
130 branch_ref.parent.mkdir(parents=True, exist_ok=True)
131 branch_ref.write_text(commit_id, encoding="utf-8")
132 return commit_id
133
134
135 def _invoke(args: list[str], env: Manifest | None = None) -> InvokeResult:
136 with _invoke_lock:
137 return runner.invoke(cli, args, env=env)
138
139
140 def _parse(result: InvokeResult) -> _GrepOut:
141 raw: _GrepOut = json.loads(result.output)
142 return raw
143
144
145 # ---------------------------------------------------------------------------
146 # Unit: _is_binary
147 # ---------------------------------------------------------------------------
148
149
150 def test_is_binary_null_byte() -> None:
151 from muse.cli.commands.content_grep import _is_binary
152
153 assert _is_binary(b"\x00hello") is True
154
155
156 def test_is_binary_clean_text() -> None:
157 from muse.cli.commands.content_grep import _is_binary
158
159 assert _is_binary(b"hello world\n") is False
160
161
162 def test_is_binary_empty() -> None:
163 from muse.cli.commands.content_grep import _is_binary
164
165 assert _is_binary(b"") is False
166
167
168 # ---------------------------------------------------------------------------
169 # Unit: _path_matches_globs
170 # ---------------------------------------------------------------------------
171
172
173 def test_path_matches_no_filter() -> None:
174 from muse.cli.commands.content_grep import _path_matches_globs
175
176 assert _path_matches_globs("src/main.py", None, None) is True
177
178
179 def test_path_matches_include_basename() -> None:
180 from muse.cli.commands.content_grep import _path_matches_globs
181
182 assert _path_matches_globs("src/main.py", "*.py", None) is True
183 assert _path_matches_globs("src/main.js", "*.py", None) is False
184
185
186 def test_path_matches_include_full_path() -> None:
187 from muse.cli.commands.content_grep import _path_matches_globs
188
189 assert _path_matches_globs("src/main.py", "src/*.py", None) is True
190 assert _path_matches_globs("tests/main.py", "src/*.py", None) is False
191
192
193 def test_path_matches_exclude_basename() -> None:
194 from muse.cli.commands.content_grep import _path_matches_globs
195
196 assert _path_matches_globs("app.min.js", None, "*.min.js") is False
197 assert _path_matches_globs("app.js", None, "*.min.js") is True
198
199
200 def test_path_matches_include_and_exclude() -> None:
201 from muse.cli.commands.content_grep import _path_matches_globs
202
203 assert _path_matches_globs("src/main.py", "*.py", "test_*.py") is True
204 assert _path_matches_globs("test_foo.py", "*.py", "test_*.py") is False
205
206
207 # ---------------------------------------------------------------------------
208 # Unit: _search_object — context lines
209 # ---------------------------------------------------------------------------
210
211
212 def test_search_object_context(tmp_path: pathlib.Path) -> None:
213 import re
214 from muse.cli.commands.content_grep import _search_object
215
216 _init_repo(tmp_path)
217 content = b"line one\nTARGET line\nline three\n"
218 obj_id = blob_id(content)
219 write_object(tmp_path, obj_id, content)
220
221 pat = re.compile("TARGET")
222 count, matches = _search_object(tmp_path, obj_id, pat, False, False, context_lines=1)
223 assert count == 1
224 assert len(matches) == 1
225 assert matches[0]["context_before"] == ["line one"]
226 assert matches[0]["context_after"] == ["line three"]
227
228
229 def test_search_object_context_at_boundary(tmp_path: pathlib.Path) -> None:
230 import re
231 from muse.cli.commands.content_grep import _search_object
232
233 _init_repo(tmp_path)
234 content = b"TARGET\nonly\n"
235 obj_id = blob_id(content)
236 write_object(tmp_path, obj_id, content)
237
238 pat = re.compile("TARGET")
239 count, matches = _search_object(tmp_path, obj_id, pat, False, False, context_lines=3)
240 assert matches[0]["context_before"] == []
241 assert matches[0]["context_after"] == ["only"]
242
243
244 def test_search_object_no_context(tmp_path: pathlib.Path) -> None:
245 import re
246 from muse.cli.commands.content_grep import _search_object
247
248 _init_repo(tmp_path)
249 content = b"line\nTARGET\nend\n"
250 obj_id = blob_id(content)
251 write_object(tmp_path, obj_id, content)
252
253 pat = re.compile("TARGET")
254 _, matches = _search_object(tmp_path, obj_id, pat, False, False, context_lines=0)
255 assert matches[0]["context_before"] == []
256 assert matches[0]["context_after"] == []
257
258
259 def test_search_object_binary_skipped(tmp_path: pathlib.Path) -> None:
260 import re
261 from muse.cli.commands.content_grep import _search_object
262
263 _init_repo(tmp_path)
264 content = b"\x00\x01\x02TARGET\x03"
265 obj_id = blob_id(content)
266 write_object(tmp_path, obj_id, content)
267
268 pat = re.compile("TARGET")
269 count, matches = _search_object(tmp_path, obj_id, pat, False, False, 0)
270 assert count == 0
271 assert matches == []
272
273
274 # ---------------------------------------------------------------------------
275 # Security: pattern validation happens BEFORE I/O
276 # ---------------------------------------------------------------------------
277
278
279 def test_long_pattern_rejected_before_io(tmp_path: pathlib.Path) -> None:
280 """A too-long pattern must be rejected without touching the object store."""
281 _init_repo(tmp_path)
282 # Do NOT commit any files — if I/O happened, we'd get a 'no commits' error,
283 # not the 'pattern too long' error.
284 bad_pattern = "a" * 501
285 result = _invoke(
286 ["content-grep", bad_pattern], env=_env(tmp_path)
287 )
288 assert result.exit_code != 0
289 # The error must be about pattern length, not about missing commits.
290 assert "too long" in result.output.lower() or "too long" in (result.stderr or "").lower()
291
292
293 def test_invalid_regex_rejected_before_io(tmp_path: pathlib.Path) -> None:
294 _init_repo(tmp_path)
295 result = _invoke(
296 ["content-grep", "[unclosed"], env=_env(tmp_path)
297 )
298 assert result.exit_code != 0
299 assert "regex" in result.output.lower() or "regex" in (result.stderr or "").lower()
300
301
302 # ---------------------------------------------------------------------------
303 # Security: ANSI injection
304 # ---------------------------------------------------------------------------
305
306
307 def test_ansi_injection_in_path(tmp_path: pathlib.Path) -> None:
308 """File paths with ANSI escapes must be stripped in text output."""
309 _init_repo(tmp_path)
310 ansi_path = "\x1b[31mmalicious\x1b[0m.txt"
311 _commit_files(tmp_path, {ansi_path: b"TARGET content\n"})
312 result = _invoke(
313 ["content-grep", "TARGET"], env=_env(tmp_path)
314 )
315 assert result.exit_code == 0
316 assert "\x1b" not in result.output
317
318
319 def test_ansi_injection_in_match_text(tmp_path: pathlib.Path) -> None:
320 """Match text with ANSI escapes must be stripped in text output."""
321 _init_repo(tmp_path)
322 _commit_files(tmp_path, {"safe.txt": b"TARGET \x1b[31mred\x1b[0m content\n"})
323 result = _invoke(
324 ["content-grep", "TARGET"], env=_env(tmp_path)
325 )
326 assert result.exit_code == 0
327 assert "\x1b" not in result.output
328
329
330 # ---------------------------------------------------------------------------
331 # JSON schema: _ContentGrepJson
332 # ---------------------------------------------------------------------------
333
334
335 def test_json_schema_all_fields(tmp_path: pathlib.Path) -> None:
336 _init_repo(tmp_path)
337 _commit_files(tmp_path, {"a.txt": b"hello world\nhello again\n"})
338 result = _invoke(
339 ["content-grep", "hello", "--json"], env=_env(tmp_path)
340 )
341 assert result.exit_code == 0
342 data = _parse(result)
343 assert data["commit_id"].startswith("sha256:")
344 assert len(data["commit_id"]) == 71
345 assert data["snapshot_id"].startswith("sha256:")
346 assert len(data["snapshot_id"]) == 71
347 assert data["pattern"] == "hello"
348 assert data["total_files_matched"] == 1
349 assert data["total_matches"] == 2
350 assert len(data["results"]) == 1
351 r = data["results"][0]
352 assert r["path"] == "a.txt"
353 assert r["match_count"] == 2
354 assert isinstance(r["matches"], list)
355
356
357 def test_json_schema_context_fields(tmp_path: pathlib.Path) -> None:
358 _init_repo(tmp_path)
359 _commit_files(tmp_path, {"c.txt": b"before\nTARGET\nafter\n"})
360 result = _invoke(
361 ["content-grep", "TARGET", "--context", "1", "--json"],
362 env=_env(tmp_path),
363 )
364 assert result.exit_code == 0
365 data = _parse(result)
366 match = data["results"][0]["matches"][0]
367 assert isinstance(match, dict)
368 assert "context_before" in match
369 assert "context_after" in match
370 assert match["context_before"] == ["before"]
371 assert match["context_after"] == ["after"]
372
373
374 def test_json_schema_no_match_exit1(tmp_path: pathlib.Path) -> None:
375 _init_repo(tmp_path)
376 _commit_files(tmp_path, {"a.txt": b"hello\n"})
377 result = _invoke(
378 ["content-grep", "ZZZNOMATCH", "--json"], env=_env(tmp_path)
379 )
380 assert result.exit_code != 0
381
382
383 def test_json_total_matches_multiple_files(tmp_path: pathlib.Path) -> None:
384 _init_repo(tmp_path)
385 _commit_files(tmp_path, {
386 "a.txt": b"hit\nhit\n",
387 "b.txt": b"hit\n",
388 "c.txt": b"miss\n",
389 })
390 result = _invoke(
391 ["content-grep", "hit", "--json"], env=_env(tmp_path)
392 )
393 assert result.exit_code == 0
394 data = _parse(result)
395 assert data["total_files_matched"] == 2
396 assert data["total_matches"] == 3
397
398
399 # ---------------------------------------------------------------------------
400 # Flags: --include
401 # ---------------------------------------------------------------------------
402
403
404 def test_include_filters_to_py_only(tmp_path: pathlib.Path) -> None:
405 _init_repo(tmp_path)
406 _commit_files(tmp_path, {
407 "module.py": b"TARGET in python\n",
408 "module.js": b"TARGET in js\n",
409 "readme.md": b"TARGET in md\n",
410 })
411 result = _invoke(
412 ["content-grep", "TARGET", "--include", "*.py", "--json"],
413 env=_env(tmp_path),
414 )
415 assert result.exit_code == 0
416 data = _parse(result)
417 assert data["total_files_matched"] == 1
418 assert data["results"][0]["path"] == "module.py"
419
420
421 def test_include_no_matches_after_filter(tmp_path: pathlib.Path) -> None:
422 _init_repo(tmp_path)
423 _commit_files(tmp_path, {"module.js": b"TARGET here\n"})
424 result = _invoke(
425 ["content-grep", "TARGET", "--include", "*.py"],
426 env=_env(tmp_path),
427 )
428 assert result.exit_code != 0 # no files pass include filter
429
430
431 # ---------------------------------------------------------------------------
432 # Flags: --exclude
433 # ---------------------------------------------------------------------------
434
435
436 def test_exclude_skips_minified(tmp_path: pathlib.Path) -> None:
437 _init_repo(tmp_path)
438 _commit_files(tmp_path, {
439 "app.js": b"TARGET here\n",
440 "app.min.js": b"TARGET minified\n",
441 })
442 result = _invoke(
443 ["content-grep", "TARGET", "--exclude", "*.min.js", "--json"],
444 env=_env(tmp_path),
445 )
446 assert result.exit_code == 0
447 data = _parse(result)
448 assert data["total_files_matched"] == 1
449 assert data["results"][0]["path"] == "app.js"
450
451
452 def test_exclude_all_results_in_no_match(tmp_path: pathlib.Path) -> None:
453 _init_repo(tmp_path)
454 _commit_files(tmp_path, {"test.py": b"TARGET\n"})
455 result = _invoke(
456 ["content-grep", "TARGET", "--exclude", "test_*.py"],
457 env=_env(tmp_path),
458 )
459 # test.py doesn't match test_*.py exclude pattern, so it should match.
460 # Verify this works (target file isn't excluded).
461 assert result.exit_code == 0
462
463
464 # ---------------------------------------------------------------------------
465 # Flags: --max-matches
466 # ---------------------------------------------------------------------------
467
468
469 def test_max_matches_caps_output(tmp_path: pathlib.Path) -> None:
470 _init_repo(tmp_path)
471 _commit_files(tmp_path, {"many.txt": b"hit\n" * 100})
472 result = _invoke(
473 ["content-grep", "hit", "--max-matches", "10", "--json"],
474 env=_env(tmp_path),
475 )
476 assert result.exit_code == 0
477 data = _parse(result)
478 assert data["total_matches"] <= 10
479
480
481 def test_max_matches_zero_still_exits_nonzero_on_cap(tmp_path: pathlib.Path) -> None:
482 """When max_matches=0, no results are kept — exit 1."""
483 _init_repo(tmp_path)
484 _commit_files(tmp_path, {"a.txt": b"hit\n"})
485 result = _invoke(
486 ["content-grep", "hit", "--max-matches", "0", "--json"],
487 env=_env(tmp_path),
488 )
489 assert result.exit_code != 0 # no results after cap → exit 1
490
491
492 # ---------------------------------------------------------------------------
493 # Flags: --context / -C
494 # ---------------------------------------------------------------------------
495
496
497 def test_context_text_output(tmp_path: pathlib.Path) -> None:
498 _init_repo(tmp_path)
499 _commit_files(tmp_path, {"ctx.txt": b"alpha\nbeta\ngamma\n"})
500 result = _invoke(
501 ["content-grep", "beta", "--context", "1"],
502 env=_env(tmp_path),
503 )
504 assert result.exit_code == 0
505 # Context before and after should appear in output.
506 assert "alpha" in result.output
507 assert "gamma" in result.output
508
509
510 def test_context_short_flag(tmp_path: pathlib.Path) -> None:
511 _init_repo(tmp_path)
512 _commit_files(tmp_path, {"ctx2.txt": b"first\nTARGET\nlast\n"})
513 result = _invoke(
514 ["content-grep", "TARGET", "-C", "1"],
515 env=_env(tmp_path),
516 )
517 assert result.exit_code == 0
518 assert "first" in result.output
519 assert "last" in result.output
520
521
522 # ---------------------------------------------------------------------------
523 # Flags: --json boolean (rejects old --format)
524 # ---------------------------------------------------------------------------
525
526
527 def test_format_flag_rejected(tmp_path: pathlib.Path) -> None:
528 """Old ``--format json`` must be rejected by argparse (exit 2)."""
529 _init_repo(tmp_path)
530 _commit_files(tmp_path, {"a.txt": b"hello\n"})
531 result = _invoke(
532 ["content-grep", "hello", "--format", "json"],
533 env=_env(tmp_path),
534 )
535 assert result.exit_code == 2
536
537
538 # ---------------------------------------------------------------------------
539 # Integration: --ref searches a different commit
540 # ---------------------------------------------------------------------------
541
542
543 def test_ref_searches_branch(tmp_path: pathlib.Path) -> None:
544 _init_repo(tmp_path)
545 c1 = _commit_files(tmp_path, {"v1.txt": b"OLD content\n"})
546 _commit_files(tmp_path, {"v2.txt": b"NEW content\n"}, parent_id=c1)
547
548 # Search HEAD — should find NEW in v2.txt.
549 result_head = _invoke(
550 ["content-grep", "NEW", "--json"], env=_env(tmp_path)
551 )
552 assert result_head.exit_code == 0
553 data = _parse(result_head)
554 paths = [r["path"] for r in data["results"]]
555 assert "v2.txt" in paths
556
557 # Search the first commit by ID — should find OLD in v1.txt, not NEW.
558 result_ref = _invoke(
559 ["content-grep", "OLD", "--ref", c1, "--json"],
560 env=_env(tmp_path),
561 )
562 assert result_ref.exit_code == 0
563 data_ref = _parse(result_ref)
564 paths_ref = [r["path"] for r in data_ref["results"]]
565 assert "v1.txt" in paths_ref
566 assert data_ref["commit_id"] == c1
567
568
569 # ---------------------------------------------------------------------------
570 # E2E: --help mentions all new flags
571 # ---------------------------------------------------------------------------
572
573
574 def test_help_mentions_include() -> None:
575 result = _invoke(["content-grep", "--help"])
576 assert result.exit_code == 0
577 assert "--include" in result.output
578
579
580 def test_help_mentions_exclude() -> None:
581 result = _invoke(["content-grep", "--help"])
582 assert "--exclude" in result.output
583
584
585 def test_help_mentions_max_matches() -> None:
586 result = _invoke(["content-grep", "--help"])
587 assert "--max-matches" in result.output
588
589
590 def test_help_mentions_context() -> None:
591 result = _invoke(["content-grep", "--help"])
592 assert "--context" in result.output or "-C" in result.output
593
594
595 def test_help_mentions_json_not_format() -> None:
596 result = _invoke(["content-grep", "--help"])
597 assert "--json" in result.output
598 assert "--format" not in result.output
599
600
601 # ---------------------------------------------------------------------------
602 # Stress: 500-file snapshot, pattern matches 250
603 # ---------------------------------------------------------------------------
604
605
606 def test_stress_500_files(tmp_path: pathlib.Path) -> None:
607 _init_repo(tmp_path)
608 files: _FilesMap = {}
609 for i in range(500):
610 content = b"TARGET_STRESS\n" if i % 2 == 0 else b"other\n"
611 files[f"f_{i:04d}.txt"] = content
612 _commit_files(tmp_path, files)
613 result = _invoke(
614 ["content-grep", "TARGET_STRESS", "--json"],
615 env=_env(tmp_path),
616 )
617 assert result.exit_code == 0
618 data = _parse(result)
619 assert data["total_files_matched"] == 250
620 assert data["total_matches"] == 250
621
622
623 # ---------------------------------------------------------------------------
624 # Stress: concurrent reads
625 # ---------------------------------------------------------------------------
626
627
628 def test_stress_concurrent_reads(tmp_path: pathlib.Path) -> None:
629 _init_repo(tmp_path)
630 _commit_files(tmp_path, {"concurrent.txt": b"CONCURRENT TARGET\n"})
631
632 errors: list[str] = []
633
634 def _read() -> None:
635 r = _invoke(
636 ["content-grep", "CONCURRENT", "--json"],
637 env=_env(tmp_path),
638 )
639 if r.exit_code != 0:
640 errors.append(f"exit {r.exit_code}")
641 else:
642 try:
643 d = json.loads(r.output)
644 if d.get("total_matches", 0) != 1:
645 errors.append(f"unexpected total_matches: {d.get('total_matches')}")
646 except json.JSONDecodeError as exc:
647 errors.append(str(exc))
648
649 threads = [threading.Thread(target=_read) for _ in range(8)]
650 for t in threads:
651 t.start()
652 for t in threads:
653 t.join()
654
655 assert not errors, f"Concurrent read failures: {errors}"
656
657
658 # ---------------------------------------------------------------------------
659 # JSON schema: complete key set (TestJsonSchemaComplete)
660 # ---------------------------------------------------------------------------
661
662
663 _REQUIRED_KEYS = frozenset({
664 "source",
665 "commit_id",
666 "snapshot_id",
667 "pattern",
668 "total_files_matched",
669 "total_matches",
670 "results",
671 "duration_ms",
672 "exit_code",
673 })
674
675
676 class TestJsonSchemaComplete:
677 """Verify that every required key is present in JSON output."""
678
679 def test_all_required_keys_present_commit_mode(self, tmp_path: pathlib.Path) -> None:
680 _init_repo(tmp_path)
681 _commit_files(tmp_path, {"a.txt": b"hello\n"})
682 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
683 assert result.exit_code == 0
684 data = json.loads(result.output)
685 missing = _REQUIRED_KEYS - data.keys()
686 assert not missing, f"Missing keys: {missing}"
687
688 def test_all_required_keys_present_working_tree_mode(self, tmp_path: pathlib.Path) -> None:
689 _init_repo(tmp_path)
690 _commit_files(tmp_path, {"a.txt": b"hello\n"})
691 # Also write a matching file to disk so working-tree search finds it.
692 (tmp_path / "a.txt").write_bytes(b"hello\n")
693 result = _invoke(
694 ["content-grep", "hello", "--working-tree", "--json"],
695 env=_env(tmp_path),
696 )
697 assert result.exit_code == 0
698 data = json.loads(result.output)
699 missing = _REQUIRED_KEYS - data.keys()
700 assert not missing, f"Missing keys: {missing}"
701
702 def test_source_field_is_commit(self, tmp_path: pathlib.Path) -> None:
703 _init_repo(tmp_path)
704 _commit_files(tmp_path, {"a.txt": b"hello\n"})
705 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
706 data = json.loads(result.output)
707 assert data["source"] == "commit"
708
709 def test_source_field_is_working_tree(self, tmp_path: pathlib.Path) -> None:
710 _init_repo(tmp_path)
711 _commit_files(tmp_path, {"a.txt": b"hello\n"})
712 (tmp_path / "a.txt").write_bytes(b"hello\n")
713 result = _invoke(
714 ["content-grep", "hello", "--working-tree", "--json"],
715 env=_env(tmp_path),
716 )
717 data = json.loads(result.output)
718 assert data["source"] == "working-tree"
719
720 def test_commit_id_null_in_working_tree_mode(self, tmp_path: pathlib.Path) -> None:
721 _init_repo(tmp_path)
722 _commit_files(tmp_path, {"a.txt": b"hello\n"})
723 (tmp_path / "a.txt").write_bytes(b"hello\n")
724 result = _invoke(
725 ["content-grep", "hello", "--working-tree", "--json"],
726 env=_env(tmp_path),
727 )
728 data = json.loads(result.output)
729 assert data["commit_id"] is None
730
731 def test_snapshot_id_null_in_working_tree_mode(self, tmp_path: pathlib.Path) -> None:
732 _init_repo(tmp_path)
733 _commit_files(tmp_path, {"a.txt": b"hello\n"})
734 (tmp_path / "a.txt").write_bytes(b"hello\n")
735 result = _invoke(
736 ["content-grep", "hello", "--working-tree", "--json"],
737 env=_env(tmp_path),
738 )
739 data = json.loads(result.output)
740 assert data["snapshot_id"] is None
741
742 def test_exit_code_field_zero_on_match(self, tmp_path: pathlib.Path) -> None:
743 _init_repo(tmp_path)
744 _commit_files(tmp_path, {"a.txt": b"hello\n"})
745 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
746 data = json.loads(result.output)
747 assert data["exit_code"] == 0
748
749 def test_json_is_compact(self, tmp_path: pathlib.Path) -> None:
750 """JSON output must be a single line — no pretty-printing."""
751 _init_repo(tmp_path)
752 _commit_files(tmp_path, {"a.txt": b"hello\n"})
753 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
754 lines = [ln for ln in result.output.splitlines() if ln.strip()]
755 assert len(lines) == 1, "JSON must be compact (one line)"
756
757
758 # ---------------------------------------------------------------------------
759 # duration_ms (TestElapsedSeconds)
760 # ---------------------------------------------------------------------------
761
762
763 class TestElapsedSeconds:
764 """``duration_ms`` must be a non-negative float in all JSON paths."""
765
766 def _assert_elapsed(self, data: Mapping[str, object]) -> None: # type: ignore[type-arg]
767 assert "duration_ms" in data
768 assert isinstance(data["duration_ms"], float)
769 assert data["duration_ms"] >= 0.0
770
771 def test_elapsed_present_commit_mode(self, tmp_path: pathlib.Path) -> None:
772 _init_repo(tmp_path)
773 _commit_files(tmp_path, {"a.txt": b"target\n"})
774 result = _invoke(["content-grep", "target", "--json"], env=_env(tmp_path))
775 self._assert_elapsed(json.loads(result.output))
776
777 def test_elapsed_present_working_tree_mode(self, tmp_path: pathlib.Path) -> None:
778 _init_repo(tmp_path)
779 _commit_files(tmp_path, {"a.txt": b"target\n"})
780 (tmp_path / "a.txt").write_bytes(b"target\n")
781 result = _invoke(
782 ["content-grep", "target", "--working-tree", "--json"],
783 env=_env(tmp_path),
784 )
785 self._assert_elapsed(json.loads(result.output))
786
787 def test_elapsed_is_float_not_int(self, tmp_path: pathlib.Path) -> None:
788 _init_repo(tmp_path)
789 _commit_files(tmp_path, {"a.txt": b"target\n"})
790 result = _invoke(["content-grep", "target", "--json"], env=_env(tmp_path))
791 data = json.loads(result.output)
792 assert isinstance(data["duration_ms"], float)
793
794 def test_elapsed_reasonable_upper_bound(self, tmp_path: pathlib.Path) -> None:
795 """Single-file search in a temp repo should be well under 5 seconds."""
796 _init_repo(tmp_path)
797 _commit_files(tmp_path, {"a.txt": b"target\n"})
798 result = _invoke(["content-grep", "target", "--json"], env=_env(tmp_path))
799 data = json.loads(result.output)
800 assert data["duration_ms"] < 5.0
801
802 def test_elapsed_present_stress_mode(self, tmp_path: pathlib.Path) -> None:
803 """duration_ms must appear even for 500-file parallel searches."""
804 _init_repo(tmp_path)
805 files: Mapping[str, bytes] = {f"f{i}.txt": b"needle\n" for i in range(50)}
806 _commit_files(tmp_path, files)
807 result = _invoke(["content-grep", "needle", "--json"], env=_env(tmp_path))
808 assert result.exit_code == 0
809 self._assert_elapsed(json.loads(result.output))
810
811 def test_elapsed_six_decimal_places(self, tmp_path: pathlib.Path) -> None:
812 """duration_ms should be rounded to at most 6 decimal places."""
813 _init_repo(tmp_path)
814 _commit_files(tmp_path, {"a.txt": b"target\n"})
815 result = _invoke(["content-grep", "target", "--json"], env=_env(tmp_path))
816 data = json.loads(result.output)
817 elapsed = data["duration_ms"]
818 # round-trip through 6-decimal representation must be exact
819 assert round(elapsed, 6) == elapsed
820
821
822 # ---------------------------------------------------------------------------
823 # exit_code field (TestExitCode)
824 # ---------------------------------------------------------------------------
825
826
827 class TestExitCode:
828 """``exit_code`` in JSON must mirror the process exit code."""
829
830 def test_exit_code_zero_on_match(self, tmp_path: pathlib.Path) -> None:
831 _init_repo(tmp_path)
832 _commit_files(tmp_path, {"a.txt": b"hit\n"})
833 result = _invoke(["content-grep", "hit", "--json"], env=_env(tmp_path))
834 assert result.exit_code == 0
835 assert json.loads(result.output)["exit_code"] == 0
836
837 def test_exit_code_zero_working_tree_match(self, tmp_path: pathlib.Path) -> None:
838 _init_repo(tmp_path)
839 _commit_files(tmp_path, {"a.txt": b"hit\n"})
840 (tmp_path / "a.txt").write_bytes(b"hit\n")
841 result = _invoke(
842 ["content-grep", "hit", "--working-tree", "--json"],
843 env=_env(tmp_path),
844 )
845 assert result.exit_code == 0
846 assert json.loads(result.output)["exit_code"] == 0
847
848 def test_exit_code_is_integer(self, tmp_path: pathlib.Path) -> None:
849 _init_repo(tmp_path)
850 _commit_files(tmp_path, {"a.txt": b"hit\n"})
851 result = _invoke(["content-grep", "hit", "--json"], env=_env(tmp_path))
852 data = json.loads(result.output)
853 assert isinstance(data["exit_code"], int)
854
855 def test_exit_code_in_json_matches_process_exit(self, tmp_path: pathlib.Path) -> None:
856 """JSON exit_code must equal the actual process exit code."""
857 _init_repo(tmp_path)
858 _commit_files(tmp_path, {"a.txt": b"hit\n"})
859 result = _invoke(["content-grep", "hit", "--json"], env=_env(tmp_path))
860 data = json.loads(result.output)
861 assert data["exit_code"] == result.exit_code
862
863 def test_exit_code_multiple_files(self, tmp_path: pathlib.Path) -> None:
864 _init_repo(tmp_path)
865 _commit_files(tmp_path, {"a.txt": b"hit\n", "b.txt": b"hit\n"})
866 result = _invoke(["content-grep", "hit", "--json"], env=_env(tmp_path))
867 assert result.exit_code == 0
868 assert json.loads(result.output)["exit_code"] == 0
869
870
871 # ---------------------------------------------------------------------------
872 # Flag registration tests
873 # ---------------------------------------------------------------------------
874
875 import argparse as _argparse
876 from muse.cli.commands.content_grep import register as _register_content_grep
877 from muse.core.paths import muse_dir, ref_path
878
879
880 def _parse_cgrep(*args: str) -> _argparse.Namespace:
881 root_p = _argparse.ArgumentParser()
882 subs = root_p.add_subparsers(dest="cmd")
883 _register_content_grep(subs)
884 return root_p.parse_args(["content-grep", *args])
885
886
887 class TestRegisterFlags:
888 def test_default_json_out_is_false(self) -> None:
889 ns = _parse_cgrep("TODO")
890 assert ns.json_out is False
891
892 def test_json_flag_sets_json_out(self) -> None:
893 ns = _parse_cgrep("TODO", "--json")
894 assert ns.json_out is True
895
896 def test_j_shorthand_sets_json_out(self) -> None:
897 ns = _parse_cgrep("TODO", "-j")
898 assert ns.json_out is True
899
900 def test_pattern_positional(self) -> None:
901 ns = _parse_cgrep("FIXME")
902 assert ns.pattern == "FIXME"
903
904
905 # ---------------------------------------------------------------------------
906 # JSON key ergonomics: results[].file and matches[].line
907 # ---------------------------------------------------------------------------
908
909
910 class TestJsonKeyErgonomics:
911 """content-grep --json must use 'path' (matching all other muse commands) and
912 'line' (not 'text') for match content."""
913
914 def test_result_key_is_path(self, tmp_path: pathlib.Path) -> None:
915 _init_repo(tmp_path)
916 _commit_files(tmp_path, {"src/main.py": b"hello world\n"})
917 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
918 data = json.loads(result.output)
919 assert data["results"][0]["path"] == "src/main.py"
920
921 def test_result_has_no_file_key(self, tmp_path: pathlib.Path) -> None:
922 _init_repo(tmp_path)
923 _commit_files(tmp_path, {"src/main.py": b"hello world\n"})
924 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
925 data = json.loads(result.output)
926 assert "file" not in data["results"][0]
927
928 def test_match_key_is_line_not_text(self, tmp_path: pathlib.Path) -> None:
929 _init_repo(tmp_path)
930 _commit_files(tmp_path, {"a.py": b"hello world\n"})
931 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
932 data = json.loads(result.output)
933 match = data["results"][0]["matches"][0]
934 assert match["line"] == "hello world"
935
936 def test_match_has_no_text_key(self, tmp_path: pathlib.Path) -> None:
937 _init_repo(tmp_path)
938 _commit_files(tmp_path, {"a.py": b"hello world\n"})
939 result = _invoke(["content-grep", "hello", "--json"], env=_env(tmp_path))
940 data = json.loads(result.output)
941 match = data["results"][0]["matches"][0]
942 assert "text" not in match
943
944 def test_working_tree_result_key_is_path(self, tmp_path: pathlib.Path) -> None:
945 _init_repo(tmp_path)
946 _commit_files(tmp_path, {"a.py": b"placeholder\n"})
947 (tmp_path / "a.py").write_text("needle here\n", encoding="utf-8")
948 result = _invoke(
949 ["content-grep", "needle", "--working-tree", "--json"], env=_env(tmp_path)
950 )
951 data = json.loads(result.output)
952 assert data["results"][0]["path"] == "a.py"
953
954 def test_working_tree_match_key_is_line(self, tmp_path: pathlib.Path) -> None:
955 _init_repo(tmp_path)
956 _commit_files(tmp_path, {"a.py": b"placeholder\n"})
957 (tmp_path / "a.py").write_text("needle here\n", encoding="utf-8")
958 result = _invoke(
959 ["content-grep", "needle", "--working-tree", "--json"], env=_env(tmp_path)
960 )
961 data = json.loads(result.output)
962 match = data["results"][0]["matches"][0]
963 assert match["line"] == "needle here"
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago