gabriel / muse public
test_cmd_ls_files.py python
371 lines 13.7 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Comprehensive tests for ``muse ls-files``.
2
3 Coverage tiers
4 --------------
5 - Integration: JSON/text format, --commit, --path-prefix, empty manifest
6 - Security: ANSI in file path stripped in text mode, JSON mode safe
7 - Stress: 1 000-file manifest, 200 sequential calls
8 """
9 from __future__ import annotations
10
11 type _FileStore = dict[str, bytes]
12
13 import datetime
14 import json
15 import pathlib
16
17 from muse.core.errors import ExitCode
18 from muse.core.object_store import write_object
19 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
20 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
21 from muse.core.types import Manifest, blob_id, long_id, split_id
22 from muse.core.paths import muse_dir, ref_path
23 from tests.cli_test_helper import CliRunner, InvokeResult
24
25 runner = CliRunner()
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
33 repo = tmp_path / "repo"
34 dot_muse = muse_dir(repo)
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (dot_muse / sub).mkdir(parents=True)
37 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
38 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"}))
39 return repo
40
41
42 def _oid(content: bytes) -> str:
43 return blob_id(content)
44
45
46 def _add_commit(
47 repo: pathlib.Path,
48 manifest: _FileStore,
49 *,
50 commit_suffix: str = "a",
51 branch: str = "main",
52 set_head: bool = True,
53 ) -> str:
54 """Store objects, snapshot, and commit; return commit_id."""
55 stored: Manifest = {}
56 for path, content in manifest.items():
57 oid = _oid(content)
58 write_object(repo, oid, content)
59 stored[path] = oid
60
61 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
62 snap_id = compute_snapshot_id(stored)
63 write_snapshot(repo, SnapshotRecord(
64 snapshot_id=snap_id,
65 manifest=stored,
66 created_at=committed_at,
67 ))
68 commit_id = compute_commit_id( parent_ids=[],
69 snapshot_id=snap_id,
70 message="test",
71 committed_at_iso=committed_at.isoformat(),
72 author="tester",
73 )
74 write_commit(repo, CommitRecord(
75 commit_id=commit_id,
76 repo_id="test-repo",
77 branch=branch,
78 snapshot_id=snap_id,
79 message="test",
80 committed_at=committed_at,
81 author="tester",
82 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(
94 cli,
95 ["ls-files", *args],
96 env={"MUSE_REPO_ROOT": str(repo)},
97 )
98
99
100 def _lsj(repo: pathlib.Path, *args: str) -> InvokeResult:
101 """Like _ls but always passes --json."""
102 return _ls(repo, "--json", *args)
103
104
105 # ---------------------------------------------------------------------------
106 # Integration — JSON format
107 # ---------------------------------------------------------------------------
108
109
110 class TestJsonFormat:
111 def test_lists_files(self, tmp_path: pathlib.Path) -> None:
112 repo = _make_repo(tmp_path)
113 cid = _add_commit(repo, {"src/main.py": b"# main", "README.md": b"# readme"})
114 result = _lsj(repo)
115 assert result.exit_code == 0
116 data = json.loads(result.output)
117 assert data["file_count"] == 2
118 paths = [f["path"] for f in data["files"]]
119 assert "src/main.py" in paths
120 assert "README.md" in paths
121
122 def test_files_sorted_alphabetically(self, tmp_path: pathlib.Path) -> None:
123 repo = _make_repo(tmp_path)
124 _add_commit(repo, {"z.py": b"z", "a.py": b"a", "m.py": b"m"})
125 data = json.loads(_lsj(repo).output)
126 paths = [f["path"] for f in data["files"]]
127 assert paths == sorted(paths)
128
129 def test_json_has_commit_and_snapshot_id(self, tmp_path: pathlib.Path) -> None:
130 repo = _make_repo(tmp_path)
131 cid = _add_commit(repo, {"f.py": b"x"})
132 data = json.loads(_lsj(repo).output)
133 assert data["commit_id"] == cid
134 assert data["snapshot_id"].startswith("sha256:")
135
136 def test_empty_manifest(self, tmp_path: pathlib.Path) -> None:
137 repo = _make_repo(tmp_path)
138 _add_commit(repo, {})
139 data = json.loads(_lsj(repo).output)
140 assert data["file_count"] == 0
141 assert data["files"] == []
142
143 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
144 repo = _make_repo(tmp_path)
145 _add_commit(repo, {"f.py": b"x"})
146 result = _lsj(repo)
147 assert result.exit_code == 0
148 data = json.loads(result.output)
149 assert data["file_count"] == 1
150
151 def test_object_ids_are_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
152 repo = _make_repo(tmp_path)
153 _add_commit(repo, {"a.py": b"content"})
154 data = json.loads(_lsj(repo).output)
155 for f in data["files"]:
156 assert f["object_id"].startswith("sha256:")
157 _, hex_part = split_id(f["object_id"])
158 assert len(hex_part) == 64
159 assert all(c in "0123456789abcdef" for c in hex_part)
160
161
162 # ---------------------------------------------------------------------------
163 # Integration — text format
164 # ---------------------------------------------------------------------------
165
166
167 class TestTextFormat:
168 def test_text_tab_separated(self, tmp_path: pathlib.Path) -> None:
169 repo = _make_repo(tmp_path)
170 _add_commit(repo, {"hello.py": b"hi"})
171 # Default (no --json) emits text: <oid>\t<path> per line
172 result = _ls(repo)
173 assert result.exit_code == 0
174 line = result.output.strip()
175 parts = line.split("\t")
176 assert len(parts) == 2
177 assert parts[0].startswith("sha256:") # canonical object_id
178 assert parts[1] == "hello.py"
179
180 def test_text_oid_matches_json_oid(self, tmp_path: pathlib.Path) -> None:
181 repo = _make_repo(tmp_path)
182 _add_commit(repo, {"check.py": b"content"})
183 json_data = json.loads(_lsj(repo).output)
184 text_out = _ls(repo).output.strip()
185 json_oid = json_data["files"][0]["object_id"]
186 text_oid = text_out.split("\t")[0]
187 assert json_oid == text_oid
188
189
190 # ---------------------------------------------------------------------------
191 # Integration — --commit flag
192 # ---------------------------------------------------------------------------
193
194
195 class TestCommitFlag:
196 def test_explicit_commit_resolves(self, tmp_path: pathlib.Path) -> None:
197 repo = _make_repo(tmp_path)
198 cid = _add_commit(repo, {"explicit.py": b"content"})
199 result = _lsj(repo, "--commit", cid)
200 assert result.exit_code == 0
201 data = json.loads(result.output)
202 assert data["commit_id"] == cid
203
204 def test_invalid_commit_id_errors(self, tmp_path: pathlib.Path) -> None:
205 repo = _make_repo(tmp_path)
206 result = _ls(repo, "--commit", "not-a-valid-id")
207 assert result.exit_code == ExitCode.USER_ERROR
208
209 def test_nonexistent_commit_id_errors(self, tmp_path: pathlib.Path) -> None:
210 repo = _make_repo(tmp_path)
211 result = _ls(repo, "--commit", long_id("f" * 64))
212 assert result.exit_code == ExitCode.USER_ERROR
213
214 def test_no_commits_on_branch_errors(self, tmp_path: pathlib.Path) -> None:
215 repo = _make_repo(tmp_path)
216 result = _ls(repo)
217 assert result.exit_code == ExitCode.USER_ERROR
218
219
220 # ---------------------------------------------------------------------------
221 # Integration — --path-prefix filter
222 # ---------------------------------------------------------------------------
223
224
225 class TestPathPrefix:
226 def test_prefix_filters_to_subtree(self, tmp_path: pathlib.Path) -> None:
227 repo = _make_repo(tmp_path)
228 _add_commit(repo, {
229 "src/main.py": b"main",
230 "src/utils.py": b"utils",
231 "tests/test_main.py": b"test",
232 "README.md": b"readme",
233 })
234 data = json.loads(_lsj(repo, "--path-prefix", "src/").output)
235 paths = [f["path"] for f in data["files"]]
236 assert all(p.startswith("src/") for p in paths)
237 assert len(paths) == 2
238
239 def test_prefix_file_count_reflects_filter(self, tmp_path: pathlib.Path) -> None:
240 repo = _make_repo(tmp_path)
241 _add_commit(repo, {
242 "a/x.py": b"x",
243 "a/y.py": b"y",
244 "b/z.py": b"z",
245 })
246 data = json.loads(_lsj(repo, "--path-prefix", "a/").output)
247 assert data["file_count"] == 2
248
249 def test_prefix_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
250 repo = _make_repo(tmp_path)
251 _add_commit(repo, {"src/main.py": b"main"})
252 data = json.loads(_lsj(repo, "--path-prefix", "tests/").output)
253 assert data["file_count"] == 0
254 assert data["files"] == []
255
256 def test_prefix_text_format(self, tmp_path: pathlib.Path) -> None:
257 repo = _make_repo(tmp_path)
258 _add_commit(repo, {"src/a.py": b"a", "tests/b.py": b"b"})
259 # Default (no --json) with --path-prefix emits text lines
260 result = _ls(repo, "--path-prefix", "src/")
261 assert result.exit_code == 0
262 lines = [l for l in result.output.strip().splitlines() if l]
263 assert len(lines) == 1
264 assert "src/a.py" in lines[0]
265
266
267 # ---------------------------------------------------------------------------
268 # Security
269 # ---------------------------------------------------------------------------
270
271
272 class TestSecurity:
273 def test_ansi_in_path_stripped_in_text_mode(self, tmp_path: pathlib.Path) -> None:
274 """File path with ANSI escape must be sanitized in text mode."""
275 repo = _make_repo(tmp_path)
276 malicious_path = "src/\x1b[31mmalicious\x1b[0m.py"
277 _add_commit(repo, {malicious_path: b"content"})
278 # Default (no --json) emits sanitized text
279 result = _ls(repo)
280 assert result.exit_code == 0
281 assert "\x1b" not in result.output
282
283 def test_ansi_in_path_preserved_in_json(self, tmp_path: pathlib.Path) -> None:
284 """JSON mode encodes ANSI as \\u001b — never emits raw escape sequences."""
285 repo = _make_repo(tmp_path)
286 malicious_path = "src/\x1b[31mmalicious\x1b[0m.py"
287 _add_commit(repo, {malicious_path: b"content"})
288 result = _lsj(repo)
289 assert result.exit_code == 0
290 # No raw ANSI bytes in stdout — json.dumps encodes \x1b as \u001b
291 assert "\x1b" not in result.output
292 data = json.loads(result.output)
293 # The path is preserved in the JSON payload (as \u001b-encoded)
294 paths = [f["path"] for f in data["files"]]
295 assert any("\x1b" in p or "\u001b" in p for p in paths)
296
297 def test_path_traversal_commit_id_rejected(self, tmp_path: pathlib.Path) -> None:
298 repo = _make_repo(tmp_path)
299 result = _ls(repo, "--commit", "../../../etc/passwd")
300 assert result.exit_code == ExitCode.USER_ERROR
301
302 def test_no_traceback_on_invalid_input(self, tmp_path: pathlib.Path) -> None:
303 repo = _make_repo(tmp_path)
304 result = _ls(repo, "--commit", "bad!")
305 assert "Traceback" not in result.output
306
307
308 # ---------------------------------------------------------------------------
309 # Stress
310 # ---------------------------------------------------------------------------
311
312
313 class TestStress:
314 def test_1000_file_manifest(self, tmp_path: pathlib.Path) -> None:
315 """1 000-file manifest lists and returns in reasonable time."""
316 repo = _make_repo(tmp_path)
317 manifest = {f"src/file_{i:04d}.py": f"content {i}".encode() for i in range(1000)}
318 _add_commit(repo, manifest)
319 result = _lsj(repo)
320 assert result.exit_code == 0
321 data = json.loads(result.output)
322 assert data["file_count"] == 1000
323
324 def test_1000_file_prefix_filter(self, tmp_path: pathlib.Path) -> None:
325 repo = _make_repo(tmp_path)
326 manifest = {f"a/file_{i:04d}.py": b"a" for i in range(500)}
327 manifest.update({f"b/file_{i:04d}.py": b"b" for i in range(500)})
328 _add_commit(repo, manifest)
329 data = json.loads(_lsj(repo, "--path-prefix", "a/").output)
330 assert data["file_count"] == 500
331
332 def test_200_sequential_calls(self, tmp_path: pathlib.Path) -> None:
333 repo = _make_repo(tmp_path)
334 _add_commit(repo, {"stable.py": b"content"})
335 for i in range(200):
336 result = _lsj(repo)
337 assert result.exit_code == 0, f"failed at iteration {i}"
338 assert json.loads(result.output)["file_count"] == 1
339
340
341 class TestRegisterFlags:
342 def test_json_short_flag(self) -> None:
343 import argparse
344 from muse.cli.commands.ls_files import register
345 p = argparse.ArgumentParser()
346 subs = p.add_subparsers()
347 register(subs)
348 args = p.parse_args(["ls-files", "-j"])
349 assert args.json_out is True
350
351 def test_json_long_flag(self) -> None:
352 import argparse
353 from muse.cli.commands.ls_files import register
354 p = argparse.ArgumentParser()
355 subs = p.add_subparsers()
356 register(subs)
357 args = p.parse_args(["ls-files", "--json"])
358 assert args.json_out is True
359
360 def test_default_no_json(self) -> None:
361 import argparse
362 from muse.cli.commands.ls_files import register
363 p = argparse.ArgumentParser()
364 subs = p.add_subparsers()
365 register(subs)
366 # Command-specific required args may differ; just check dest exists when possible
367 try:
368 args = p.parse_args(["ls-files"])
369 assert args.json_out is False
370 except SystemExit:
371 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 ⚠ 121 days ago