gabriel / muse public
test_algo_prefix_globs.py python
322 lines 11.6 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Regression tests for algo-prefixed filesystem glob patterns.
2
3 After the algo-prefix sprint every content-addressed store path gained an
4 algorithm directory segment:
5
6 commits/sha256/<hex>.msgpack (was: commits/<hex>.msgpack)
7 snapshots/sha256/<hex>.msgpack (was: snapshots/<hex>.msgpack)
8
9 Any glob that scans only the top level of the commits/ or snapshots/ dir
10 silently returns an empty list. These tests catch that class of bug by
11 writing real records through the store, then asserting that every listing
12 and prefix-resolution function returns them.
13
14 Covered functions
15 -----------------
16 muse.core.store
17 get_all_commits
18 _find_commit_by_prefix
19 find_commits_by_prefix
20
21 muse.cli.commands.snapshot_cmd
22 _list_all_snapshots (internal listing helper)
23 _resolve_snapshot (full-id + prefix-scan paths)
24
25 muse.cli.commands.snapshot_diff
26 _resolve_snapshot_prefix (short sha256: prefix resolution)
27 _resolve_to_snapshot_id (end-to-end resolver)
28 """
29
30 from __future__ import annotations
31
32 import datetime
33 import json
34 import pathlib
35
36 import pytest
37
38 from muse.core._types import fake_id, long_id, short_id
39 from muse.core.snapshot import compute_snapshot_id
40 from muse.core.store import (
41 CommitRecord,
42 SnapshotRecord,
43 find_commits_by_prefix,
44 get_all_commits,
45 write_commit,
46 write_snapshot,
47 )
48 from muse.core.store import _find_commit_by_prefix # type: ignore[attr-defined]
49
50 # ---------------------------------------------------------------------------
51 # Helpers
52 # ---------------------------------------------------------------------------
53
54 _REPO_ID = fake_id("repo")
55 _NOW = datetime.datetime.now(datetime.timezone.utc)
56
57
58 def _init_repo(path: pathlib.Path) -> pathlib.Path:
59 muse = path / ".muse"
60 (muse / "refs" / "heads").mkdir(parents=True)
61 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
62 (muse / "repo.json").write_text(
63 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
64 )
65 return path
66
67
68 def _make_snapshot(repo: pathlib.Path, seed: str) -> SnapshotRecord:
69 manifest = {f"{seed}.py": fake_id("obj")}
70 sid = compute_snapshot_id(manifest)
71 snap = SnapshotRecord(
72 snapshot_id=sid,
73 manifest=manifest,
74 directories=[],
75 created_at=_NOW,
76 note=None,
77 )
78 write_snapshot(repo, snap)
79 return snap
80
81
82 def _make_commit(
83 repo: pathlib.Path,
84 snap: SnapshotRecord,
85 msg: str,
86 parent: str | None = None,
87 agent_id: str | None = None,
88 ) -> CommitRecord:
89 from muse.core.store import compute_commit_id
90
91 parents = [parent] if parent else []
92 cid = compute_commit_id(
93 repo_id=_REPO_ID,
94 parent_ids=parents,
95 snapshot_id=snap.snapshot_id,
96 message=msg,
97 committed_at_iso=_NOW.isoformat(),
98 )
99 c = CommitRecord(
100 commit_id=cid,
101 repo_id=_REPO_ID,
102 created_on_branch="main",
103 snapshot_id=snap.snapshot_id,
104 message=msg,
105 committed_at=_NOW,
106 parent_commit_id=parent,
107 agent_id=agent_id,
108 )
109 write_commit(repo, c)
110 return c
111
112
113 # ---------------------------------------------------------------------------
114 # store.get_all_commits
115 # ---------------------------------------------------------------------------
116
117
118 class TestGetAllCommits:
119 def test_returns_written_commit(self, tmp_path: pathlib.Path) -> None:
120 repo = _init_repo(tmp_path)
121 snap = _make_snapshot(repo, "alpha")
122 c = _make_commit(repo, snap, "first")
123 commits = get_all_commits(repo)
124 ids = [x.commit_id for x in commits]
125 assert c.commit_id in ids
126
127 def test_returns_all_written_commits(self, tmp_path: pathlib.Path) -> None:
128 repo = _init_repo(tmp_path)
129 snap_a = _make_snapshot(repo, "file_a")
130 snap_b = _make_snapshot(repo, "file_b")
131 snap_c = _make_snapshot(repo, "file_c")
132 c1 = _make_commit(repo, snap_a, "one")
133 c2 = _make_commit(repo, snap_b, "two", parent=c1.commit_id)
134 c3 = _make_commit(repo, snap_c, "three", parent=c2.commit_id)
135 commits = get_all_commits(repo)
136 ids = {x.commit_id for x in commits}
137 assert {c1.commit_id, c2.commit_id, c3.commit_id} == ids
138
139 def test_empty_repo_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
140 repo = _init_repo(tmp_path)
141 assert get_all_commits(repo) == []
142
143 def test_missing_commits_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
144 repo = _init_repo(tmp_path)
145 assert get_all_commits(repo) == []
146
147
148 # ---------------------------------------------------------------------------
149 # store._find_commit_by_prefix / find_commits_by_prefix
150 # ---------------------------------------------------------------------------
151
152
153 class TestCommitPrefixScan:
154 def test_find_by_bare_hex_prefix(self, tmp_path: pathlib.Path) -> None:
155 repo = _init_repo(tmp_path)
156 snap = _make_snapshot(repo, "prefix_test")
157 c = _make_commit(repo, snap, "prefix commit")
158 bare_hex = long_id(c.commit_id, strip=True)
159 prefix = bare_hex[:12]
160 found = _find_commit_by_prefix(repo, prefix)
161 assert found is not None
162 assert found.commit_id == c.commit_id
163
164 def test_find_commits_by_prefix_returns_list(self, tmp_path: pathlib.Path) -> None:
165 repo = _init_repo(tmp_path)
166 snap = _make_snapshot(repo, "prefix_multi")
167 c = _make_commit(repo, snap, "multi prefix")
168 bare_hex = long_id(c.commit_id, strip=True)
169 results = find_commits_by_prefix(repo, bare_hex[:8])
170 assert any(x.commit_id == c.commit_id for x in results)
171
172 def test_no_match_returns_none(self, tmp_path: pathlib.Path) -> None:
173 repo = _init_repo(tmp_path)
174 assert _find_commit_by_prefix(repo, "0" * 64) is None
175
176 def test_find_commits_by_prefix_no_match_returns_empty(
177 self, tmp_path: pathlib.Path
178 ) -> None:
179 repo = _init_repo(tmp_path)
180 assert find_commits_by_prefix(repo, "0" * 64) == []
181
182
183 # ---------------------------------------------------------------------------
184 # snapshot_cmd._list_all_snapshots (internal listing helper)
185 # ---------------------------------------------------------------------------
186
187
188 class TestGetAllSnapshots:
189 def test_returns_written_snapshot(self, tmp_path: pathlib.Path) -> None:
190 from muse.cli.commands.snapshot_cmd import _list_all_snapshots
191
192 repo = _init_repo(tmp_path)
193 snap = _make_snapshot(repo, "listed")
194 results = _list_all_snapshots(repo)
195 ids = [s.snapshot_id for s in results]
196 assert snap.snapshot_id in ids
197
198 def test_returns_all_written_snapshots(self, tmp_path: pathlib.Path) -> None:
199 from muse.cli.commands.snapshot_cmd import _list_all_snapshots
200
201 repo = _init_repo(tmp_path)
202 s1 = _make_snapshot(repo, "snap_one")
203 s2 = _make_snapshot(repo, "snap_two")
204 s3 = _make_snapshot(repo, "snap_three")
205 results = _list_all_snapshots(repo)
206 ids = {s.snapshot_id for s in results}
207 assert {s1.snapshot_id, s2.snapshot_id, s3.snapshot_id} == ids
208
209 def test_empty_repo_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
210 from muse.cli.commands.snapshot_cmd import _list_all_snapshots
211
212 repo = _init_repo(tmp_path)
213 assert _list_all_snapshots(repo) == []
214
215
216 # ---------------------------------------------------------------------------
217 # snapshot_cmd._resolve_snapshot — full ID path
218 # ---------------------------------------------------------------------------
219
220
221 class TestResolveSnapshotFullId:
222 def test_full_prefixed_id_resolves(self, tmp_path: pathlib.Path) -> None:
223 from muse.cli.commands.snapshot_cmd import _resolve_snapshot
224
225 repo = _init_repo(tmp_path)
226 snap = _make_snapshot(repo, "full_resolve")
227 result = _resolve_snapshot(repo, snap.snapshot_id)
228 assert result is not None
229 assert result.snapshot_id == snap.snapshot_id
230
231 def test_unknown_id_returns_none(self, tmp_path: pathlib.Path) -> None:
232 from muse.cli.commands.snapshot_cmd import _resolve_snapshot
233
234 repo = _init_repo(tmp_path)
235 _make_snapshot(repo, "some_snap")
236 result = _resolve_snapshot(repo, fake_id("f"))
237 assert result is None
238
239
240 # ---------------------------------------------------------------------------
241 # snapshot_cmd._resolve_snapshot — prefix-scan path (short sha256: prefix)
242 # ---------------------------------------------------------------------------
243
244
245 class TestResolveSnapshotShortPrefix:
246 def test_short_prefixed_id_resolves(self, tmp_path: pathlib.Path) -> None:
247 from muse.cli.commands.snapshot_cmd import _resolve_snapshot
248
249 repo = _init_repo(tmp_path)
250 snap = _make_snapshot(repo, "short_prefix")
251 short = short_id(snap.snapshot_id) # sha256:<12hex>
252 result = _resolve_snapshot(repo, short)
253 assert result is not None, f"prefix {short!r} must resolve; got None"
254 assert result.snapshot_id == snap.snapshot_id
255
256 def test_unrecognized_prefix_returns_none(self, tmp_path: pathlib.Path) -> None:
257 from muse.cli.commands.snapshot_cmd import _resolve_snapshot
258
259 repo = _init_repo(tmp_path)
260 _make_snapshot(repo, "exists")
261 result = _resolve_snapshot(repo, "sha256:000000000000")
262 assert result is None
263
264
265 # ---------------------------------------------------------------------------
266 # snapshot_diff._resolve_snapshot_prefix
267 # ---------------------------------------------------------------------------
268
269
270 class TestResolveDiffSnapshotPrefix:
271 def test_short_prefixed_id_resolves(self, tmp_path: pathlib.Path) -> None:
272 from muse.cli.commands.snapshot_diff import _resolve_snapshot_prefix
273
274 repo = _init_repo(tmp_path)
275 snap = _make_snapshot(repo, "diff_prefix")
276 short = short_id(snap.snapshot_id) # sha256:<12hex>
277 result = _resolve_snapshot_prefix(repo, short)
278 assert result is not None, f"prefix {short!r} must resolve; got None"
279 assert result == snap.snapshot_id
280
281 def test_bare_hex_rejected(self, tmp_path: pathlib.Path) -> None:
282 from muse.cli.commands.snapshot_diff import _resolve_snapshot_prefix
283
284 repo = _init_repo(tmp_path)
285 snap = _make_snapshot(repo, "bare_hex")
286 bare = long_id(snap.snapshot_id, strip=True)[:12]
287 result = _resolve_snapshot_prefix(repo, bare)
288 assert result is None, "bare hex without sha256: prefix must be rejected"
289
290 def test_unknown_prefix_returns_none(self, tmp_path: pathlib.Path) -> None:
291 from muse.cli.commands.snapshot_diff import _resolve_snapshot_prefix
292
293 repo = _init_repo(tmp_path)
294 _make_snapshot(repo, "some_snap")
295 result = _resolve_snapshot_prefix(repo, "sha256:000000000000")
296 assert result is None
297
298
299 # ---------------------------------------------------------------------------
300 # snapshot_diff._resolve_to_snapshot_id — end-to-end via short prefix
301 # ---------------------------------------------------------------------------
302
303
304 class TestResolveToSnapshotId:
305 def test_short_prefixed_snapshot_id_resolves(
306 self, tmp_path: pathlib.Path
307 ) -> None:
308 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
309
310 repo = _init_repo(tmp_path)
311 snap = _make_snapshot(repo, "e2e_short")
312 short = short_id(snap.snapshot_id)
313 result = _resolve_to_snapshot_id(repo, short)
314 assert result == snap.snapshot_id
315
316 def test_full_snapshot_id_resolves(self, tmp_path: pathlib.Path) -> None:
317 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
318
319 repo = _init_repo(tmp_path)
320 snap = _make_snapshot(repo, "e2e_full")
321 result = _resolve_to_snapshot_id(repo, snap.snapshot_id)
322 assert result == snap.snapshot_id
File History 1 commit
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago