gabriel / muse public
test_cmd_core_cat.py python
794 lines 30.3 KB
Raw
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 20 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 from collections.abc import Mapping
21
22 import json
23 import pathlib
24 import textwrap
25 import time
26 import hashlib
27 import datetime
28
29 import pytest
30
31 from tests.cli_test_helper import CliRunner
32 from muse.core.object_store import write_object
33 from muse.core.ids import hash_commit, hash_snapshot
34 from muse.core.commits import (
35 CommitRecord,
36 write_commit,
37 )
38 from muse.core.snapshots import (
39 SnapshotRecord,
40 write_snapshot,
41 )
42 from muse.core.types import long_id, blob_id
43 from muse.core.paths import muse_dir, ref_path
44 from muse.plugins.code.stage import make_entry, write_stage
45
46 cli = None
47 runner = CliRunner()
48
49 _REPO_ID = "core-cat-test"
50 _counter = 0
51
52
53 # ---------------------------------------------------------------------------
54 # Helpers
55 # ---------------------------------------------------------------------------
56
57
58
59
60 def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path:
61 dot_muse = muse_dir(path)
62 for d in ("commits", "snapshots", "objects", "refs/heads"):
63 (dot_muse / d).mkdir(parents=True, exist_ok=True)
64 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
65 (dot_muse / "repo.json").write_text(
66 json.dumps({"repo_id": repo_id, "domain": "code"}), encoding="utf-8"
67 )
68 return path
69
70
71 def _env(repo: pathlib.Path) -> Mapping[str, str]:
72 return {"MUSE_REPO_ROOT": str(repo)}
73
74
75 def _add_file(repo: pathlib.Path, rel_path: str, content: bytes) -> str:
76 obj_id = long_id(blob_id(content))
77 write_object(repo, obj_id, content)
78 full = repo / rel_path
79 full.parent.mkdir(parents=True, exist_ok=True)
80 full.write_bytes(content)
81 return obj_id
82
83
84 def _commit(
85 repo: pathlib.Path,
86 files: Mapping[str, bytes],
87 message: str = "c",
88 parent_id: str | None = None,
89 branch: str = "main",
90 ) -> str:
91 global _counter
92 _counter += 1
93 manifest = {p: _add_file(repo, p, c) for p, c in files.items()}
94 snap_id = hash_snapshot(manifest)
95 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
96 now = datetime.datetime.now(datetime.timezone.utc)
97 cid = hash_commit( parent_ids=[parent_id] if parent_id else [],
98 snapshot_id=snap_id,
99 message=message,
100 committed_at_iso=now.isoformat(),
101 )
102 write_commit(repo, CommitRecord(
103 commit_id=cid, branch=branch,
104 snapshot_id=snap_id, message=message, committed_at=now,
105 parent_commit_id=parent_id,
106 ))
107 (ref_path(repo, branch)).write_text(cid, encoding="utf-8")
108 return cid
109
110
111 # ---------------------------------------------------------------------------
112 # Fixtures
113 # ---------------------------------------------------------------------------
114
115
116 _CONTENT_V1 = b"# version 1\nHELLO = 'world'\n"
117 _CONTENT_V2 = b"# version 2\nHELLO = 'updated'\n"
118
119
120 @pytest.fixture
121 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
122 _init_repo(tmp_path)
123 _commit(tmp_path, {"readme.md": b"# readme\n", "src/main.py": _CONTENT_V1})
124 return tmp_path
125
126
127 @pytest.fixture
128 def two_commit_repo(tmp_path: pathlib.Path) -> pathlib.Path:
129 _init_repo(tmp_path)
130 c1 = _commit(tmp_path, {"src/main.py": _CONTENT_V1}, message="v1")
131 _commit(tmp_path, {"src/main.py": _CONTENT_V2}, message="v2", parent_id=c1)
132 return tmp_path
133
134
135 # ---------------------------------------------------------------------------
136 # Unit: argument validation
137 # ---------------------------------------------------------------------------
138
139
140 class TestArgumentValidation:
141 def test_no_args_exits_nonzero(self, repo: pathlib.Path) -> None:
142 result = runner.invoke(cli, ["cat"], env=_env(repo))
143 assert result.exit_code != 0
144
145 def test_no_args_json_is_valid_json_with_error(self, repo: pathlib.Path) -> None:
146 result = runner.invoke(cli, ["cat", "--json"], env=_env(repo))
147 assert result.exit_code != 0
148 data = json.loads(result.output)
149 assert "error" in data
150
151 def test_symbol_address_rejected(self, repo: pathlib.Path) -> None:
152 """muse cat does not accept file.py::Symbol — that is muse code cat."""
153 result = runner.invoke(cli, ["cat", "src/main.py::HELLO"], env=_env(repo))
154 assert result.exit_code != 0
155
156 def test_symbol_address_json_has_error_code(self, repo: pathlib.Path) -> None:
157 result = runner.invoke(
158 cli, ["cat", "src/main.py::HELLO", "--json"], env=_env(repo)
159 )
160 assert result.exit_code != 0
161 data = json.loads(result.output)
162 assert "error" in data
163
164 def test_untracked_file_exits_nonzero(self, repo: pathlib.Path) -> None:
165 result = runner.invoke(cli, ["cat", "nothere.txt"], env=_env(repo))
166 assert result.exit_code != 0
167
168 def test_untracked_file_json_has_error_code(self, repo: pathlib.Path) -> None:
169 result = runner.invoke(
170 cli, ["cat", "nothere.txt", "--json"], env=_env(repo)
171 )
172 assert result.exit_code != 0
173 data = json.loads(result.output)
174 # Single file error: multi-file schema with errors list
175 assert "errors" in data
176 assert len(data["errors"]) == 1
177 assert "error_code" in data["errors"][0]
178
179
180 # ---------------------------------------------------------------------------
181 # Integration: single file
182 # ---------------------------------------------------------------------------
183
184
185 class TestSingleFile:
186 def test_prints_file_content(self, repo: pathlib.Path) -> None:
187 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
188 assert result.exit_code == 0
189 assert "# readme" in result.output
190
191 def test_json_schema_single_file(self, repo: pathlib.Path) -> None:
192 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
193 assert result.exit_code == 0
194 data = json.loads(result.output)
195 assert "path" in data
196 assert "content" in data
197 assert "size_bytes" in data
198 assert "source_ref" in data
199 assert "duration_ms" in data
200
201 def test_json_file_path_correct(self, repo: pathlib.Path) -> None:
202 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
203 data = json.loads(result.output)
204 assert data["path"] == "readme.md"
205
206 def test_json_source_ref_working_tree(self, repo: pathlib.Path) -> None:
207 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
208 data = json.loads(result.output)
209 assert data["source_ref"] == "working tree"
210
211 def test_json_size_bytes_accurate(self, repo: pathlib.Path) -> None:
212 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
213 data = json.loads(result.output)
214 assert data["size_bytes"] == len(b"# readme\n")
215
216 def test_json_shorthand_flag(self, repo: pathlib.Path) -> None:
217 result = runner.invoke(cli, ["cat", "readme.md", "-j"], env=_env(repo))
218 assert result.exit_code == 0
219 json.loads(result.output) # valid JSON
220
221 def test_subdirectory_file(self, repo: pathlib.Path) -> None:
222 result = runner.invoke(cli, ["cat", "src/main.py"], env=_env(repo))
223 assert result.exit_code == 0
224 assert "HELLO" in result.output
225
226
227 # ---------------------------------------------------------------------------
228 # Integration: multi-file
229 # ---------------------------------------------------------------------------
230
231
232 class TestMultiFile:
233 def test_multi_file_json_has_files_key(self, repo: pathlib.Path) -> None:
234 result = runner.invoke(
235 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
236 )
237 assert result.exit_code == 0
238 data = json.loads(result.output)
239 assert "files" in data
240 assert len(data["files"]) == 2
241
242 def test_multi_file_json_each_has_schema(self, repo: pathlib.Path) -> None:
243 result = runner.invoke(
244 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
245 )
246 data = json.loads(result.output)
247 for entry in data["files"]:
248 assert "path" in entry
249 assert "content" in entry
250 assert "size_bytes" in entry
251
252 def test_multi_file_text_prints_all(self, repo: pathlib.Path) -> None:
253 result = runner.invoke(
254 cli, ["cat", "readme.md", "src/main.py"], env=_env(repo)
255 )
256 assert result.exit_code == 0
257 assert "# readme" in result.output
258 assert "HELLO" in result.output
259
260 def test_multi_file_one_missing_exits_nonzero(self, repo: pathlib.Path) -> None:
261 result = runner.invoke(
262 cli, ["cat", "readme.md", "missing.txt", "--json"], env=_env(repo)
263 )
264 assert result.exit_code != 0
265 data = json.loads(result.output)
266 assert len(data["errors"]) == 1
267 assert len(data["files"]) == 1 # the valid file still returned
268
269
270 # ---------------------------------------------------------------------------
271 # Integration: --at ref
272 # ---------------------------------------------------------------------------
273
274
275 class TestAtRef:
276 def test_at_old_commit_shows_old_content(
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], env=_env(two_commit_repo)
283 )
284 assert result.exit_code == 0
285 assert "version 1" in result.output
286 assert "updated" not in result.output
287
288 def test_at_json_source_ref_contains_commit(
289 self, two_commit_repo: pathlib.Path
290 ) -> None:
291 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
292 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
293 result = runner.invoke(
294 cli, ["cat", "src/main.py", "--at", old_cid, "--json"],
295 env=_env(two_commit_repo),
296 )
297 data = json.loads(result.output)
298 assert "commit" in data["source_ref"]
299
300 def test_at_branch_name(self, two_commit_repo: pathlib.Path) -> None:
301 result = runner.invoke(
302 cli, ["cat", "src/main.py", "--at", "main"], env=_env(two_commit_repo)
303 )
304 assert result.exit_code == 0
305 assert "version 2" in result.output
306
307 def test_at_bad_ref_exits_nonzero(self, repo: pathlib.Path) -> None:
308 result = runner.invoke(
309 cli, ["cat", "readme.md", "--at", "deadbeef00"], env=_env(repo)
310 )
311 assert result.exit_code != 0
312
313 def test_at_bad_ref_json_has_error(self, repo: pathlib.Path) -> None:
314 result = runner.invoke(
315 cli, ["cat", "readme.md", "--at", "deadbeef00", "--json"], env=_env(repo)
316 )
317 assert result.exit_code != 0
318 data = json.loads(result.output)
319 assert "error" in data
320
321
322 # ---------------------------------------------------------------------------
323 # E2E: working tree vs historical
324 # ---------------------------------------------------------------------------
325
326
327 class TestE2E:
328 def test_head_shows_latest_content(self, two_commit_repo: pathlib.Path) -> None:
329 result = runner.invoke(
330 cli, ["cat", "src/main.py"], env=_env(two_commit_repo)
331 )
332 assert result.exit_code == 0
333 assert "version 2" in result.output
334
335 def test_old_ref_shows_old_content(self, two_commit_repo: pathlib.Path) -> None:
336 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
337 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
338 result = runner.invoke(
339 cli, ["cat", "src/main.py", "--at", old_cid], env=_env(two_commit_repo)
340 )
341 assert result.exit_code == 0
342 assert "version 1" in result.output
343
344 def test_working_tree_edit_visible_without_at(
345 self, repo: pathlib.Path
346 ) -> None:
347 """Uncommitted edit on disk should appear when no --at is given."""
348 (repo / "readme.md").write_text("# modified\n", encoding="utf-8")
349 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
350 assert result.exit_code == 0
351 assert "modified" in result.output
352
353 def test_requires_repo(self, tmp_path: pathlib.Path) -> None:
354 no_repo = tmp_path / "no_repo"
355 no_repo.mkdir()
356 result = runner.invoke(cli, ["cat", "file.py"], env=_env(no_repo))
357 assert result.exit_code != 0
358
359
360 # ---------------------------------------------------------------------------
361 # Security
362 # ---------------------------------------------------------------------------
363
364
365 class TestSecurity:
366 def test_symlink_rejected(self, repo: pathlib.Path) -> None:
367 link = repo / "link.md"
368 link.symlink_to("/etc/passwd")
369 result = runner.invoke(cli, ["cat", "link.md"], env=_env(repo))
370 assert result.exit_code != 0
371
372 def test_path_traversal_rejected(self, repo: pathlib.Path) -> None:
373 result = runner.invoke(
374 cli, ["cat", "../../../etc/passwd"], env=_env(repo)
375 )
376 assert result.exit_code != 0
377
378 def test_ansi_in_path_not_in_output(self, repo: pathlib.Path) -> None:
379 result = runner.invoke(
380 cli, ["cat", "\x1b[31mreadme.md\x1b[0m"], env=_env(repo)
381 )
382 assert result.exit_code != 0
383 assert "\x1b[31m" not in result.output
384
385 def test_newline_in_path_rejected(self, repo: pathlib.Path) -> None:
386 result = runner.invoke(cli, ["cat", "read\nme.md"], env=_env(repo))
387 assert result.exit_code != 0
388
389 def test_null_byte_in_path_rejected(self, repo: pathlib.Path) -> None:
390 result = runner.invoke(cli, ["cat", "read\x00me.md"], env=_env(repo))
391 assert result.exit_code != 0
392
393
394 # ---------------------------------------------------------------------------
395 # Data integrity
396 # ---------------------------------------------------------------------------
397
398
399 class TestDataIntegrity:
400 def test_json_content_equals_disk_bytes(self, repo: pathlib.Path) -> None:
401 result = runner.invoke(cli, ["cat", "readme.md", "--json"], env=_env(repo))
402 data = json.loads(result.output)
403 disk = (repo / "readme.md").read_bytes().decode("utf-8", errors="replace")
404 assert data["content"] == disk
405
406 def test_json_size_bytes_equals_len_of_content_utf8(
407 self, repo: pathlib.Path
408 ) -> None:
409 result = runner.invoke(cli, ["cat", "src/main.py", "--json"], env=_env(repo))
410 data = json.loads(result.output)
411 assert data["size_bytes"] == len(data["content"].encode("utf-8"))
412
413 def test_at_content_differs_from_head(
414 self, two_commit_repo: pathlib.Path
415 ) -> None:
416 log = runner.invoke(cli, ["log", "--json"], env=_env(two_commit_repo))
417 old_cid = json.loads(log.output)["commits"][-1]["commit_id"]
418 head = json.loads(
419 runner.invoke(
420 cli, ["cat", "src/main.py", "--json"], env=_env(two_commit_repo)
421 ).output
422 )["content"]
423 old = json.loads(
424 runner.invoke(
425 cli, ["cat", "src/main.py", "--at", old_cid, "--json"],
426 env=_env(two_commit_repo),
427 ).output
428 )["content"]
429 assert head != old
430
431 def test_multi_file_sizes_sum_correctly(self, repo: pathlib.Path) -> None:
432 result = runner.invoke(
433 cli, ["cat", "readme.md", "src/main.py", "--json"], env=_env(repo)
434 )
435 data = json.loads(result.output)
436 for entry in data["files"]:
437 assert entry["size_bytes"] == len(
438 entry["content"].encode("utf-8", errors="replace")
439 )
440
441 def test_empty_file_handled(self, tmp_path: pathlib.Path) -> None:
442 _init_repo(tmp_path)
443 _commit(tmp_path, {"empty.txt": b""})
444 result = runner.invoke(
445 cli, ["cat", "empty.txt", "--json"], env=_env(tmp_path)
446 )
447 assert result.exit_code == 0
448 data = json.loads(result.output)
449 assert data["content"] == ""
450 assert data["size_bytes"] == 0
451
452
453 # ---------------------------------------------------------------------------
454 # Stress
455 # ---------------------------------------------------------------------------
456
457
458 class TestStress:
459 def test_large_file_1mib(self, tmp_path: pathlib.Path) -> None:
460 _init_repo(tmp_path)
461 content = b"x" * (1024 * 1024)
462 _commit(tmp_path, {"large.bin": content})
463 result = runner.invoke(cli, ["cat", "large.bin"], env=_env(tmp_path))
464 assert result.exit_code == 0
465 assert len(result.output.encode()) >= 1024 * 1024
466
467 def test_10_files_json(self, tmp_path: pathlib.Path) -> None:
468 _init_repo(tmp_path)
469 files = {f"file_{i}.txt": f"content {i}\n".encode() for i in range(10)}
470 _commit(tmp_path, files)
471 args = ["cat"] + list(files.keys()) + ["--json"]
472 result = runner.invoke(cli, args, env=_env(tmp_path))
473 assert result.exit_code == 0
474 data = json.loads(result.output)
475 assert len(data["files"]) == 10
476
477
478 # ---------------------------------------------------------------------------
479 # Performance
480 # ---------------------------------------------------------------------------
481
482
483 class TestPerformance:
484 def test_single_file_under_300ms(self, repo: pathlib.Path) -> None:
485 t0 = time.monotonic()
486 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
487 elapsed = time.monotonic() - t0
488 assert result.exit_code == 0
489 assert elapsed < 0.3
490
491 def test_10_files_under_1s(self, tmp_path: pathlib.Path) -> None:
492 _init_repo(tmp_path)
493 files = {f"f{i}.txt": f"line {i}\n".encode() for i in range(10)}
494 _commit(tmp_path, files)
495 args = ["cat"] + list(files.keys()) + ["--json"]
496 t0 = time.monotonic()
497 result = runner.invoke(cli, args, env=_env(tmp_path))
498 elapsed = time.monotonic() - t0
499 assert result.exit_code == 0
500 assert elapsed < 1.0
501
502 def test_large_file_json_under_1s(self, tmp_path: pathlib.Path) -> None:
503 _init_repo(tmp_path)
504 content = b"a" * (512 * 1024)
505 _commit(tmp_path, {"half_mib.txt": content})
506 t0 = time.monotonic()
507 result = runner.invoke(
508 cli, ["cat", "half_mib.txt", "--json"], env=_env(tmp_path)
509 )
510 elapsed = time.monotonic() - t0
511 assert result.exit_code == 0
512 assert elapsed < 1.0
513
514
515 # ---------------------------------------------------------------------------
516 # Phase 2: untracked files error; staged files read from object store
517 # ---------------------------------------------------------------------------
518
519
520 def _stage_file(repo: pathlib.Path, rel_path: str, content: bytes) -> str:
521 """Write blob to object store and add a stage entry — simulates muse code add."""
522 obj_id = long_id(blob_id(content))
523 write_object(repo, obj_id, content)
524 entries = {rel_path: make_entry(obj_id, "A")}
525 write_stage(repo, entries)
526 return obj_id
527
528
529 class TestUntrackedFileErrors:
530 """muse cat must error on files that exist on disk but are not tracked."""
531
532 def test_untracked_on_disk_exits_nonzero(self, repo: pathlib.Path) -> None:
533 (repo / "ghost.py").write_text("x = 1\n")
534 result = runner.invoke(cli, ["cat", "ghost.py"], env=_env(repo))
535 assert result.exit_code != 0
536
537 def test_untracked_on_disk_json_error_code_is_file_not_tracked(
538 self, repo: pathlib.Path
539 ) -> None:
540 (repo / "ghost.py").write_text("x = 1\n")
541 result = runner.invoke(cli, ["cat", "ghost.py", "--json"], env=_env(repo))
542 assert result.exit_code != 0
543 data = json.loads(result.output)
544 assert data["errors"][0]["error_code"] == "FILE_NOT_TRACKED"
545
546 def test_untracked_on_disk_does_not_leak_content(
547 self, repo: pathlib.Path
548 ) -> None:
549 """Content of an untracked file must never appear in the output."""
550 (repo / "secret.py").write_text("password = 'hunter2'\n")
551 result = runner.invoke(cli, ["cat", "secret.py"], env=_env(repo))
552 assert "hunter2" not in result.output
553
554 def test_untracked_on_disk_json_does_not_leak_content(
555 self, repo: pathlib.Path
556 ) -> None:
557 (repo / "secret.py").write_text("password = 'hunter2'\n")
558 result = runner.invoke(cli, ["cat", "secret.py", "--json"], env=_env(repo))
559 assert "hunter2" not in result.output
560
561 def test_tracked_file_still_readable(self, repo: pathlib.Path) -> None:
562 """Tracked files must not be broken by the untracked-check change."""
563 result = runner.invoke(cli, ["cat", "readme.md"], env=_env(repo))
564 assert result.exit_code == 0
565 assert "# readme" in result.output
566
567
568 class TestStagedFileReads:
569 """muse cat must read staged files (not yet committed) from the object store."""
570
571 def test_staged_added_file_readable(self, tmp_path: pathlib.Path) -> None:
572 _init_repo(tmp_path)
573 _commit(tmp_path, {"existing.py": b"old = 1\n"})
574 content = b"new_fn = lambda: 42\n"
575 (tmp_path / "new_file.py").write_bytes(content)
576 _stage_file(tmp_path, "new_file.py", content)
577 result = runner.invoke(cli, ["cat", "new_file.py"], env=_env(tmp_path))
578 assert result.exit_code == 0
579 assert "new_fn" in result.output
580
581 def test_staged_added_file_json_schema(self, tmp_path: pathlib.Path) -> None:
582 _init_repo(tmp_path)
583 _commit(tmp_path, {"base.py": b"x = 1\n"})
584 content = b"staged = True\n"
585 (tmp_path / "staged.py").write_bytes(content)
586 _stage_file(tmp_path, "staged.py", content)
587 result = runner.invoke(cli, ["cat", "staged.py", "--json"], env=_env(tmp_path))
588 assert result.exit_code == 0
589 data = json.loads(result.output)
590 assert "path" in data
591 assert "content" in data
592 assert "staged" in data["content"]
593
594 def test_staged_but_not_on_disk_reads_from_store(
595 self, tmp_path: pathlib.Path
596 ) -> None:
597 """Staged blob readable even after the file is deleted from disk."""
598 _init_repo(tmp_path)
599 _commit(tmp_path, {"base.py": b"x = 1\n"})
600 content = b"will_be_deleted = True\n"
601 disk_path = tmp_path / "staged_only.py"
602 disk_path.write_bytes(content)
603 _stage_file(tmp_path, "staged_only.py", content)
604 disk_path.unlink() # simulate file removed from disk after staging
605 result = runner.invoke(cli, ["cat", "staged_only.py"], env=_env(tmp_path))
606 assert result.exit_code == 0
607 assert "will_be_deleted" in result.output
608
609 def test_staged_file_json_content_matches_blob(
610 self, tmp_path: pathlib.Path
611 ) -> None:
612 _init_repo(tmp_path)
613 _commit(tmp_path, {"base.py": b"x = 1\n"})
614 content = b"blob_content = 'exact'\n"
615 (tmp_path / "check.py").write_bytes(content)
616 _stage_file(tmp_path, "check.py", content)
617 result = runner.invoke(cli, ["cat", "check.py", "--json"], env=_env(tmp_path))
618 data = json.loads(result.output)
619 assert data["content"] == content.decode("utf-8")
620 assert data["size_bytes"] == len(content)
621
622
623 class TestRegisterFlags:
624 def test_default_json_out_is_false(self) -> None:
625 import argparse
626 from muse.cli.commands.core_cat import register
627 p = argparse.ArgumentParser()
628 subs = p.add_subparsers()
629 register(subs)
630 args = p.parse_args(["cat"])
631 assert args.json_out is False
632
633 def test_json_flag_sets_json_out(self) -> None:
634 import argparse
635 from muse.cli.commands.core_cat import register
636 p = argparse.ArgumentParser()
637 subs = p.add_subparsers()
638 register(subs)
639 args = p.parse_args(["cat", "--json"])
640 assert args.json_out is True
641
642 def test_j_shorthand_sets_json_out(self) -> None:
643 import argparse
644 from muse.cli.commands.core_cat import register
645 p = argparse.ArgumentParser()
646 subs = p.add_subparsers()
647 register(subs)
648 args = p.parse_args(["cat", "-j"])
649 assert args.json_out is True
650
651
652 # ---------------------------------------------------------------------------
653 # Phase 4: --staged flag
654 # ---------------------------------------------------------------------------
655
656
657 def _stage_file_core(repo: pathlib.Path, rel_path: str, content: bytes) -> str:
658 """Write blob to object store and add a stage entry — simulates muse code add."""
659 obj_id = long_id(blob_id(content))
660 write_object(repo, obj_id, content)
661 entries = {rel_path: make_entry(obj_id, "A")}
662 write_stage(repo, entries)
663 full = repo / rel_path
664 full.parent.mkdir(parents=True, exist_ok=True)
665 full.write_bytes(content)
666 return obj_id
667
668
669 class TestCatStaged:
670 """muse cat --staged reads the staged index rather than disk or a commit."""
671
672 def test_staged_reads_staged_blob_not_disk(self, tmp_path: pathlib.Path) -> None:
673 """--staged returns staged content even when disk has a different version."""
674 _init_repo(tmp_path)
675 committed = b"committed = 1\n"
676 _commit(tmp_path, {"f.py": committed})
677 staged = b"staged = 2\n"
678 obj_id = long_id(blob_id(staged))
679 write_object(tmp_path, obj_id, staged)
680 write_stage(tmp_path, {"f.py": make_entry(obj_id, "M")})
681 # Disk still has old content — stage has new content
682 result = runner.invoke(cli, ["cat", "f.py", "--staged"], env=_env(tmp_path))
683 assert result.exit_code == 0
684 assert "staged = 2" in result.output
685 assert "committed = 1" not in result.output
686
687 def test_staged_json_source_ref_is_staged(self, tmp_path: pathlib.Path) -> None:
688 _init_repo(tmp_path)
689 _commit(tmp_path, {"f.py": b"x = 1\n"})
690 content = b"y = 2\n"
691 _stage_file_core(tmp_path, "f.py", content)
692 result = runner.invoke(
693 cli, ["cat", "f.py", "--staged", "--json"], env=_env(tmp_path)
694 )
695 assert result.exit_code == 0
696 data = json.loads(result.output)
697 assert data["source_ref"] == "staged"
698
699 def test_staged_on_committed_file_without_restaing_shows_committed_content(
700 self, tmp_path: pathlib.Path
701 ) -> None:
702 """A committed file with no stage entry returns the HEAD version."""
703 _init_repo(tmp_path)
704 content = b"committed_only = True\n"
705 _commit(tmp_path, {"only.py": content})
706 result = runner.invoke(
707 cli, ["cat", "only.py", "--staged"], env=_env(tmp_path)
708 )
709 assert result.exit_code == 0
710 assert "committed_only" in result.output
711
712 def test_staged_deletion_gives_file_not_tracked(self, tmp_path: pathlib.Path) -> None:
713 """A file staged for deletion (mode=D) is not readable via --staged."""
714 _init_repo(tmp_path)
715 _commit(tmp_path, {"del.py": b"to_delete = 1\n"})
716 write_stage(tmp_path, {"del.py": make_entry("a" * 64, "D")})
717 result = runner.invoke(
718 cli, ["cat", "del.py", "--staged"], env=_env(tmp_path)
719 )
720 assert result.exit_code != 0
721
722 def test_staged_deletion_json_error_code(self, tmp_path: pathlib.Path) -> None:
723 _init_repo(tmp_path)
724 _commit(tmp_path, {"del.py": b"to_delete = 1\n"})
725 write_stage(tmp_path, {"del.py": make_entry("a" * 64, "D")})
726 result = runner.invoke(
727 cli, ["cat", "del.py", "--staged", "--json"], env=_env(tmp_path)
728 )
729 assert result.exit_code != 0
730 data = json.loads(result.output)
731 assert data["errors"][0]["error_code"] == "FILE_NOT_TRACKED"
732
733 def test_staged_and_at_mutually_exclusive(self, tmp_path: pathlib.Path) -> None:
734 _init_repo(tmp_path)
735 _commit(tmp_path, {"f.py": b"x = 1\n"})
736 result = runner.invoke(
737 cli, ["cat", "f.py", "--staged", "--at", "HEAD"], env=_env(tmp_path)
738 )
739 assert result.exit_code != 0
740
741 def test_staged_and_at_json_error_code(self, tmp_path: pathlib.Path) -> None:
742 _init_repo(tmp_path)
743 _commit(tmp_path, {"f.py": b"x = 1\n"})
744 result = runner.invoke(
745 cli, ["cat", "f.py", "--staged", "--at", "HEAD", "--json"],
746 env=_env(tmp_path),
747 )
748 assert result.exit_code != 0
749 data = json.loads(result.output)
750 assert "error_code" in data
751 assert data["error_code"] == "MUTUALLY_EXCLUSIVE"
752
753 def test_staged_untracked_file_not_readable(self, tmp_path: pathlib.Path) -> None:
754 _init_repo(tmp_path)
755 _commit(tmp_path, {"f.py": b"x = 1\n"})
756 result = runner.invoke(
757 cli, ["cat", "untracked.py", "--staged"], env=_env(tmp_path)
758 )
759 assert result.exit_code != 0
760
761 def test_staged_ignores_working_tree_edits(self, tmp_path: pathlib.Path) -> None:
762 """--staged must not be influenced by on-disk edits after staging."""
763 _init_repo(tmp_path)
764 _commit(tmp_path, {"f.py": b"v1 = 1\n"})
765 staged_content = b"v2 = 2\n"
766 obj_id = long_id(blob_id(staged_content))
767 write_object(tmp_path, obj_id, staged_content)
768 write_stage(tmp_path, {"f.py": make_entry(obj_id, "M")})
769 # After staging, modify the file again on disk (not re-staged)
770 (tmp_path / "f.py").write_text("v3 = 3\n", encoding="utf-8")
771 result = runner.invoke(
772 cli, ["cat", "f.py", "--staged"], env=_env(tmp_path)
773 )
774 assert result.exit_code == 0
775 assert "v2 = 2" in result.output
776 assert "v3 = 3" not in result.output
777
778 def test_staged_flag_registered(self) -> None:
779 import argparse
780 from muse.cli.commands.core_cat import register
781 p = argparse.ArgumentParser()
782 subs = p.add_subparsers()
783 register(subs)
784 args = p.parse_args(["cat", "--staged", "f.py"])
785 assert args.staged is True
786
787 def test_staged_default_is_false(self) -> None:
788 import argparse
789 from muse.cli.commands.core_cat import register
790 p = argparse.ArgumentParser()
791 subs = p.add_subparsers()
792 register(subs)
793 args = p.parse_args(["cat", "f.py"])
794 assert args.staged is False
File History 1 commit
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 20 days ago