gabriel / muse public
test_ls_files_supercharge.py python
452 lines 17.5 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Supercharge tests for ``muse ls-files``.
2
3 Coverage tiers
4 --------------
5 I JSON envelope schema — status, error, branch, path_prefix, duration_ms, exit_code
6 II Error payload shape — consistent {status, error, exit_code}; no prose in JSON mode
7 III branch field — reflects the branch HEAD resolved to
8 IV path_prefix echoed — agents can verify which filter was applied
9 V TypedDicts — _LsFilesJson and _LsFilesErrorJson exist with correct annotations
10 VI Docstring — documents all envelope fields
11 VII Data integrity — object_ids are sha256:-prefixed in all output modes
12 VIII No prose pollution in JSON mode
13 """
14 from __future__ import annotations
15 from collections.abc import Mapping
16
17 import datetime
18 import json
19 import pathlib
20
21 import pytest
22
23 from muse.core.object_store import write_object
24 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
25 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
26 from muse.core.types import Manifest, blob_id, long_id
27 from muse.core.paths import ref_path, muse_dir
28 from tests.cli_test_helper import CliRunner, InvokeResult
29
30 runner = CliRunner()
31
32 _ENVELOPE_KEYS = {"duration_ms", "exit_code", "muse_version", "schema", "timestamp", "warnings"}
33 _REQUIRED_SUCCESS_KEYS = {
34 "status", "error", "commit_id", "snapshot_id", "branch",
35 "path_prefix", "file_count", "files",
36 } | _ENVELOPE_KEYS
37 _REQUIRED_ERROR_KEYS = {"status", "error"} | _ENVELOPE_KEYS
38
39 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
40
41
42 # ---------------------------------------------------------------------------
43 # Helpers
44 # ---------------------------------------------------------------------------
45
46 def _oid(content: bytes) -> str:
47 return blob_id(content)
48
49
50 def _make_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path:
51 repo = tmp_path / "repo"
52 dot_muse = muse_dir(repo)
53 for sub in ("objects", "commits", "snapshots", "refs/heads"):
54 (dot_muse / sub).mkdir(parents=True)
55 (dot_muse / "HEAD").write_text(f"ref: refs/heads/{branch}")
56 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"}))
57 return repo
58
59
60 def _add_commit(
61 repo: pathlib.Path,
62 files: dict[str, bytes],
63 *,
64 branch: str = "main",
65 set_head: bool = True,
66 ) -> str:
67 stored: Manifest = {}
68 for path, content in files.items():
69 oid = _oid(content)
70 write_object(repo, oid, content)
71 stored[path] = oid
72 snap_id = compute_snapshot_id(stored)
73 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=stored, created_at=_TS))
74 commit_id = compute_commit_id( parent_ids=[],
75 snapshot_id=snap_id,
76 message="test",
77 committed_at_iso=_TS.isoformat(),
78 author="tester",)
79 write_commit(repo, CommitRecord(
80 commit_id=commit_id, repo_id="test", branch=branch,
81 snapshot_id=snap_id, message="test", committed_at=_TS,
82 author="tester", parent_commit_id=None,
83 ))
84 if set_head:
85 ref = ref_path(repo, branch)
86 ref.parent.mkdir(parents=True, exist_ok=True)
87 ref.write_text(commit_id)
88 return commit_id
89
90
91 def _ls(repo: pathlib.Path, *args: str) -> InvokeResult:
92 from muse.cli.app import main as cli
93 return runner.invoke(cli, ["ls-files", *args], env={"MUSE_REPO_ROOT": str(repo)})
94
95
96 def _ls_json(repo: pathlib.Path, *args: str) -> Mapping[str, object]:
97 result = _ls(repo, "--json", *args)
98 assert result.exit_code == 0, f"ls-files --json failed:\n{result.output}"
99 return json.loads(result.output.strip())
100
101
102 # ---------------------------------------------------------------------------
103 # I JSON envelope schema
104 # ---------------------------------------------------------------------------
105
106
107 class TestJsonEnvelopeSchema:
108 def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None:
109 repo = _make_repo(tmp_path)
110 _add_commit(repo, {"a.py": b"a"})
111 data = _ls_json(repo)
112 missing = _REQUIRED_SUCCESS_KEYS - set(data.keys())
113 assert not missing, f"Missing envelope keys: {missing}"
114
115 def test_no_extra_undocumented_keys(self, tmp_path: pathlib.Path) -> None:
116 repo = _make_repo(tmp_path)
117 _add_commit(repo, {"a.py": b"a"})
118 data = _ls_json(repo)
119 extra = set(data.keys()) - _REQUIRED_SUCCESS_KEYS
120 assert not extra, f"Undocumented extra keys: {extra}"
121
122 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
123 repo = _make_repo(tmp_path)
124 _add_commit(repo, {"a.py": b"a"})
125 data = _ls_json(repo)
126 assert data["status"] == "ok"
127
128 def test_error_empty_string_on_success(self, tmp_path: pathlib.Path) -> None:
129 repo = _make_repo(tmp_path)
130 _add_commit(repo, {"a.py": b"a"})
131 data = _ls_json(repo)
132 assert data["error"] == ""
133
134 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
135 repo = _make_repo(tmp_path)
136 _add_commit(repo, {"a.py": b"a"})
137 data = _ls_json(repo)
138 assert data["exit_code"] == 0
139
140 def test_duration_ms_nonnegative_float(self, tmp_path: pathlib.Path) -> None:
141 repo = _make_repo(tmp_path)
142 _add_commit(repo, {"a.py": b"a"})
143 data = _ls_json(repo)
144 assert isinstance(data["duration_ms"], float)
145 assert data["duration_ms"] >= 0.0
146
147 def test_file_count_matches_files_length(self, tmp_path: pathlib.Path) -> None:
148 repo = _make_repo(tmp_path)
149 _add_commit(repo, {"a.py": b"a", "b.py": b"b", "c.py": b"c"})
150 data = _ls_json(repo)
151 assert data["file_count"] == len(data["files"])
152
153 def test_files_is_list(self, tmp_path: pathlib.Path) -> None:
154 repo = _make_repo(tmp_path)
155 _add_commit(repo, {"a.py": b"a"})
156 data = _ls_json(repo)
157 assert isinstance(data["files"], list)
158
159 def test_path_prefix_none_when_not_filtered(self, tmp_path: pathlib.Path) -> None:
160 repo = _make_repo(tmp_path)
161 _add_commit(repo, {"a.py": b"a"})
162 data = _ls_json(repo)
163 assert data["path_prefix"] is None
164
165 def test_path_prefix_echoed_when_filtered(self, tmp_path: pathlib.Path) -> None:
166 repo = _make_repo(tmp_path)
167 _add_commit(repo, {"src/a.py": b"a", "tests/b.py": b"b"})
168 data = _ls_json(repo, "--path-prefix", "src/")
169 assert data["path_prefix"] == "src/"
170
171 def test_commit_id_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
172 repo = _make_repo(tmp_path)
173 _add_commit(repo, {"a.py": b"a"})
174 data = _ls_json(repo)
175 assert data["commit_id"].startswith("sha256:")
176
177 def test_snapshot_id_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 _add_commit(repo, {"a.py": b"a"})
180 data = _ls_json(repo)
181 assert data["snapshot_id"].startswith("sha256:")
182
183
184 # ---------------------------------------------------------------------------
185 # II Error payload shape
186 # ---------------------------------------------------------------------------
187
188
189 class TestErrorPayloadShape:
190 def test_error_keys_present(self, tmp_path: pathlib.Path) -> None:
191 repo = _make_repo(tmp_path) # no commits
192 result = _ls(repo, "--json")
193 assert result.exit_code != 0
194 data = json.loads(result.output.strip())
195 assert _REQUIRED_ERROR_KEYS.issubset(set(data.keys()))
196
197 def test_error_status_is_error(self, tmp_path: pathlib.Path) -> None:
198 repo = _make_repo(tmp_path)
199 result = _ls(repo, "--json")
200 assert result.exit_code != 0
201 data = json.loads(result.output.strip())
202 assert data["status"] == "error"
203
204 def test_error_message_nonempty(self, tmp_path: pathlib.Path) -> None:
205 repo = _make_repo(tmp_path)
206 result = _ls(repo, "--json")
207 data = json.loads(result.output.strip())
208 assert isinstance(data["error"], str) and len(data["error"]) > 0
209
210 def test_error_exit_code_nonzero(self, tmp_path: pathlib.Path) -> None:
211 repo = _make_repo(tmp_path)
212 result = _ls(repo, "--json")
213 data = json.loads(result.output.strip())
214 assert isinstance(data["exit_code"], int) and data["exit_code"] != 0
215
216 def test_invalid_commit_error_payload(self, tmp_path: pathlib.Path) -> None:
217 repo = _make_repo(tmp_path)
218 result = _ls(repo, "--json", "--commit", "not-valid")
219 assert result.exit_code != 0
220 data = json.loads(result.output.strip())
221 assert data["status"] == "error"
222 assert "exit_code" in data
223
224 def test_nonexistent_commit_error_payload(self, tmp_path: pathlib.Path) -> None:
225 repo = _make_repo(tmp_path)
226 result = _ls(repo, "--json", "--commit", long_id("f" * 64))
227 assert result.exit_code != 0
228 data = json.loads(result.output.strip())
229 assert data["status"] == "error"
230
231
232 # ---------------------------------------------------------------------------
233 # III branch field
234 # ---------------------------------------------------------------------------
235
236
237 class TestBranchField:
238 def test_branch_is_main_when_on_main(self, tmp_path: pathlib.Path) -> None:
239 repo = _make_repo(tmp_path, branch="main")
240 _add_commit(repo, {"a.py": b"a"}, branch="main")
241 data = _ls_json(repo)
242 assert data["branch"] == "main"
243
244 def test_branch_is_dev_when_on_dev(self, tmp_path: pathlib.Path) -> None:
245 repo = _make_repo(tmp_path, branch="dev")
246 _add_commit(repo, {"a.py": b"a"}, branch="dev")
247 data = _ls_json(repo)
248 assert data["branch"] == "dev"
249
250 def test_branch_is_none_when_explicit_commit_given(
251 self, tmp_path: pathlib.Path
252 ) -> None:
253 """When --commit is given explicitly, no branch resolution occurs."""
254 repo = _make_repo(tmp_path)
255 cid = _add_commit(repo, {"a.py": b"a"})
256 data = _ls_json(repo, "--commit", cid)
257 # branch is null when commit was specified directly, not via HEAD
258 assert data["branch"] is None
259
260
261 # ---------------------------------------------------------------------------
262 # IV path_prefix echoed
263 # ---------------------------------------------------------------------------
264
265
266 class TestPathPrefixEchoed:
267 def test_path_prefix_null_without_filter(self, tmp_path: pathlib.Path) -> None:
268 repo = _make_repo(tmp_path)
269 _add_commit(repo, {"a.py": b"a"})
270 data = _ls_json(repo)
271 assert data["path_prefix"] is None
272
273 def test_path_prefix_echoed_src(self, tmp_path: pathlib.Path) -> None:
274 repo = _make_repo(tmp_path)
275 _add_commit(repo, {"src/a.py": b"a"})
276 data = _ls_json(repo, "--path-prefix", "src/")
277 assert data["path_prefix"] == "src/"
278
279 def test_path_prefix_echoed_nested(self, tmp_path: pathlib.Path) -> None:
280 repo = _make_repo(tmp_path)
281 _add_commit(repo, {"a/b/c.py": b"c"})
282 data = _ls_json(repo, "--path-prefix", "a/b/")
283 assert data["path_prefix"] == "a/b/"
284
285
286 # ---------------------------------------------------------------------------
287 # V TypedDicts
288 # ---------------------------------------------------------------------------
289
290
291 class TestTypedDicts:
292 def test_ls_files_json_typed_dict_exists(self) -> None:
293 from muse.cli.commands.ls_files import _LsFilesJson # type: ignore[attr-defined]
294 assert _LsFilesJson is not None
295
296 def test_ls_files_error_json_typed_dict_exists(self) -> None:
297 from muse.cli.commands.ls_files import _LsFilesErrorJson # type: ignore[attr-defined]
298 assert _LsFilesErrorJson is not None
299
300 def test_ls_files_json_has_all_annotations(self) -> None:
301 from muse.cli.commands.ls_files import _LsFilesJson # type: ignore[attr-defined]
302 hints = _LsFilesJson.__annotations__
303 required = {"status", "error", "commit_id", "snapshot_id", "branch",
304 "path_prefix", "file_count", "files", "duration_ms", "exit_code"}
305 assert not (required - set(hints)), f"Missing: {required - set(hints)}"
306
307 def test_ls_files_error_json_has_all_annotations(self) -> None:
308 from muse.cli.commands.ls_files import _LsFilesErrorJson # type: ignore[attr-defined]
309 hints = _LsFilesErrorJson.__annotations__
310 assert not ({"status", "error", "exit_code"} - set(hints))
311
312
313 # ---------------------------------------------------------------------------
314 # VI Docstring
315 # ---------------------------------------------------------------------------
316
317
318 class TestDocstring:
319 def test_docstring_documents_status(self) -> None:
320 import muse.cli.commands.ls_files as m
321 assert '"status"' in (m.__doc__ or "")
322
323 def test_docstring_documents_branch(self) -> None:
324 import muse.cli.commands.ls_files as m
325 assert '"branch"' in (m.__doc__ or "")
326
327 def test_docstring_documents_path_prefix(self) -> None:
328 import muse.cli.commands.ls_files as m
329 assert '"path_prefix"' in (m.__doc__ or "")
330
331 def test_docstring_documents_duration_ms(self) -> None:
332 import muse.cli.commands.ls_files as m
333 assert "duration_ms" in (m.__doc__ or "")
334
335 def test_docstring_documents_exit_code(self) -> None:
336 import muse.cli.commands.ls_files as m
337 assert "exit_code" in (m.__doc__ or "")
338
339 def test_docstring_documents_error(self) -> None:
340 import muse.cli.commands.ls_files as m
341 assert '"error"' in (m.__doc__ or "")
342
343
344 # ---------------------------------------------------------------------------
345 # VII Data integrity
346 # ---------------------------------------------------------------------------
347
348
349 class TestDataIntegrity:
350 def test_object_ids_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None:
351 repo = _make_repo(tmp_path)
352 _add_commit(repo, {"a.py": b"content", "b.py": b"more"})
353 data = _ls_json(repo)
354 for f in data["files"]:
355 assert f["object_id"].startswith("sha256:"), (
356 f"object_id not sha256:-prefixed: {f['object_id']!r}"
357 )
358
359 def test_object_ids_sha256_prefixed_in_text(self, tmp_path: pathlib.Path) -> None:
360 repo = _make_repo(tmp_path)
361 _add_commit(repo, {"a.py": b"content"})
362 # Default (no --json) emits text: <oid>\t<path> per line
363 result = _ls(repo)
364 assert result.exit_code == 0
365 for line in result.output.strip().splitlines():
366 oid = line.split("\t")[0]
367 assert oid.startswith("sha256:"), f"text OID not prefixed: {oid!r}"
368
369 def test_commit_id_matches_stored(self, tmp_path: pathlib.Path) -> None:
370 repo = _make_repo(tmp_path)
371 cid = _add_commit(repo, {"a.py": b"a"})
372 data = _ls_json(repo)
373 assert data["commit_id"] == cid
374
375 def test_files_sorted_alphabetically(self, tmp_path: pathlib.Path) -> None:
376 repo = _make_repo(tmp_path)
377 _add_commit(repo, {"z.py": b"z", "a.py": b"a", "m.py": b"m"})
378 data = _ls_json(repo)
379 paths = [f["path"] for f in data["files"]]
380 assert paths == sorted(paths)
381
382 def test_path_prefix_file_count_consistent(self, tmp_path: pathlib.Path) -> None:
383 repo = _make_repo(tmp_path)
384 _add_commit(repo, {"src/a.py": b"a", "src/b.py": b"b", "tests/c.py": b"c"})
385 data = _ls_json(repo, "--path-prefix", "src/")
386 assert data["file_count"] == len(data["files"]) == 2
387
388
389 # ---------------------------------------------------------------------------
390 # VIII No prose pollution in JSON mode
391 # ---------------------------------------------------------------------------
392
393
394 class TestNoProsePollution:
395 def test_success_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
396 repo = _make_repo(tmp_path)
397 _add_commit(repo, {"a.py": b"a"})
398 result = _ls(repo, "--json")
399 json.loads(result.output.strip()) # must not raise
400
401 def test_no_emoji_in_json_success_output(self, tmp_path: pathlib.Path) -> None:
402 repo = _make_repo(tmp_path)
403 _add_commit(repo, {"a.py": b"a"})
404 result = _ls(repo, "--json")
405 assert "❌" not in result.output
406 assert "✅" not in result.output
407
408 def test_no_emoji_in_json_error_output(self, tmp_path: pathlib.Path) -> None:
409 """Errors in --json mode must not emit prose emoji to stdout."""
410 repo = _make_repo(tmp_path) # no commits
411 result = _ls(repo, "--json")
412 assert "❌" not in result.output
413 data = json.loads(result.output.strip())
414 assert data["status"] == "error"
415
416 def test_error_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
417 repo = _make_repo(tmp_path)
418 result = _ls(repo, "--json")
419 json.loads(result.output.strip()) # must not raise
420
421
422 class TestRegisterFlags:
423 def test_json_short_flag(self) -> None:
424 import argparse
425 from muse.cli.commands.ls_files import register
426 p = argparse.ArgumentParser()
427 subs = p.add_subparsers()
428 register(subs)
429 args = p.parse_args(["ls-files", "-j"])
430 assert args.json_out is True
431
432 def test_json_long_flag(self) -> None:
433 import argparse
434 from muse.cli.commands.ls_files import register
435 p = argparse.ArgumentParser()
436 subs = p.add_subparsers()
437 register(subs)
438 args = p.parse_args(["ls-files", "--json"])
439 assert args.json_out is True
440
441 def test_default_no_json(self) -> None:
442 import argparse
443 from muse.cli.commands.ls_files import register
444 p = argparse.ArgumentParser()
445 subs = p.add_subparsers()
446 register(subs)
447 # Command-specific required args may differ; just check dest exists when possible
448 try:
449 args = p.parse_args(["ls-files"])
450 assert args.json_out is False
451 except SystemExit:
452 pass # required positional args missing — flag default still correct
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago