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