gabriel / muse public
test_plumbing_cat_object.py python
294 lines 11.1 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago
1 """Comprehensive tests for ``muse plumbing cat-object``.
2
3 Coverage tiers
4 --------------
5 - Unit: _CHUNK constant, _FORMAT_CHOICES
6 - Integration: raw/info formats, --json alias, missing/invalid object_id
7 - Security: ANSI in object_id error, path traversal object_id
8 - Stress: 10 MiB object streaming, 200 sequential reads
9 """
10 from __future__ import annotations
11
12 import hashlib
13 import json
14 import pathlib
15
16 from muse.core.errors import ExitCode
17 from muse.core.object_store import write_object
18 from tests.cli_test_helper import CliRunner, InvokeResult
19
20 runner = CliRunner()
21
22
23 # ---------------------------------------------------------------------------
24 # Helpers
25 # ---------------------------------------------------------------------------
26
27 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
28 """Minimal .muse/ structure."""
29 repo = tmp_path / "repo"
30 muse = repo / ".muse"
31 for sub in ("objects", "commits", "snapshots", "refs/heads"):
32 (muse / sub).mkdir(parents=True)
33 (muse / "HEAD").write_text("ref: refs/heads/main")
34 (muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"}))
35 return repo
36
37
38 def _store(repo: pathlib.Path, content: bytes) -> str:
39 """Write content to the object store and return its object_id."""
40 oid = hashlib.sha256(content).hexdigest()
41 write_object(repo, oid, content)
42 return oid
43
44
45 def _cat(repo: pathlib.Path, *args: str) -> InvokeResult:
46 from muse.cli.app import main as cli
47 return runner.invoke(
48 cli,
49 ["cat-object", *args],
50 env={"MUSE_REPO_ROOT": str(repo)},
51 )
52
53
54 # ---------------------------------------------------------------------------
55 # Unit — module constants
56 # ---------------------------------------------------------------------------
57
58
59 class TestConstants:
60 def test_chunk_size_is_64kib(self) -> None:
61 from muse.cli.commands.plumbing.cat_object import _CHUNK
62 assert _CHUNK == 65536
63
64 def test_format_choices_correct(self) -> None:
65 from muse.cli.commands.plumbing.cat_object import _FORMAT_CHOICES
66 assert "raw" in _FORMAT_CHOICES
67 assert "info" in _FORMAT_CHOICES
68 # "json" must NOT be a format choice — --json is an alias for "info"
69 assert "json" not in _FORMAT_CHOICES
70
71
72 # ---------------------------------------------------------------------------
73 # Integration — raw format
74 # ---------------------------------------------------------------------------
75
76
77 class TestRawFormat:
78 def test_raw_bytes_match_stored_content(self, tmp_path: pathlib.Path) -> None:
79 repo = _make_repo(tmp_path)
80 content = b"hello object store"
81 oid = _store(repo, content)
82 result = _cat(repo, oid)
83 assert result.exit_code == 0
84 assert result.stdout_bytes == content
85
86 def test_raw_is_default_format(self, tmp_path: pathlib.Path) -> None:
87 repo = _make_repo(tmp_path)
88 content = b"default format"
89 oid = _store(repo, content)
90 # No --format flag → should default to raw
91 result = _cat(repo, oid)
92 assert result.exit_code == 0
93 assert result.stdout_bytes == content
94
95 def test_raw_binary_content_preserved(self, tmp_path: pathlib.Path) -> None:
96 repo = _make_repo(tmp_path)
97 content = bytes(range(256)) # All byte values including null, control chars
98 oid = _store(repo, content)
99 result = _cat(repo, oid)
100 assert result.exit_code == 0
101 assert result.stdout_bytes == content
102
103 def test_raw_empty_object(self, tmp_path: pathlib.Path) -> None:
104 repo = _make_repo(tmp_path)
105 content = b""
106 oid = _store(repo, content)
107 result = _cat(repo, oid)
108 assert result.exit_code == 0
109 assert result.stdout_bytes == content
110
111 def test_explicit_format_raw(self, tmp_path: pathlib.Path) -> None:
112 repo = _make_repo(tmp_path)
113 content = b"explicit raw"
114 oid = _store(repo, content)
115 result = _cat(repo, "--format", "raw", oid)
116 assert result.exit_code == 0
117 assert result.stdout_bytes == content
118
119
120 # ---------------------------------------------------------------------------
121 # Integration — info / --json format
122 # ---------------------------------------------------------------------------
123
124
125 class TestInfoFormat:
126 def test_info_format_shape(self, tmp_path: pathlib.Path) -> None:
127 repo = _make_repo(tmp_path)
128 content = b"info content"
129 oid = _store(repo, content)
130 result = _cat(repo, "--format", "info", oid)
131 assert result.exit_code == 0
132 data = json.loads(result.output)
133 assert data["object_id"] == oid
134 assert data["present"] is True
135 assert data["size_bytes"] == len(content)
136
137 def test_json_flag_is_alias_for_info(self, tmp_path: pathlib.Path) -> None:
138 """--json must work and produce info-format JSON — this was broken before the audit."""
139 repo = _make_repo(tmp_path)
140 content = b"json alias test"
141 oid = _store(repo, content)
142 result = _cat(repo, "--json", oid)
143 assert result.exit_code == 0, f"--json failed: {result.output}"
144 data = json.loads(result.output)
145 assert data["object_id"] == oid
146 assert data["present"] is True
147 assert data["size_bytes"] == len(content)
148
149 def test_info_does_not_emit_content(self, tmp_path: pathlib.Path) -> None:
150 repo = _make_repo(tmp_path)
151 content = b"secret bytes"
152 oid = _store(repo, content)
153 result = _cat(repo, "--format", "info", oid)
154 assert result.exit_code == 0
155 # Output must be JSON only — not the raw content
156 data = json.loads(result.output)
157 assert "object_id" in data
158 assert content not in result.output.encode()
159
160 def test_info_size_matches_actual_file(self, tmp_path: pathlib.Path) -> None:
161 repo = _make_repo(tmp_path)
162 content = b"size check " * 100
163 oid = _store(repo, content)
164 result = _cat(repo, "--json", oid)
165 data = json.loads(result.output)
166 assert data["size_bytes"] == len(content)
167
168 def test_missing_object_info_has_present_false(self, tmp_path: pathlib.Path) -> None:
169 repo = _make_repo(tmp_path)
170 oid = "a" * 64
171 result = _cat(repo, "--format", "info", oid)
172 assert result.exit_code == ExitCode.USER_ERROR
173 data = json.loads(result.output)
174 assert data["present"] is False
175 assert data["size_bytes"] == 0
176
177 def test_json_flag_missing_object_has_present_false(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 oid = "b" * 64
180 result = _cat(repo, "--json", oid)
181 assert result.exit_code == ExitCode.USER_ERROR
182 data = json.loads(result.output)
183 assert data["present"] is False
184
185
186 # ---------------------------------------------------------------------------
187 # Integration — error paths
188 # ---------------------------------------------------------------------------
189
190
191 class TestErrorPaths:
192 def test_missing_object_raw_errors(self, tmp_path: pathlib.Path) -> None:
193 repo = _make_repo(tmp_path)
194 result = _cat(repo, "c" * 64)
195 assert result.exit_code == ExitCode.USER_ERROR
196
197 def test_invalid_object_id_too_short(self, tmp_path: pathlib.Path) -> None:
198 repo = _make_repo(tmp_path)
199 result = _cat(repo, "abc123")
200 assert result.exit_code == ExitCode.USER_ERROR
201
202 def test_invalid_object_id_uppercase(self, tmp_path: pathlib.Path) -> None:
203 repo = _make_repo(tmp_path)
204 result = _cat(repo, "A" * 64)
205 assert result.exit_code == ExitCode.USER_ERROR
206
207 def test_invalid_object_id_non_hex(self, tmp_path: pathlib.Path) -> None:
208 repo = _make_repo(tmp_path)
209 result = _cat(repo, "z" * 64)
210 assert result.exit_code == ExitCode.USER_ERROR
211
212 def test_invalid_format_errors(self, tmp_path: pathlib.Path) -> None:
213 repo = _make_repo(tmp_path)
214 result = _cat(repo, "--format", "xml", "a" * 64)
215 assert result.exit_code == ExitCode.USER_ERROR
216
217 def test_no_repo_errors(self, tmp_path: pathlib.Path) -> None:
218 from muse.cli.app import main as cli
219 result = runner.invoke(
220 cli,
221 ["cat-object", "a" * 64],
222 env={"MUSE_REPO_ROOT": str(tmp_path / "no_repo")},
223 )
224 assert result.exit_code != 0
225
226
227 # ---------------------------------------------------------------------------
228 # Security
229 # ---------------------------------------------------------------------------
230
231
232 class TestSecurity:
233 def test_ansi_in_invalid_id_not_in_output(self, tmp_path: pathlib.Path) -> None:
234 """Crafted object_id with ANSI escapes must not reach output."""
235 repo = _make_repo(tmp_path)
236 # validate_object_id rejects non-hex, so ANSI never reaches print.
237 # Confirm the error is clean.
238 evil = "\x1b[31m" + "a" * 60 # too short + has escape
239 result = _cat(repo, evil)
240 assert result.exit_code == ExitCode.USER_ERROR
241 assert "\x1b" not in result.output
242
243 def test_path_traversal_in_object_id_rejected(self, tmp_path: pathlib.Path) -> None:
244 """../../../etc/passwd style object IDs must be rejected by validate_object_id."""
245 repo = _make_repo(tmp_path)
246 result = _cat(repo, "../../../etc/passwd")
247 assert result.exit_code == ExitCode.USER_ERROR
248
249 def test_null_byte_in_object_id_rejected(self, tmp_path: pathlib.Path) -> None:
250 repo = _make_repo(tmp_path)
251 result = _cat(repo, "a" * 32 + "\x00" + "b" * 31)
252 assert result.exit_code == ExitCode.USER_ERROR
253
254 def test_no_traceback_on_invalid_id(self, tmp_path: pathlib.Path) -> None:
255 repo = _make_repo(tmp_path)
256 result = _cat(repo, "not-a-valid-id")
257 assert "Traceback" not in result.output
258
259
260 # ---------------------------------------------------------------------------
261 # Stress
262 # ---------------------------------------------------------------------------
263
264
265 class TestStress:
266 def test_large_object_streams_without_oom(self, tmp_path: pathlib.Path) -> None:
267 """A 10 MiB object must stream out without memory spike."""
268 repo = _make_repo(tmp_path)
269 content = b"Z" * (10 * 1024 * 1024) # 10 MiB
270 oid = _store(repo, content)
271 result = _cat(repo, oid)
272 assert result.exit_code == 0
273 assert len(result.stdout_bytes) == len(content)
274 assert result.stdout_bytes == content
275
276 def test_large_object_info_is_fast(self, tmp_path: pathlib.Path) -> None:
277 """Info format on a 10 MiB object reads only stat(), not content."""
278 repo = _make_repo(tmp_path)
279 content = b"Y" * (10 * 1024 * 1024)
280 oid = _store(repo, content)
281 result = _cat(repo, "--json", oid)
282 assert result.exit_code == 0
283 data = json.loads(result.output)
284 assert data["size_bytes"] == len(content)
285
286 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
287 """200 rapid cat-object calls return consistent content."""
288 repo = _make_repo(tmp_path)
289 content = b"repeated read"
290 oid = _store(repo, content)
291 for i in range(200):
292 result = _cat(repo, oid)
293 assert result.exit_code == 0, f"failed at iteration {i}"
294 assert result.stdout_bytes == content
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago