gabriel / muse public
test_cmd_core_cat.py python
500 lines 18.7 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago
1 """Tests for ``muse cat`` — file-level, domain-agnostic content reader.
2
3 ``muse cat`` is the core-VCS primitive: give me the raw bytes of a tracked
4 file at HEAD or any ref. It does not parse symbols — that is ``muse code cat``.
5 Mirrors the relationship between ``muse blame`` (line-level) and
6 ``muse code blame`` (symbol-level).
7
8 7-tier coverage
9 ---------------
10 Unit argument parsing, address validation (:: rejected)
11 Integration single file, multi-file, --at ref, --json schema
12 E2E historical ref shows old content; working-tree shows new
13 Security symlink rejected, path traversal rejected, ANSI in path
14 Stress large file (1 MiB) completes fast
15 Data integrity JSON content == disk bytes; --at content != HEAD content
16 Performance single file < 0.3s; multi-file 10 files < 1s
17 """
18
19 from __future__ import annotations
20
21 import json
22 import pathlib
23 import textwrap
24 import time
25 import hashlib
26 import datetime
27
28 import pytest
29
30 from tests.cli_test_helper import CliRunner
31 from muse.core.object_store import write_object
32 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
33 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
34 from muse.core._types import long_id
35
36 cli = None
37 runner = CliRunner()
38
39 _REPO_ID = "core-cat-test"
40 _counter = 0
41
42
43 # ---------------------------------------------------------------------------
44 # Helpers
45 # ---------------------------------------------------------------------------
46
47
48 def _sha(data: bytes) -> str:
49 return hashlib.sha256(data).hexdigest()
50
51
52 def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path:
53 muse = path / ".muse"
54 for d in ("commits", "snapshots", "objects", "refs/heads"):
55 (muse / d).mkdir(parents=True, exist_ok=True)
56 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
57 (muse / "repo.json").write_text(
58 json.dumps({"repo_id": repo_id, "domain": "code"}), encoding="utf-8"
59 )
60 return path
61
62
63 def _env(repo: pathlib.Path) -> dict[str, str]:
64 return {"MUSE_REPO_ROOT": str(repo)}
65
66
67 def _add_file(repo: pathlib.Path, rel_path: str, content: bytes) -> str:
68 obj_id = long_id(_sha(content))
69 write_object(repo, obj_id, content)
70 full = repo / rel_path
71 full.parent.mkdir(parents=True, exist_ok=True)
72 full.write_bytes(content)
73 return obj_id
74
75
76 def _commit(
77 repo: pathlib.Path,
78 files: dict[str, bytes],
79 message: str = "c",
80 parent_id: str | None = None,
81 branch: str = "main",
82 ) -> str:
83 global _counter
84 _counter += 1
85 manifest = {p: _add_file(repo, p, c) for p, c in files.items()}
86 snap_id = compute_snapshot_id(manifest)
87 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
88 now = datetime.datetime.now(datetime.timezone.utc)
89 cid = compute_commit_id([parent_id] if parent_id else [], snap_id, message, now.isoformat())
90 write_commit(repo, CommitRecord(
91 commit_id=cid, repo_id=_REPO_ID, branch=branch,
92 snapshot_id=snap_id, message=message, committed_at=now,
93 parent_commit_id=parent_id,
94 ))
95 (repo / ".muse" / "refs" / "heads" / branch).write_text(cid, encoding="utf-8")
96 return cid
97
98
99 # ---------------------------------------------------------------------------
100 # Fixtures
101 # ---------------------------------------------------------------------------
102
103
104 _CONTENT_V1 = b"# version 1\nHELLO = 'world'\n"
105 _CONTENT_V2 = b"# version 2\nHELLO = 'updated'\n"
106
107
108 @pytest.fixture
109 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
110 _init_repo(tmp_path)
111 _commit(tmp_path, {"readme.md": b"# readme\n", "src/main.py": _CONTENT_V1})
112 return tmp_path
113
114
115 @pytest.fixture
116 def two_commit_repo(tmp_path: pathlib.Path) -> pathlib.Path:
117 _init_repo(tmp_path)
118 c1 = _commit(tmp_path, {"src/main.py": _CONTENT_V1}, message="v1")
119 _commit(tmp_path, {"src/main.py": _CONTENT_V2}, message="v2", parent_id=c1)
120 return tmp_path
121
122
123 # ---------------------------------------------------------------------------
124 # Unit: argument validation
125 # ---------------------------------------------------------------------------
126
127
128 class TestArgumentValidation:
129 def test_no_args_exits_nonzero(self, repo: pathlib.Path) -> None:
130 result = runner.invoke(cli, ["cat"], env=_env(repo))
131 assert result.exit_code != 0
132
133 def test_no_args_json_is_valid_json_with_error(self, repo: pathlib.Path) -> None:
134 result = runner.invoke(cli, ["cat", "--json"], env=_env(repo))
135 assert result.exit_code != 0
136 data = json.loads(result.output)
137 assert "error" in data
138
139 def test_symbol_address_rejected(self, repo: pathlib.Path) -> None:
140 """muse cat does not accept file.py::Symbol — that is muse code cat."""
141 result = runner.invoke(cli, ["cat", "src/main.py::HELLO"], env=_env(repo))
142 assert result.exit_code != 0
143
144 def test_symbol_address_json_has_error_code(self, repo: pathlib.Path) -> None:
145 result = runner.invoke(
146 cli, ["cat", "src/main.py::HELLO", "--json"], env=_env(repo)
147 )
148 assert result.exit_code != 0
149 data = json.loads(result.output)
150 assert "error" in data
151
152 def test_untracked_file_exits_nonzero(self, repo: pathlib.Path) -> None:
153 result = runner.invoke(cli, ["cat", "nothere.txt"], env=_env(repo))
154 assert result.exit_code != 0
155
156 def test_untracked_file_json_has_error_code(self, repo: pathlib.Path) -> None:
157 result = runner.invoke(
158 cli, ["cat", "nothere.txt", "--json"], env=_env(repo)
159 )
160 assert result.exit_code != 0
161 data = json.loads(result.output)
162 # Single file error: multi-file schema with errors list
163 assert "errors" in data
164 assert len(data["errors"]) == 1
165 assert "error_code" in data["errors"][0]
166
167
168 # ---------------------------------------------------------------------------
169 # Integration: single file
170 # ---------------------------------------------------------------------------
171
172
173 class TestSingleFile:
174 def test_prints_file_content(self, repo: pathlib.Path) -> None:
175 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
176 assert result.exit_code == 0
177 assert "# readme" in result.output
178
179 def test_json_schema_single_file(self, repo: pathlib.Path) -> None:
180 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
181 assert result.exit_code == 0
182 data = json.loads(result.output)
183 assert "file_path" in data
184 assert "content" in data
185 assert "size_bytes" in data
186 assert "source_ref" in data
187 assert "duration_ms" in data
188
189 def test_json_file_path_correct(self, repo: pathlib.Path) -> None:
190 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
191 data = json.loads(result.output)
192 assert data["file_path"] == "readme.md"
193
194 def test_json_source_ref_working_tree(self, repo: pathlib.Path) -> None:
195 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
196 data = json.loads(result.output)
197 assert data["source_ref"] == "working tree"
198
199 def test_json_size_bytes_accurate(self, repo: pathlib.Path) -> None:
200 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
201 data = json.loads(result.output)
202 assert data["size_bytes"] == len(b"# readme\n")
203
204 def test_json_shorthand_flag(self, repo: pathlib.Path) -> None:
205 result = runner.invoke(cli, ["cat", "readme.md", "-j"], env=_env(repo))
206 assert result.exit_code == 0
207 json.loads(result.output) # valid JSON
208
209 def test_subdirectory_file(self, repo: pathlib.Path) -> None:
210 result = runner.invoke(cli, ["cat", "src/main.py"], env=_env(repo))
211 assert result.exit_code == 0
212 assert "HELLO" in result.output
213
214
215 # ---------------------------------------------------------------------------
216 # Integration: multi-file
217 # ---------------------------------------------------------------------------
218
219
220 class TestMultiFile:
221 def test_multi_file_json_has_files_key(self, repo: pathlib.Path) -> None:
222 result = runner.invoke(
223 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
224 )
225 assert result.exit_code == 0
226 data = json.loads(result.output)
227 assert "files" in data
228 assert len(data["files"]) == 2
229
230 def test_multi_file_json_each_has_schema(self, repo: pathlib.Path) -> None:
231 result = runner.invoke(
232 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
233 )
234 data = json.loads(result.output)
235 for entry in data["files"]:
236 assert "file_path" in entry
237 assert "content" in entry
238 assert "size_bytes" in entry
239
240 def test_multi_file_text_prints_all(self, repo: pathlib.Path) -> None:
241 result = runner.invoke(
242 cli, ["cat", "readme.md", "src/main.py"], env=_env(repo)
243 )
244 assert result.exit_code == 0
245 assert "# readme" in result.output
246 assert "HELLO" in result.output
247
248 def test_multi_file_one_missing_exits_nonzero(self, repo: pathlib.Path) -> None:
249 result = runner.invoke(
250 cli, ["cat", "readme.md", "missing.txt", "--json"], env=_env(repo)
251 )
252 assert result.exit_code != 0
253 data = json.loads(result.output)
254 assert len(data["errors"]) == 1
255 assert len(data["files"]) == 1 # the valid file still returned
256
257
258 # ---------------------------------------------------------------------------
259 # Integration: --at ref
260 # ---------------------------------------------------------------------------
261
262
263 class TestAtRef:
264 def test_at_old_commit_shows_old_content(
265 self, two_commit_repo: pathlib.Path
266 ) -> None:
267 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
268 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
269 result = runner.invoke(
270 cli, ["cat", "src/main.py", "--at", old_cid], env=_env(two_commit_repo)
271 )
272 assert result.exit_code == 0
273 assert "version 1" in result.output
274 assert "updated" not in result.output
275
276 def test_at_json_source_ref_contains_commit(
277 self, two_commit_repo: pathlib.Path
278 ) -> None:
279 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
280 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
281 result = runner.invoke(
282 cli, ["cat", "src/main.py", "--at", old_cid, "--json"],
283 env=_env(two_commit_repo),
284 )
285 data = json.loads(result.output)
286 assert "commit" in data["source_ref"]
287
288 def test_at_branch_name(self, two_commit_repo: pathlib.Path) -> None:
289 result = runner.invoke(
290 cli, ["cat", "src/main.py", "--at", "main"], env=_env(two_commit_repo)
291 )
292 assert result.exit_code == 0
293 assert "version 2" in result.output
294
295 def test_at_bad_ref_exits_nonzero(self, repo: pathlib.Path) -> None:
296 result = runner.invoke(
297 cli, ["cat", "readme.md", "--at", "deadbeef00"], env=_env(repo)
298 )
299 assert result.exit_code != 0
300
301 def test_at_bad_ref_json_has_error(self, repo: pathlib.Path) -> None:
302 result = runner.invoke(
303 cli, ["cat", "readme.md", "--at", "deadbeef00", "--json"], env=_env(repo)
304 )
305 assert result.exit_code != 0
306 data = json.loads(result.output)
307 assert "error" in data
308
309
310 # ---------------------------------------------------------------------------
311 # E2E: working tree vs historical
312 # ---------------------------------------------------------------------------
313
314
315 class TestE2E:
316 def test_head_shows_latest_content(self, two_commit_repo: pathlib.Path) -> None:
317 result = runner.invoke(
318 cli, ["cat", "src/main.py"], env=_env(two_commit_repo)
319 )
320 assert result.exit_code == 0
321 assert "version 2" in result.output
322
323 def test_old_ref_shows_old_content(self, two_commit_repo: pathlib.Path) -> None:
324 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
325 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
326 result = runner.invoke(
327 cli, ["cat", "src/main.py", "--at", old_cid], env=_env(two_commit_repo)
328 )
329 assert result.exit_code == 0
330 assert "version 1" in result.output
331
332 def test_working_tree_edit_visible_without_at(
333 self, repo: pathlib.Path
334 ) -> None:
335 """Uncommitted edit on disk should appear when no --at is given."""
336 (repo / "readme.md").write_text("# modified\n", encoding="utf-8")
337 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
338 assert result.exit_code == 0
339 assert "modified" in result.output
340
341 def test_requires_repo(self, tmp_path: pathlib.Path) -> None:
342 no_repo = tmp_path / "no_repo"
343 no_repo.mkdir()
344 result = runner.invoke(cli, ["cat", "file.py"], env=_env(no_repo))
345 assert result.exit_code != 0
346
347
348 # ---------------------------------------------------------------------------
349 # Security
350 # ---------------------------------------------------------------------------
351
352
353 class TestSecurity:
354 def test_symlink_rejected(self, repo: pathlib.Path) -> None:
355 link = repo / "link.md"
356 link.symlink_to("/etc/passwd")
357 result = runner.invoke(cli, ["cat", "link.md"], env=_env(repo))
358 assert result.exit_code != 0
359
360 def test_path_traversal_rejected(self, repo: pathlib.Path) -> None:
361 result = runner.invoke(
362 cli, ["cat", "../../../etc/passwd"], env=_env(repo)
363 )
364 assert result.exit_code != 0
365
366 def test_ansi_in_path_not_in_output(self, repo: pathlib.Path) -> None:
367 result = runner.invoke(
368 cli, ["cat", "\x1b[31mreadme.md\x1b[0m"], env=_env(repo)
369 )
370 assert result.exit_code != 0
371 assert "\x1b[31m" not in result.output
372
373 def test_newline_in_path_rejected(self, repo: pathlib.Path) -> None:
374 result = runner.invoke(cli, ["cat", "read\nme.md"], env=_env(repo))
375 assert result.exit_code != 0
376
377 def test_null_byte_in_path_rejected(self, repo: pathlib.Path) -> None:
378 result = runner.invoke(cli, ["cat", "read\x00me.md"], env=_env(repo))
379 assert result.exit_code != 0
380
381
382 # ---------------------------------------------------------------------------
383 # Data integrity
384 # ---------------------------------------------------------------------------
385
386
387 class TestDataIntegrity:
388 def test_json_content_equals_disk_bytes(self, repo: pathlib.Path) -> None:
389 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
390 data = json.loads(result.output)
391 disk = (repo / "readme.md").read_bytes().decode("utf-8", errors="replace")
392 assert data["content"] == disk
393
394 def test_json_size_bytes_equals_len_of_content_utf8(
395 self, repo: pathlib.Path
396 ) -> None:
397 result = runner.invoke(cli, ["cat", "src/main.py", "--json"], env=_env(repo))
398 data = json.loads(result.output)
399 assert data["size_bytes"] == len(data["content"].encode("utf-8"))
400
401 def test_at_content_differs_from_head(
402 self, two_commit_repo: pathlib.Path
403 ) -> None:
404 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
405 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
406 head = json.loads(
407 runner.invoke(
408 cli, ["cat", "src/main.py", "--json"], env=_env(two_commit_repo)
409 ).output
410 )["content"]
411 old = json.loads(
412 runner.invoke(
413 cli, ["cat", "src/main.py", "--at", old_cid, "--json"],
414 env=_env(two_commit_repo),
415 ).output
416 )["content"]
417 assert head != old
418
419 def test_multi_file_sizes_sum_correctly(self, repo: pathlib.Path) -> None:
420 result = runner.invoke(
421 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
422 )
423 data = json.loads(result.output)
424 for entry in data["files"]:
425 assert entry["size_bytes"] == len(
426 entry["content"].encode("utf-8", errors="replace")
427 )
428
429 def test_empty_file_handled(self, tmp_path: pathlib.Path) -> None:
430 _init_repo(tmp_path)
431 _commit(tmp_path, {"empty.txt": b""})
432 result = runner.invoke(
433 cli, ["cat", "empty.txt", "--json"], env=_env(tmp_path)
434 )
435 assert result.exit_code == 0
436 data = json.loads(result.output)
437 assert data["content"] == ""
438 assert data["size_bytes"] == 0
439
440
441 # ---------------------------------------------------------------------------
442 # Stress
443 # ---------------------------------------------------------------------------
444
445
446 class TestStress:
447 def test_large_file_1mib(self, tmp_path: pathlib.Path) -> None:
448 _init_repo(tmp_path)
449 content = b"x" * (1024 * 1024)
450 _commit(tmp_path, {"large.bin": content})
451 result = runner.invoke(cli, ["cat", "large.bin"], env=_env(tmp_path))
452 assert result.exit_code == 0
453 assert len(result.output.encode()) >= 1024 * 1024
454
455 def test_10_files_json(self, tmp_path: pathlib.Path) -> None:
456 _init_repo(tmp_path)
457 files = {f"file_{i}.txt": f"content {i}\n".encode() for i in range(10)}
458 _commit(tmp_path, files)
459 args = ["cat"] + list(files.keys()) + ["--json"]
460 result = runner.invoke(cli, args, env=_env(tmp_path))
461 assert result.exit_code == 0
462 data = json.loads(result.output)
463 assert len(data["files"]) == 10
464
465
466 # ---------------------------------------------------------------------------
467 # Performance
468 # ---------------------------------------------------------------------------
469
470
471 class TestPerformance:
472 def test_single_file_under_300ms(self, repo: pathlib.Path) -> None:
473 t0 = time.monotonic()
474 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
475 elapsed = time.monotonic() - t0
476 assert result.exit_code == 0
477 assert elapsed < 0.3
478
479 def test_10_files_under_1s(self, tmp_path: pathlib.Path) -> None:
480 _init_repo(tmp_path)
481 files = {f"f{i}.txt": f"line {i}\n".encode() for i in range(10)}
482 _commit(tmp_path, files)
483 args = ["cat"] + list(files.keys()) + ["--json"]
484 t0 = time.monotonic()
485 result = runner.invoke(cli, args, env=_env(tmp_path))
486 elapsed = time.monotonic() - t0
487 assert result.exit_code == 0
488 assert elapsed < 1.0
489
490 def test_large_file_json_under_1s(self, tmp_path: pathlib.Path) -> None:
491 _init_repo(tmp_path)
492 content = b"a" * (512 * 1024)
493 _commit(tmp_path, {"half_mib.txt": content})
494 t0 = time.monotonic()
495 result = runner.invoke(
496 cli, ["cat", "half_mib.txt", "--json"], env=_env(tmp_path)
497 )
498 elapsed = time.monotonic() - t0
499 assert result.exit_code == 0
500 assert elapsed < 1.0
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago