gabriel / muse public
test_core_store.py python
316 lines 13.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Tests for muse.core.store — file-based commit and snapshot storage."""
2
3 import datetime
4 import json
5 import pathlib
6
7 import pytest
8
9 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
10
11 from muse.core.types import Manifest, fake_id, long_id
12 from muse.core.paths import muse_dir, heads_dir, remote_tracking_dir, remotes_dir
13 from muse.core.store import (
14 CommitDict,
15 CommitRecord,
16 SnapshotRecord,
17 TagRecord,
18 _resolve_branch_commit_id,
19 find_commits_by_prefix,
20 get_all_commits,
21 get_all_tags,
22 get_commits_for_branch,
23 get_head_commit_id,
24 get_head_snapshot_id,
25 get_head_snapshot_manifest,
26 get_tags_for_commit,
27 read_commit,
28 read_snapshot,
29 update_commit_metadata,
30 write_commit,
31 write_snapshot,
32 write_tag,
33 )
34
35
36 @pytest.fixture
37 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
38 """Create a minimal .muse/ directory structure."""
39 dot_muse = muse_dir(tmp_path)
40 (dot_muse / "commits").mkdir(parents=True)
41 (dot_muse / "snapshots").mkdir(parents=True)
42 (dot_muse / "refs" / "heads").mkdir(parents=True)
43 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"}))
44 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
45 (dot_muse / "refs" / "heads" / "main").write_text("")
46 return tmp_path
47
48
49 def _make_commit(
50 root: pathlib.Path,
51 commit_id: str,
52 snapshot_id: str,
53 message: str,
54 parent: str | None = None,
55 ) -> CommitRecord:
56 """Write a commit whose ID is computed from its content fields.
57
58 The ``commit_id`` parameter is kept for call-site readability but is
59 ignored — the real ID is derived by :func:`~muse.core.snapshot.compute_commit_id`
60 so every stored commit satisfies the I-10 content-hash verification.
61 """
62 now = datetime.datetime.now(datetime.timezone.utc)
63 parents: list[str] = [parent] if parent else []
64 real_id = compute_commit_id(
65 parent_ids=parents,
66 snapshot_id=snapshot_id,
67 message=message,
68 committed_at_iso=now.isoformat(),
69 )
70 c = CommitRecord(
71 repo_id="test-repo",
72 commit_id=real_id,
73 branch="main",
74 snapshot_id=snapshot_id,
75 message=message,
76 committed_at=now,
77 parent_commit_id=parent,
78 )
79 write_commit(root, c)
80 return c
81
82
83 def _make_snapshot(root: pathlib.Path, snapshot_id: str, manifest: Manifest) -> SnapshotRecord:
84 """Write a snapshot whose ID is computed from its manifest.
85
86 The ``snapshot_id`` parameter is kept for call-site readability but is
87 ignored — the real ID is derived by
88 :func:`~muse.core.snapshot.compute_snapshot_id` so every stored snapshot
89 satisfies the I-10 content-hash verification.
90 """
91 real_id = compute_snapshot_id(manifest)
92 s = SnapshotRecord(snapshot_id=real_id, manifest=manifest)
93 write_snapshot(root, s)
94 return s
95
96
97 class TestWriteReadCommit:
98 def test_roundtrip(self, repo: pathlib.Path) -> None:
99 c = _make_commit(repo, "ignored", fake_id("snap"), "Initial commit")
100 loaded = read_commit(repo, c.commit_id)
101 assert loaded is not None
102 assert loaded.commit_id == c.commit_id
103 assert loaded.message == "Initial commit"
104 assert loaded.repo_id == "test-repo"
105
106 def test_read_missing_returns_none(self, repo: pathlib.Path) -> None:
107 assert read_commit(repo, fake_id("nonexistent-commit")) is None
108
109 def test_idempotent_write(self, repo: pathlib.Path) -> None:
110 c = _make_commit(repo, "ignored", fake_id("snap"), "First")
111 _make_commit(repo, "ignored", fake_id("snap"), "Second") # different timestamp → different ID; should write
112 loaded = read_commit(repo, c.commit_id)
113 assert loaded is not None
114 assert loaded.message == "First"
115
116 def test_metadata_preserved(self, repo: pathlib.Path) -> None:
117 now = datetime.datetime.now(datetime.timezone.utc)
118 snap_id = fake_id("snap")
119 cid = compute_commit_id(parent_ids=[], snapshot_id=snap_id, message="With metadata", committed_at_iso=now.isoformat())
120 c = CommitRecord(
121 repo_id="test-repo",
122 commit_id=cid,
123 branch="main",
124 snapshot_id=snap_id,
125 message="With metadata",
126 committed_at=now,
127 metadata={"section": "chorus", "emotion": "joyful"},
128 )
129 write_commit(repo, c)
130 loaded = read_commit(repo, cid)
131 assert loaded is not None
132 assert loaded.metadata["section"] == "chorus"
133 assert loaded.metadata["emotion"] == "joyful"
134
135
136 class TestUpdateCommitMetadata:
137 def test_set_key(self, repo: pathlib.Path) -> None:
138 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
139 result = update_commit_metadata(repo, c.commit_id, "tempo_bpm", "120.0")
140 assert result is True
141 loaded = read_commit(repo, c.commit_id)
142 assert loaded is not None
143 assert loaded.metadata["tempo_bpm"] == "120.0"
144
145 def test_missing_commit_returns_false(self, repo: pathlib.Path) -> None:
146 assert update_commit_metadata(repo, fake_id("nonexistent-commit"), "k", "v") is False
147
148
149 class TestWriteReadSnapshot:
150 def test_roundtrip(self, repo: pathlib.Path) -> None:
151 _drum_id = fake_id("deadbeef")
152 s = _make_snapshot(repo, "ignored", {"tracks/drums.mid": _drum_id})
153 loaded = read_snapshot(repo, s.snapshot_id)
154 assert loaded is not None
155 assert loaded.manifest == {"tracks/drums.mid": _drum_id}
156
157 def test_read_missing_returns_none(self, repo: pathlib.Path) -> None:
158 assert read_snapshot(repo, fake_id("nonexistent-snapshot")) is None
159
160
161 class TestHeadQueries:
162 def test_get_head_commit_id_empty_branch(self, repo: pathlib.Path) -> None:
163 assert get_head_commit_id(repo, "main") is None
164
165 def test_get_head_commit_id(self, repo: pathlib.Path) -> None:
166 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
167 (heads_dir(repo) / "main").write_text(c.commit_id)
168 assert get_head_commit_id(repo, "main") == c.commit_id
169
170 def test_get_head_snapshot_id(self, repo: pathlib.Path) -> None:
171 _f_id = fake_id("f.mid-content")
172 snap = _make_snapshot(repo, "ignored", {"f.mid": _f_id})
173 c = _make_commit(repo, "ignored", snap.snapshot_id, "msg")
174 (heads_dir(repo) / "main").write_text(c.commit_id)
175 assert get_head_snapshot_id(repo, "test-repo", "main") == snap.snapshot_id
176
177 def test_get_head_snapshot_manifest(self, repo: pathlib.Path) -> None:
178 _f_id = fake_id("f.mid-content")
179 snap = _make_snapshot(repo, "ignored", {"f.mid": _f_id})
180 c = _make_commit(repo, "ignored", snap.snapshot_id, "msg")
181 (heads_dir(repo) / "main").write_text(c.commit_id)
182 manifest = get_head_snapshot_manifest(repo, "test-repo", "main")
183 assert manifest == {"f.mid": _f_id}
184
185
186 class TestResolveRemoteBranchCommitId:
187 """_resolve_branch_commit_id handles local branches and remote tracking refs."""
188
189 def test_local_branch_resolved(self, repo: pathlib.Path) -> None:
190 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
191 (heads_dir(repo) / "main").write_text(c.commit_id)
192 assert _resolve_branch_commit_id(repo, "main") == c.commit_id
193
194 def test_empty_local_branch_returns_none(self, repo: pathlib.Path) -> None:
195 assert _resolve_branch_commit_id(repo, "main") is None
196
197 def test_remote_tracking_ref_resolved(self, repo: pathlib.Path) -> None:
198 remote_dir = remote_tracking_dir(repo, "origin")
199 remote_dir.mkdir(parents=True)
200 cid = long_id("a" * 64)
201 (remote_dir / "dev").write_text(cid)
202 assert _resolve_branch_commit_id(repo, "origin/dev") == cid
203
204 def test_remote_tracking_ref_missing_returns_none(self, repo: pathlib.Path) -> None:
205 remote_tracking_dir(repo, "origin").mkdir(parents=True)
206 assert _resolve_branch_commit_id(repo, "origin/nonexistent") is None
207
208 def test_remote_tracking_ref_no_remotes_dir_returns_none(self, repo: pathlib.Path) -> None:
209 assert _resolve_branch_commit_id(repo, "origin/dev") is None
210
211 def test_ref_without_slash_only_checks_local(self, repo: pathlib.Path) -> None:
212 # "dev" has no slash — never checks .muse/remotes/
213 assert _resolve_branch_commit_id(repo, "dev") is None
214
215 def test_local_branch_takes_priority_over_remote(self, repo: pathlib.Path) -> None:
216 local_id = long_id("b" * 64)
217 remote_id = long_id("c" * 64)
218 # Both a local branch named "origin/dev" (pathological) and a remote ref exist.
219 local_branch_dir = heads_dir(repo) / "origin"
220 local_branch_dir.mkdir(parents=True)
221 (local_branch_dir / "dev").write_text(local_id)
222 remote_dir = remote_tracking_dir(repo, "origin")
223 remote_dir.mkdir(parents=True)
224 (remote_dir / "dev").write_text(remote_id)
225 # Local branch wins.
226 assert _resolve_branch_commit_id(repo, "origin/dev") == local_id
227
228
229 class TestGetCommitsForBranch:
230 def test_chain(self, repo: pathlib.Path) -> None:
231 root = _make_commit(repo, "ignored", fake_id("snap-root"), "Root")
232 child = _make_commit(repo, "ignored", fake_id("snap-child"), "Child", parent=root.commit_id)
233 grandchild = _make_commit(repo, "ignored", fake_id("snap-grandchild"), "Grandchild", parent=child.commit_id)
234 (heads_dir(repo) / "main").write_text(grandchild.commit_id)
235
236 commits = get_commits_for_branch(repo, "test-repo", "main")
237 assert [c.commit_id for c in commits] == [
238 grandchild.commit_id, child.commit_id, root.commit_id
239 ]
240
241 def test_empty_branch(self, repo: pathlib.Path) -> None:
242 assert get_commits_for_branch(repo, "test-repo", "main") == []
243
244 def test_remote_tracking_ref(self, repo: pathlib.Path) -> None:
245 """get_commits_for_branch resolves 'origin/dev' via .muse/remotes/."""
246 c = _make_commit(repo, "ignored", fake_id("snap"), "Remote commit")
247 remote_dir = remotes_dir(repo) / "origin"
248 remote_dir.mkdir(parents=True)
249 (remote_dir / "dev").write_text(c.commit_id)
250
251 commits = get_commits_for_branch(repo, "test-repo", "origin/dev")
252 assert len(commits) == 1
253 assert commits[0].commit_id == c.commit_id
254
255 def test_remote_tracking_ref_chain(self, repo: pathlib.Path) -> None:
256 root = _make_commit(repo, "ignored", fake_id("snap-root"), "Root")
257 tip = _make_commit(repo, "ignored", fake_id("snap-tip"), "Tip", parent=root.commit_id)
258 remote_dir = remotes_dir(repo) / "upstream"
259 remote_dir.mkdir(parents=True)
260 (remote_dir / "main").write_text(tip.commit_id)
261
262 commits = get_commits_for_branch(repo, "test-repo", "upstream/main")
263 assert [c.commit_id for c in commits] == [tip.commit_id, root.commit_id]
264
265 def test_remote_tracking_ref_missing_returns_empty(self, repo: pathlib.Path) -> None:
266 assert get_commits_for_branch(repo, "test-repo", "origin/dev") == []
267
268 def test_max_count_with_remote_ref(self, repo: pathlib.Path) -> None:
269 commits_written = [_make_commit(repo, "ignored", fake_id(f"snap-s{i}"), f"Commit {i}") for i in range(5)]
270 for i in range(1, 5):
271 commits_written[i] = _make_commit(
272 repo, "ignored", fake_id(f"snap-t{i}"), f"C{i}", parent=commits_written[i - 1].commit_id
273 )
274 tip = _make_commit(repo, "ignored", fake_id("snap-tip"), "Tip", parent=commits_written[-1].commit_id)
275 remote_dir = remotes_dir(repo) / "origin"
276 remote_dir.mkdir(parents=True)
277 (remote_dir / "dev").write_text(tip.commit_id)
278
279 commits = get_commits_for_branch(repo, "test-repo", "origin/dev", max_count=2)
280 assert len(commits) == 2
281 assert commits[0].commit_id == tip.commit_id
282
283
284 class TestFindByPrefix:
285 def test_finds_match(self, repo: pathlib.Path) -> None:
286 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
287 hex_prefix = c.commit_id[len("sha256:"):len("sha256:") + 6]
288 results = find_commits_by_prefix(repo, hex_prefix)
289 assert len(results) == 1
290 assert results[0].commit_id == c.commit_id
291
292 def test_no_match(self, repo: pathlib.Path) -> None:
293 assert find_commits_by_prefix(repo, "zzz") == []
294
295
296 class TestTags:
297 def test_write_and_read(self, repo: pathlib.Path) -> None:
298 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
299 repo_id = fake_id("test-repo")
300 write_tag(repo, TagRecord(
301 repo_id=repo_id,
302 tag_id=fake_id("tag1"),
303 commit_id=c.commit_id,
304 tag="emotion:joyful",
305 ))
306 tags = get_tags_for_commit(repo, repo_id, c.commit_id)
307 assert len(tags) == 1
308 assert tags[0].tag == "emotion:joyful"
309
310 def test_get_all_tags(self, repo: pathlib.Path) -> None:
311 c = _make_commit(repo, "ignored", fake_id("snap"), "msg")
312 repo_id = fake_id("test-repo")
313 write_tag(repo, TagRecord(tag_id=fake_id("t1"), repo_id=repo_id, commit_id=c.commit_id, tag="stage:rough-mix"))
314 write_tag(repo, TagRecord(tag_id=fake_id("t2"), repo_id=repo_id, commit_id=c.commit_id, tag="key:Am"))
315 all_tags = get_all_tags(repo, repo_id)
316 assert len(all_tags) == 2
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago