gabriel / muse public
test_plumbing_read_snapshot.py python
277 lines 10.0 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive tests for ``muse plumbing 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.snapshot import compute_snapshot_id
18 from muse.core.store import SnapshotRecord, write_snapshot
19 from tests.cli_test_helper import CliRunner, InvokeResult
20
21 runner = CliRunner()
22
23 _CREATED_AT: datetime.datetime = datetime.datetime(
24 2026, 3, 18, 12, 0, tzinfo=datetime.timezone.utc
25 )
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
33 repo = tmp_path / "repo"
34 muse = repo / ".muse"
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (muse / sub).mkdir(parents=True)
37 (muse / "HEAD").write_text("ref: refs/heads/main")
38 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
39 return repo
40
41
42 def _snap(
43 repo: pathlib.Path,
44 manifest: Manifest | None = None,
45 ) -> str:
46 """Write a snapshot with a real content-addressed ID; return the snapshot_id."""
47 m: Manifest = manifest or {}
48 snap_id = compute_snapshot_id(m)
49 rec = SnapshotRecord(
50 snapshot_id=snap_id,
51 manifest=m,
52 created_at=_CREATED_AT,
53 )
54 write_snapshot(repo, rec)
55 return snap_id
56
57
58 def _rs(repo: pathlib.Path, *args: str) -> InvokeResult:
59 from muse.cli.app import main as cli
60 return runner.invoke(
61 cli,
62 ["read-snapshot", *args],
63 env={"MUSE_REPO_ROOT": str(repo)},
64 )
65
66
67 def _fake_oid(n: int) -> str:
68 return format(n, "064x")
69
70
71 # ---------------------------------------------------------------------------
72 # Integration — JSON format
73 # ---------------------------------------------------------------------------
74
75
76 class TestJsonFormat:
77 def test_full_output_empty_manifest(self, tmp_path: pathlib.Path) -> None:
78 repo = _make_repo(tmp_path)
79 sid = _snap(repo)
80 result = _rs(repo, sid)
81 assert result.exit_code == 0
82 data = json.loads(result.output)
83 assert data["snapshot_id"] == sid
84 assert data["file_count"] == 0
85 assert data["manifest"] == {}
86
87 def test_manifest_paths_present(self, tmp_path: pathlib.Path) -> None:
88 repo = _make_repo(tmp_path)
89 oid = "0" * 64
90 sid = _snap(repo, {"src/main.py": oid, "tests/test_main.py": oid})
91 data = json.loads(_rs(repo, sid).output)
92 assert "src/main.py" in data["manifest"]
93 assert "tests/test_main.py" in data["manifest"]
94 assert data["file_count"] == 2
95
96 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
97 repo = _make_repo(tmp_path)
98 sid = _snap(repo, {"a.py": _fake_oid(1)})
99 result = _rs(repo, "--json", sid)
100 assert result.exit_code == 0
101 assert "snapshot_id" in json.loads(result.output)
102
103 def test_created_at_iso8601(self, tmp_path: pathlib.Path) -> None:
104 repo = _make_repo(tmp_path)
105 sid = _snap(repo, {"b.py": _fake_oid(2)})
106 data = json.loads(_rs(repo, sid).output)
107 datetime.datetime.fromisoformat(data["created_at"])
108
109 def test_file_count_reflects_manifest(self, tmp_path: pathlib.Path) -> None:
110 repo = _make_repo(tmp_path)
111 manifest = {f"file{i}.py": _fake_oid(i) for i in range(5)}
112 sid = _snap(repo, manifest)
113 data = json.loads(_rs(repo, sid).output)
114 assert data["file_count"] == 5
115
116
117 # ---------------------------------------------------------------------------
118 # Integration — text format
119 # ---------------------------------------------------------------------------
120
121
122 class TestTextFormat:
123 def test_text_contains_prefix(self, tmp_path: pathlib.Path) -> None:
124 repo = _make_repo(tmp_path)
125 sid = _snap(repo, {"c.py": _fake_oid(3)})
126 result = _rs(repo, "--format", "text", sid)
127 assert result.exit_code == 0
128 assert sid[:12] in result.output
129
130 def test_text_contains_file_count(self, tmp_path: pathlib.Path) -> None:
131 repo = _make_repo(tmp_path)
132 sid = _snap(repo, {"a.py": _fake_oid(1), "b.py": _fake_oid(2)})
133 result = _rs(repo, "--format", "text", sid)
134 assert "2 files" in result.output
135
136 def test_text_single_line(self, tmp_path: pathlib.Path) -> None:
137 repo = _make_repo(tmp_path)
138 sid = _snap(repo)
139 result = _rs(repo, "--format", "text", sid)
140 lines = [l for l in result.output.splitlines() if l.strip()]
141 assert len(lines) == 1
142
143
144 # ---------------------------------------------------------------------------
145 # Integration — --no-manifest
146 # ---------------------------------------------------------------------------
147
148
149 class TestNoManifest:
150 def test_manifest_absent(self, tmp_path: pathlib.Path) -> None:
151 repo = _make_repo(tmp_path)
152 sid = _snap(repo, {"a.py": _fake_oid(1)})
153 data = json.loads(_rs(repo, "--no-manifest", sid).output)
154 assert "manifest" not in data
155 assert data["file_count"] == 1
156
157 def test_snapshot_id_and_created_at_still_present(self, tmp_path: pathlib.Path) -> None:
158 repo = _make_repo(tmp_path)
159 sid = _snap(repo)
160 data = json.loads(_rs(repo, "--no-manifest", sid).output)
161 assert data["snapshot_id"] == sid
162 assert "created_at" in data
163
164 def test_no_manifest_with_text_errors(self, tmp_path: pathlib.Path) -> None:
165 repo = _make_repo(tmp_path)
166 sid = _snap(repo)
167 result = _rs(repo, "--no-manifest", "--format", "text", sid)
168 assert result.exit_code == ExitCode.USER_ERROR
169
170
171 # ---------------------------------------------------------------------------
172 # Integration — --path-prefix
173 # ---------------------------------------------------------------------------
174
175
176 class TestPathPrefix:
177 def test_prefix_filters_manifest(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 sid = _snap(repo, {
180 "src/a.py": _fake_oid(1),
181 "src/b.py": _fake_oid(2),
182 "tests/c.py": _fake_oid(3),
183 })
184 data = json.loads(_rs(repo, "--path-prefix", "src/", sid).output)
185 assert set(data["manifest"].keys()) == {"src/a.py", "src/b.py"}
186 assert data["file_count"] == 2
187
188 def test_prefix_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
189 repo = _make_repo(tmp_path)
190 sid = _snap(repo, {"src/a.py": _fake_oid(1)})
191 data = json.loads(_rs(repo, "--path-prefix", "docs/", sid).output)
192 assert data["manifest"] == {}
193 assert data["file_count"] == 0
194
195 def test_prefix_with_text_errors(self, tmp_path: pathlib.Path) -> None:
196 repo = _make_repo(tmp_path)
197 sid = _snap(repo)
198 result = _rs(repo, "--path-prefix", "src/", "--format", "text", sid)
199 assert result.exit_code == ExitCode.USER_ERROR
200
201
202 # ---------------------------------------------------------------------------
203 # Error cases
204 # ---------------------------------------------------------------------------
205
206
207 class TestErrors:
208 def test_missing_snapshot_errors(self, tmp_path: pathlib.Path) -> None:
209 repo = _make_repo(tmp_path)
210 result = _rs(repo, "dead" + "beef" * 15)
211 assert result.exit_code == ExitCode.USER_ERROR
212 data = json.loads(result.output)
213 assert "not found" in data["error"]
214
215 def test_invalid_snapshot_id_errors(self, tmp_path: pathlib.Path) -> None:
216 repo = _make_repo(tmp_path)
217 result = _rs(repo, "not-valid")
218 assert result.exit_code == ExitCode.USER_ERROR
219
220 def test_unknown_format_errors(self, tmp_path: pathlib.Path) -> None:
221 repo = _make_repo(tmp_path)
222 sid = _snap(repo)
223 result = _rs(repo, "--format", "msgpack", sid)
224 assert result.exit_code == ExitCode.USER_ERROR
225
226
227 # ---------------------------------------------------------------------------
228 # Security
229 # ---------------------------------------------------------------------------
230
231
232 class TestSecurity:
233 def test_ansi_in_snapshot_id_rejected(self, tmp_path: pathlib.Path) -> None:
234 repo = _make_repo(tmp_path)
235 result = _rs(repo, "\x1b[31m" + "a" * 64)
236 assert result.exit_code == ExitCode.USER_ERROR
237
238 def test_no_traceback_on_bad_id(self, tmp_path: pathlib.Path) -> None:
239 repo = _make_repo(tmp_path)
240 result = _rs(repo, "bad-id")
241 assert "Traceback" not in result.output
242
243
244 # ---------------------------------------------------------------------------
245 # Stress
246 # ---------------------------------------------------------------------------
247
248
249 class TestStress:
250 def test_1000_file_manifest(self, tmp_path: pathlib.Path) -> None:
251 repo = _make_repo(tmp_path)
252 manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)}
253 sid = _snap(repo, manifest)
254 result = _rs(repo, sid)
255 assert result.exit_code == 0
256 data = json.loads(result.output)
257 assert data["file_count"] == 1000
258 assert len(data["manifest"]) == 1000
259
260 def test_1000_file_manifest_no_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 = _rs(repo, "--no-manifest", sid)
265 assert result.exit_code == 0
266 data = json.loads(result.output)
267 assert data["file_count"] == 1000
268 assert "manifest" not in data
269
270 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
271 repo = _make_repo(tmp_path)
272 sid = _snap(repo, {"a.py": _fake_oid(0)})
273 for i in range(200):
274 result = _rs(repo, sid)
275 assert result.exit_code == 0, f"failed at iteration {i}"
276 data = json.loads(result.output)
277 assert data["file_count"] == 1
File History 3 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago