gabriel / muse public
test_cmd_read_snapshot.py python
306 lines 11.0 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 read-snapshot``.
2
3 Coverage tiers
4 --------------
5 - Unit: schema keys constant
6 - Integration: JSON/text, --no-manifest, --path-prefix, manifest presence
7 - Security: ANSI in snapshot IDs rejected, no traceback on bad input
8 - Stress: 1000-path manifest, 200 sequential reads
9 """
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 from muse.core.errors import ExitCode
17 from muse.core.paths import muse_dir
18 from muse.core.snapshot import compute_snapshot_id
19 from muse.core.store import SnapshotRecord, write_snapshot
20 from tests.cli_test_helper import CliRunner, InvokeResult
21 from muse.core.types import NULL_COMMIT_ID, fake_id, long_id, short_id
22
23 runner = CliRunner()
24
25 _CREATED_AT: datetime.datetime = datetime.datetime(
26 2026, 3, 18, 12, 0, tzinfo=datetime.timezone.utc
27 )
28
29
30 # ---------------------------------------------------------------------------
31 # Helpers
32 # ---------------------------------------------------------------------------
33
34 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
35 repo = tmp_path / "repo"
36 dot_muse = muse_dir(repo)
37 for sub in ("objects", "commits", "snapshots", "refs/heads"):
38 (dot_muse / sub).mkdir(parents=True)
39 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
40 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
41 return repo
42
43
44 def _snap(
45 repo: pathlib.Path,
46 manifest: Manifest | None = None,
47 ) -> str:
48 """Write a snapshot with a real content-addressed ID; return the snapshot_id."""
49 m: Manifest = manifest or {}
50 snap_id = compute_snapshot_id(m)
51 rec = SnapshotRecord(
52 snapshot_id=snap_id,
53 manifest=m,
54 created_at=_CREATED_AT,
55 )
56 write_snapshot(repo, rec)
57 return snap_id
58
59
60 def _rs(repo: pathlib.Path, *args: str) -> InvokeResult:
61 from muse.cli.app import main as cli
62 return runner.invoke(
63 cli,
64 ["read-snapshot", *args],
65 env={"MUSE_REPO_ROOT": str(repo)},
66 )
67
68
69 def _rsj(repo: pathlib.Path, *args: str) -> InvokeResult:
70 """Like _rs but always passes --json."""
71 return _rs(repo, "--json", *args)
72
73
74 def _fake_oid(n: int) -> str:
75 return format(n, "064x")
76
77
78 # ---------------------------------------------------------------------------
79 # Integration — JSON format
80 # ---------------------------------------------------------------------------
81
82
83 class TestJsonFormat:
84 def test_full_output_empty_manifest(self, tmp_path: pathlib.Path) -> None:
85 repo = _make_repo(tmp_path)
86 sid = _snap(repo)
87 result = _rsj(repo, sid)
88 assert result.exit_code == 0
89 data = json.loads(result.output)
90 assert data["snapshot_id"] == sid
91 assert data["file_count"] == 0
92 assert data["manifest"] == {}
93
94 def test_manifest_paths_present(self, tmp_path: pathlib.Path) -> None:
95 repo = _make_repo(tmp_path)
96 oid = NULL_COMMIT_ID
97 sid = _snap(repo, {"src/main.py": oid, "tests/test_main.py": oid})
98 data = json.loads(_rsj(repo, sid).output)
99 assert "src/main.py" in data["manifest"]
100 assert "tests/test_main.py" in data["manifest"]
101 assert data["file_count"] == 2
102
103 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
104 repo = _make_repo(tmp_path)
105 sid = _snap(repo, {"a.py": _fake_oid(1)})
106 result = _rs(repo, "--json", sid)
107 assert result.exit_code == 0
108 assert "snapshot_id" in json.loads(result.output)
109
110 def test_created_at_iso8601(self, tmp_path: pathlib.Path) -> None:
111 repo = _make_repo(tmp_path)
112 sid = _snap(repo, {"b.py": _fake_oid(2)})
113 data = json.loads(_rsj(repo, sid).output)
114 datetime.datetime.fromisoformat(data["created_at"])
115
116 def test_file_count_reflects_manifest(self, tmp_path: pathlib.Path) -> None:
117 repo = _make_repo(tmp_path)
118 manifest = {f"file{i}.py": _fake_oid(i) for i in range(5)}
119 sid = _snap(repo, manifest)
120 data = json.loads(_rsj(repo, sid).output)
121 assert data["file_count"] == 5
122
123
124 # ---------------------------------------------------------------------------
125 # Integration — text format
126 # ---------------------------------------------------------------------------
127
128
129 class TestTextFormat:
130 def test_text_contains_prefix(self, tmp_path: pathlib.Path) -> None:
131 repo = _make_repo(tmp_path)
132 sid = _snap(repo, {"c.py": _fake_oid(3)})
133 result = _rs(repo, sid)
134 assert result.exit_code == 0
135 assert short_id(sid) in result.output
136
137 def test_text_contains_file_count(self, tmp_path: pathlib.Path) -> None:
138 repo = _make_repo(tmp_path)
139 sid = _snap(repo, {"a.py": _fake_oid(1), "b.py": _fake_oid(2)})
140 result = _rs(repo, sid)
141 assert "2 files" in result.output
142
143 def test_text_single_line(self, tmp_path: pathlib.Path) -> None:
144 repo = _make_repo(tmp_path)
145 sid = _snap(repo)
146 result = _rs(repo, sid)
147 lines = [l for l in result.output.splitlines() if l.strip()]
148 assert len(lines) == 1
149
150
151 # ---------------------------------------------------------------------------
152 # Integration — --no-manifest
153 # ---------------------------------------------------------------------------
154
155
156 class TestNoManifest:
157 def test_manifest_absent(self, tmp_path: pathlib.Path) -> None:
158 repo = _make_repo(tmp_path)
159 sid = _snap(repo, {"a.py": _fake_oid(1)})
160 data = json.loads(_rsj(repo, "--no-manifest", sid).output)
161 assert "manifest" not in data
162 assert data["file_count"] == 1
163
164 def test_snapshot_id_and_created_at_still_present(self, tmp_path: pathlib.Path) -> None:
165 repo = _make_repo(tmp_path)
166 sid = _snap(repo)
167 data = json.loads(_rsj(repo, "--no-manifest", sid).output)
168 assert data["snapshot_id"] == sid
169 assert "created_at" in data
170
171 def test_no_manifest_with_text_errors(self, tmp_path: pathlib.Path) -> None:
172 repo = _make_repo(tmp_path)
173 sid = _snap(repo)
174 result = _rs(repo, "--no-manifest", sid)
175 assert result.exit_code == ExitCode.USER_ERROR
176
177
178 # ---------------------------------------------------------------------------
179 # Integration — --path-prefix
180 # ---------------------------------------------------------------------------
181
182
183 class TestPathPrefix:
184 def test_prefix_filters_manifest(self, tmp_path: pathlib.Path) -> None:
185 repo = _make_repo(tmp_path)
186 sid = _snap(repo, {
187 "src/a.py": _fake_oid(1),
188 "src/b.py": _fake_oid(2),
189 "tests/c.py": _fake_oid(3),
190 })
191 data = json.loads(_rsj(repo, "--path-prefix", "src/", sid).output)
192 assert set(data["manifest"].keys()) == {"src/a.py", "src/b.py"}
193 assert data["file_count"] == 2
194
195 def test_prefix_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
196 repo = _make_repo(tmp_path)
197 sid = _snap(repo, {"src/a.py": _fake_oid(1)})
198 data = json.loads(_rsj(repo, "--path-prefix", "docs/", sid).output)
199 assert data["manifest"] == {}
200 assert data["file_count"] == 0
201
202 def test_prefix_with_text_errors(self, tmp_path: pathlib.Path) -> None:
203 repo = _make_repo(tmp_path)
204 sid = _snap(repo)
205 result = _rs(repo, "--path-prefix", "src/", sid)
206 assert result.exit_code == ExitCode.USER_ERROR
207
208
209 # ---------------------------------------------------------------------------
210 # Error cases
211 # ---------------------------------------------------------------------------
212
213
214 class TestErrors:
215 def test_missing_snapshot_errors(self, tmp_path: pathlib.Path) -> None:
216 repo = _make_repo(tmp_path)
217 # Valid sha256: format but not present in the store — must get "not found".
218 result = _rs(repo, long_id(f"dead{'beef' * 15}"))
219 assert result.exit_code == ExitCode.USER_ERROR
220
221 def test_invalid_snapshot_id_errors(self, tmp_path: pathlib.Path) -> None:
222 repo = _make_repo(tmp_path)
223 result = _rs(repo, "not-valid")
224 assert result.exit_code == ExitCode.USER_ERROR
225
226 def test_unknown_format_errors_argparse_rejects(self, tmp_path: pathlib.Path) -> None:
227 """--format flag no longer exists; argparse exits 2."""
228 repo = _make_repo(tmp_path)
229 sid = _snap(repo)
230 result = _rs(repo, "--format", "msgpack", sid)
231 assert result.exit_code != 0 # argparse rejects unknown flag with exit 2
232
233
234 # ---------------------------------------------------------------------------
235 # Security
236 # ---------------------------------------------------------------------------
237
238
239 class TestSecurity:
240 def test_ansi_in_snapshot_id_rejected(self, tmp_path: pathlib.Path) -> None:
241 repo = _make_repo(tmp_path)
242 result = _rs(repo, f"\x1b[31m{'a' * 64}")
243 assert result.exit_code == ExitCode.USER_ERROR
244
245 def test_no_traceback_on_bad_id(self, tmp_path: pathlib.Path) -> None:
246 repo = _make_repo(tmp_path)
247 result = _rs(repo, "bad-id")
248 assert "Traceback" not in result.output
249
250
251 # ---------------------------------------------------------------------------
252 # Stress
253 # ---------------------------------------------------------------------------
254
255
256 class TestStress:
257 def test_1000_file_manifest(self, tmp_path: pathlib.Path) -> None:
258 repo = _make_repo(tmp_path)
259 manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)}
260 sid = _snap(repo, manifest)
261 result = _rsj(repo, sid)
262 assert result.exit_code == 0
263 data = json.loads(result.output)
264 assert data["file_count"] == 1000
265 assert len(data["manifest"]) == 1000
266
267 def test_1000_file_manifest_no_manifest(self, tmp_path: pathlib.Path) -> None:
268 repo = _make_repo(tmp_path)
269 manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)}
270 sid = _snap(repo, manifest)
271 result = _rsj(repo, "--no-manifest", sid)
272 assert result.exit_code == 0
273 data = json.loads(result.output)
274 assert data["file_count"] == 1000
275 assert "manifest" not in data
276
277 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
278 repo = _make_repo(tmp_path)
279 sid = _snap(repo, {"a.py": _fake_oid(0)})
280 for i in range(200):
281 result = _rsj(repo, sid)
282 assert result.exit_code == 0, f"failed at iteration {i}"
283 data = json.loads(result.output)
284 assert data["file_count"] == 1
285
286
287 class TestRegisterFlags:
288 def _parse(self, *args: str) -> "argparse.Namespace":
289 import argparse
290 from muse.cli.commands.read_snapshot import register
291 p = argparse.ArgumentParser()
292 subs = p.add_subparsers()
293 register(subs)
294 return p.parse_args(["read-snapshot", fake_id("a"), *args])
295
296 def test_json_short_flag(self) -> None:
297 args = self._parse("-j")
298 assert args.json_out is True
299
300 def test_json_long_flag(self) -> None:
301 args = self._parse("--json")
302 assert args.json_out is True
303
304 def test_default_no_json(self) -> None:
305 args = self._parse()
306 assert args.json_out is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago