gabriel / muse public
test_ls_tree_supercharge.py python
439 lines 16.6 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 125 days ago
1 """Supercharge tests for ``muse ls-tree``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope schema: status, error, exit_code, duration_ms, entry_count,
6 path_prefix, recursive always present
7 - Error payload shape: exactly {status, error, exit_code} — no prose in --json mode
8 - OID integrity: blob object_ids sha256:-prefixed; synthetic tree object_ids sha256:-prefixed
9 - TypedDicts: _LsTreeJson and _LsTreeErrorJson exist and are annotated
10 - Docstring: module docstring covers all new envelope fields and error schema
11 - No-prose pollution: no emoji in JSON stdout, errors to stdout in --json mode
12 """
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 import datetime
17 import argparse
18 import json
19 import pathlib
20 from typing import get_type_hints
21
22 import pytest
23
24 from muse.core.errors import ExitCode
25 from muse.core.object_store import write_object
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
28 from muse.core.types import Manifest, blob_id, split_id
29 from muse.core.paths import ref_path, muse_dir
30 from tests.cli_test_helper import CliRunner, InvokeResult
31
32 runner = CliRunner()
33
34 _REPO_ID = "ls-tree-sg-test"
35 _counter = 0
36
37
38 # ---------------------------------------------------------------------------
39 # Helpers
40 # ---------------------------------------------------------------------------
41
42
43
44 def _init_repo(path: pathlib.Path) -> pathlib.Path:
45 dot_muse = muse_dir(path)
46 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
47 (dot_muse / d).mkdir(parents=True, exist_ok=True)
48 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
49 (dot_muse / "repo.json").write_text(
50 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
51 )
52 return path
53
54
55 def _commit_files(root: pathlib.Path, files: Mapping[str, bytes], branch: str = "main") -> str:
56 global _counter
57 _counter += 1
58 manifest: Manifest = {}
59 for rel_path, content in files.items():
60 obj_id = blob_id(content)
61 write_object(root, obj_id, content)
62 manifest[rel_path] = obj_id
63 abs_path = root / rel_path
64 abs_path.parent.mkdir(parents=True, exist_ok=True)
65 abs_path.write_bytes(content)
66 snap_id = compute_snapshot_id(manifest)
67 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
68 committed_at = datetime.datetime.now(datetime.timezone.utc)
69 commit_id = compute_commit_id(
70 parent_ids=[],
71 snapshot_id=snap_id,
72 message=f"commit {_counter}",
73 committed_at_iso=committed_at.isoformat(),
74 )
75 write_commit(root, CommitRecord(
76 repo_id=_REPO_ID,
77 commit_id=commit_id,
78 branch=branch,
79 snapshot_id=snap_id,
80 message=f"commit {_counter}",
81 committed_at=committed_at,
82 ))
83 (ref_path(root, branch)).write_text(commit_id, encoding="utf-8")
84 return commit_id
85
86
87 def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult:
88 from muse.cli.app import main as cli
89 return runner.invoke(cli, ["ls-tree", *args], env={"MUSE_REPO_ROOT": str(repo)})
90
91
92 # ---------------------------------------------------------------------------
93 # JSON envelope schema
94 # ---------------------------------------------------------------------------
95
96 class TestJsonEnvelopeSchema:
97 """Every required key is present in the success envelope."""
98
99 _REQUIRED = {
100 "status", "error", "treeish", "commit_id",
101 "path_prefix", "recursive", "entry_count", "entries",
102 "duration_ms", "exit_code",
103 }
104
105 def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None:
106 repo = _init_repo(tmp_path)
107 _commit_files(repo, {"a.py": b"# a\n"})
108 r = _invoke(repo, "HEAD", "--json")
109 assert r.exit_code == 0
110 d = json.loads(r.output)
111 missing = self._REQUIRED - d.keys()
112 assert not missing, f"Missing keys: {missing}"
113
114 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
115 repo = _init_repo(tmp_path)
116 _commit_files(repo, {"a.py": b"# a\n"})
117 r = _invoke(repo, "HEAD", "--json")
118 assert json.loads(r.output)["status"] == "ok"
119
120 def test_error_empty_on_success(self, tmp_path: pathlib.Path) -> None:
121 repo = _init_repo(tmp_path)
122 _commit_files(repo, {"a.py": b"# a\n"})
123 r = _invoke(repo, "HEAD", "--json")
124 assert json.loads(r.output)["error"] == ""
125
126 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
127 repo = _init_repo(tmp_path)
128 _commit_files(repo, {"a.py": b"# a\n"})
129 r = _invoke(repo, "HEAD", "--json")
130 assert json.loads(r.output)["exit_code"] == 0
131
132 def test_duration_ms_is_nonneg_float(self, tmp_path: pathlib.Path) -> None:
133 repo = _init_repo(tmp_path)
134 _commit_files(repo, {"a.py": b"# a\n"})
135 r = _invoke(repo, "HEAD", "--json")
136 d = json.loads(r.output)
137 assert isinstance(d["duration_ms"], float)
138 assert d["duration_ms"] >= 0.0
139
140 def test_entry_count_matches_entries_length(self, tmp_path: pathlib.Path) -> None:
141 repo = _init_repo(tmp_path)
142 _commit_files(repo, {"a.py": b"a", "b.py": b"b", "src/c.py": b"c"})
143 r = _invoke(repo, "HEAD", "--json")
144 d = json.loads(r.output)
145 assert d["entry_count"] == len(d["entries"])
146
147 def test_path_prefix_null_when_not_given(self, tmp_path: pathlib.Path) -> None:
148 repo = _init_repo(tmp_path)
149 _commit_files(repo, {"a.py": b"a"})
150 r = _invoke(repo, "HEAD", "--json")
151 d = json.loads(r.output)
152 assert d["path_prefix"] is None
153
154 def test_path_prefix_echoed_when_given(self, tmp_path: pathlib.Path) -> None:
155 repo = _init_repo(tmp_path)
156 _commit_files(repo, {"src/a.py": b"a"})
157 r = _invoke(repo, "HEAD", "src/", "--json")
158 d = json.loads(r.output)
159 assert d["path_prefix"] == "src/"
160
161 def test_recursive_false_by_default(self, tmp_path: pathlib.Path) -> None:
162 repo = _init_repo(tmp_path)
163 _commit_files(repo, {"src/a.py": b"a"})
164 r = _invoke(repo, "HEAD", "--json")
165 d = json.loads(r.output)
166 assert d["recursive"] is False
167
168 def test_recursive_true_when_flag_given(self, tmp_path: pathlib.Path) -> None:
169 repo = _init_repo(tmp_path)
170 _commit_files(repo, {"src/a.py": b"a"})
171 r = _invoke(repo, "-r", "HEAD", "--json")
172 d = json.loads(r.output)
173 assert d["recursive"] is True
174
175 def test_treeish_echoed(self, tmp_path: pathlib.Path) -> None:
176 repo = _init_repo(tmp_path)
177 _commit_files(repo, {"a.py": b"a"})
178 r = _invoke(repo, "HEAD", "--json")
179 d = json.loads(r.output)
180 assert d["treeish"] == "HEAD"
181
182 def test_commit_id_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
183 repo = _init_repo(tmp_path)
184 _commit_files(repo, {"a.py": b"a"})
185 r = _invoke(repo, "HEAD", "--json")
186 d = json.loads(r.output)
187 assert d["commit_id"].startswith("sha256:")
188
189
190 # ---------------------------------------------------------------------------
191 # Error payload shape
192 # ---------------------------------------------------------------------------
193
194 class TestErrorPayloadShape:
195 """In --json mode, errors go to stdout as {status, error, exit_code}."""
196
197 def test_error_on_empty_repo_is_json(self, tmp_path: pathlib.Path) -> None:
198 repo = _init_repo(tmp_path)
199 r = _invoke(repo, "HEAD", "--json")
200 assert r.exit_code != 0
201 d = json.loads(r.output) # must be valid JSON
202 assert d["status"] == "error"
203
204 def test_error_payload_has_required_keys(self, tmp_path: pathlib.Path) -> None:
205 repo = _init_repo(tmp_path)
206 r = _invoke(repo, "HEAD", "--json")
207 d = json.loads(r.output)
208 assert {"error", "exit_code"} <= set(d.keys())
209
210 def test_error_message_nonempty(self, tmp_path: pathlib.Path) -> None:
211 repo = _init_repo(tmp_path)
212 r = _invoke(repo, "HEAD", "--json")
213 d = json.loads(r.output)
214 assert d["error"]
215
216 def test_exit_code_nonzero_on_error(self, tmp_path: pathlib.Path) -> None:
217 repo = _init_repo(tmp_path)
218 r = _invoke(repo, "HEAD", "--json")
219 assert r.exit_code != 0
220 d = json.loads(r.output)
221 assert d["exit_code"] != 0
222
223 def test_ansi_in_ref_error_is_json(self, tmp_path: pathlib.Path) -> None:
224 repo = _init_repo(tmp_path)
225 _commit_files(repo, {"a.py": b"a"})
226 r = _invoke(repo, "\x1b[31mbad\x1b[0m", "--json")
227 assert r.exit_code != 0
228 d = json.loads(r.output)
229 assert d["status"] == "error"
230
231 def test_bad_ref_error_is_json(self, tmp_path: pathlib.Path) -> None:
232 repo = _init_repo(tmp_path)
233 _commit_files(repo, {"a.py": b"a"})
234 r = _invoke(repo, "no-such-branch", "--json")
235 assert r.exit_code != 0
236 d = json.loads(r.output)
237 assert d["status"] == "error"
238
239 def test_path_traversal_error_is_json(self, tmp_path: pathlib.Path) -> None:
240 repo = _init_repo(tmp_path)
241 _commit_files(repo, {"a.py": b"a"})
242 r = _invoke(repo, "HEAD", "../../../etc/", "--json")
243 assert r.exit_code != 0
244 d = json.loads(r.output)
245 assert d["status"] == "error"
246
247
248 # ---------------------------------------------------------------------------
249 # OID data integrity
250 # ---------------------------------------------------------------------------
251
252 class TestOidIntegrity:
253 """All object IDs in output carry the sha256: prefix."""
254
255 def test_blob_object_ids_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
256 repo = _init_repo(tmp_path)
257 _commit_files(repo, {"a.py": b"content"})
258 r = _invoke(repo, "-r", "HEAD", "--json")
259 d = json.loads(r.output)
260 for e in d["entries"]:
261 if e["type"] == "blob":
262 assert e["object_id"].startswith("sha256:"), (
263 f"blob OID not prefixed: {e['object_id']!r}"
264 )
265
266 def test_synthetic_tree_object_ids_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
267 repo = _init_repo(tmp_path)
268 _commit_files(repo, {"src/a.py": b"a", "lib/b.py": b"b"})
269 r = _invoke(repo, "HEAD", "--json")
270 d = json.loads(r.output)
271 for e in d["entries"]:
272 if e["type"] == "tree":
273 assert e["object_id"].startswith("sha256:"), (
274 f"tree OID not prefixed: {e['object_id']!r}"
275 )
276
277 def test_blob_oid_hex_part_is_64_chars(self, tmp_path: pathlib.Path) -> None:
278 repo = _init_repo(tmp_path)
279 _commit_files(repo, {"a.py": b"content"})
280 r = _invoke(repo, "-r", "HEAD", "--json")
281 d = json.loads(r.output)
282 for e in d["entries"]:
283 if e["type"] == "blob":
284 _, hex_part = split_id(e["object_id"])
285 assert len(hex_part) == 64
286 assert all(c in "0123456789abcdef" for c in hex_part)
287
288 def test_tree_oid_hex_part_is_64_chars(self, tmp_path: pathlib.Path) -> None:
289 repo = _init_repo(tmp_path)
290 _commit_files(repo, {"src/a.py": b"a"})
291 r = _invoke(repo, "HEAD", "--json")
292 d = json.loads(r.output)
293 for e in d["entries"]:
294 if e["type"] == "tree":
295 _, hex_part = split_id(e["object_id"])
296 assert len(hex_part) == 64
297 assert all(c in "0123456789abcdef" for c in hex_part)
298
299 def test_text_format_blob_oid_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
300 repo = _init_repo(tmp_path)
301 _commit_files(repo, {"a.py": b"content"})
302 r = _invoke(repo, "-r", "HEAD")
303 assert r.exit_code == 0
304 for line in r.output.strip().splitlines():
305 meta, _ = line.split("\t", 1)
306 parts = meta.split()
307 oid = parts[2]
308 assert oid.startswith("sha256:"), f"text OID not prefixed: {oid!r}"
309
310
311 # ---------------------------------------------------------------------------
312 # No-prose pollution
313 # ---------------------------------------------------------------------------
314
315 class TestNoProsePollution:
316 def test_stdout_valid_json_on_success(self, tmp_path: pathlib.Path) -> None:
317 repo = _init_repo(tmp_path)
318 _commit_files(repo, {"a.py": b"a"})
319 r = _invoke(repo, "HEAD", "--json")
320 json.loads(r.output) # must not raise
321
322 def test_no_emoji_in_json_stdout(self, tmp_path: pathlib.Path) -> None:
323 repo = _init_repo(tmp_path)
324 _commit_files(repo, {"a.py": b"a"})
325 r = _invoke(repo, "HEAD", "--json")
326 assert "❌" not in r.output
327
328 def test_error_stdout_valid_json(self, tmp_path: pathlib.Path) -> None:
329 repo = _init_repo(tmp_path)
330 r = _invoke(repo, "HEAD", "--json")
331 json.loads(r.output) # must not raise
332
333 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None:
334 repo = _init_repo(tmp_path)
335 _commit_files(repo, {"a.py": b"a"})
336 r = _invoke(repo, "ghost-branch", "--json")
337 assert "Traceback" not in r.output
338 assert "Traceback" not in r.stderr
339
340 def test_ansi_in_output_encoded_in_json(self, tmp_path: pathlib.Path) -> None:
341 """File paths with ANSI sequences must be JSON-encoded, not emitted raw."""
342 repo = _init_repo(tmp_path)
343 malicious = "src/\x1b[31mmalicious\x1b[0m.py"
344 _commit_files(repo, {malicious: b"bad"})
345 r = _invoke(repo, "-r", "HEAD", "--json")
346 assert r.exit_code == 0
347 assert "\x1b" not in r.output
348
349
350 # ---------------------------------------------------------------------------
351 # TypedDicts
352 # ---------------------------------------------------------------------------
353
354 class TestTypedDicts:
355 def test_ls_tree_json_typeddict_exists(self) -> None:
356 from muse.cli.commands.ls_tree import _LsTreeJson
357 assert _LsTreeJson is not None
358
359 def test_ls_tree_error_json_typeddict_exists(self) -> None:
360 from muse.cli.commands.ls_tree import _LsTreeErrorJson
361 assert _LsTreeErrorJson is not None
362
363 def test_ls_tree_json_has_status_annotation(self) -> None:
364 from muse.cli.commands.ls_tree import _LsTreeJson
365 hints = get_type_hints(_LsTreeJson)
366 assert "status" in hints
367
368 def test_ls_tree_json_has_all_new_fields(self) -> None:
369 from muse.cli.commands.ls_tree import _LsTreeJson
370 hints = get_type_hints(_LsTreeJson)
371 for field in ("status", "error", "entry_count", "path_prefix", "recursive",
372 "duration_ms", "exit_code"):
373 assert field in hints, f"Missing annotation: {field!r}"
374
375
376 # ---------------------------------------------------------------------------
377 # Docstring coverage
378 # ---------------------------------------------------------------------------
379
380 class TestDocstring:
381 def _doc(self) -> str:
382 import muse.cli.commands.ls_tree as mod
383 return mod.__doc__ or ""
384
385 def test_docstring_documents_status(self) -> None:
386 assert "status" in self._doc()
387
388 def test_docstring_documents_error(self) -> None:
389 assert "error" in self._doc()
390
391 def test_docstring_documents_entry_count(self) -> None:
392 assert "entry_count" in self._doc()
393
394 def test_docstring_documents_path_prefix(self) -> None:
395 assert "path_prefix" in self._doc()
396
397 def test_docstring_documents_duration_ms(self) -> None:
398 assert "duration_ms" in self._doc()
399
400 def test_docstring_documents_exit_code(self) -> None:
401 assert "exit_code" in self._doc()
402
403 def test_docstring_documents_error_schema(self) -> None:
404 doc = self._doc()
405 assert "error" in doc and "exit_code" in doc
406
407
408 # ---------------------------------------------------------------------------
409 # TestRegisterFlags — argparse-level verification
410 # ---------------------------------------------------------------------------
411
412
413 class TestRegisterFlags:
414 """Verify that register() wires --json / -j correctly."""
415
416 def _make_parser(self) -> "argparse.ArgumentParser":
417 import argparse
418 from muse.cli.commands.ls_tree import register
419 ap = argparse.ArgumentParser()
420 subs = ap.add_subparsers()
421 register(subs)
422 return ap
423
424 def test_json_flag_long(self) -> None:
425 ns = self._make_parser().parse_args(["ls-tree", "--json"])
426 assert ns.json_out is True
427
428 def test_j_alias(self) -> None:
429 ns = self._make_parser().parse_args(["ls-tree", "-j"])
430 assert ns.json_out is True
431
432 def test_default_is_text(self) -> None:
433 ns = self._make_parser().parse_args(["ls-tree"])
434 assert ns.json_out is False
435
436 def test_dest_is_json_out(self) -> None:
437 ns = self._make_parser().parse_args(["ls-tree", "-j"])
438 assert hasattr(ns, "json_out")
439 assert not hasattr(ns, "fmt")
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 125 days ago