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