test_core_store.py
python
| 1 | """Tests for muse.core.store — file-based commit and snapshot storage.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import datetime |
| 5 | import json |
| 6 | import pathlib |
| 7 | |
| 8 | import pytest |
| 9 | |
| 10 | from muse.core.store import ( |
| 11 | CommitRecord, |
| 12 | SnapshotRecord, |
| 13 | TagRecord, |
| 14 | find_commits_by_prefix, |
| 15 | get_all_commits, |
| 16 | get_all_tags, |
| 17 | get_commits_for_branch, |
| 18 | get_head_commit_id, |
| 19 | get_head_snapshot_id, |
| 20 | get_head_snapshot_manifest, |
| 21 | get_tags_for_commit, |
| 22 | read_commit, |
| 23 | read_snapshot, |
| 24 | update_commit_metadata, |
| 25 | write_commit, |
| 26 | write_snapshot, |
| 27 | write_tag, |
| 28 | ) |
| 29 | |
| 30 | |
| 31 | @pytest.fixture |
| 32 | def repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 33 | """Create a minimal .muse/ directory structure.""" |
| 34 | muse_dir = tmp_path / ".muse" |
| 35 | (muse_dir / "commits").mkdir(parents=True) |
| 36 | (muse_dir / "snapshots").mkdir(parents=True) |
| 37 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 38 | (muse_dir / "repo.json").write_text(json.dumps({"repo_id": "test-repo"})) |
| 39 | (muse_dir / "HEAD").write_text("refs/heads/main\n") |
| 40 | (muse_dir / "refs" / "heads" / "main").write_text("") |
| 41 | return tmp_path |
| 42 | |
| 43 | |
| 44 | def _make_commit(root: pathlib.Path, commit_id: str, snapshot_id: str, message: str, parent: str | None = None) -> CommitRecord: |
| 45 | c = CommitRecord( |
| 46 | commit_id=commit_id, |
| 47 | repo_id="test-repo", |
| 48 | branch="main", |
| 49 | snapshot_id=snapshot_id, |
| 50 | message=message, |
| 51 | committed_at=datetime.datetime.now(datetime.timezone.utc), |
| 52 | parent_commit_id=parent, |
| 53 | ) |
| 54 | write_commit(root, c) |
| 55 | return c |
| 56 | |
| 57 | |
| 58 | def _make_snapshot(root: pathlib.Path, snapshot_id: str, manifest: dict[str, str]) -> SnapshotRecord: |
| 59 | s = SnapshotRecord(snapshot_id=snapshot_id, manifest=manifest) |
| 60 | write_snapshot(root, s) |
| 61 | return s |
| 62 | |
| 63 | |
| 64 | class TestWriteReadCommit: |
| 65 | def test_roundtrip(self, repo: pathlib.Path) -> None: |
| 66 | c = _make_commit(repo, "abc123", "snap1", "Initial commit") |
| 67 | loaded = read_commit(repo, "abc123") |
| 68 | assert loaded is not None |
| 69 | assert loaded.commit_id == "abc123" |
| 70 | assert loaded.message == "Initial commit" |
| 71 | assert loaded.repo_id == "test-repo" |
| 72 | |
| 73 | def test_read_missing_returns_none(self, repo: pathlib.Path) -> None: |
| 74 | assert read_commit(repo, "nonexistent") is None |
| 75 | |
| 76 | def test_idempotent_write(self, repo: pathlib.Path) -> None: |
| 77 | _make_commit(repo, "abc123", "snap1", "First") |
| 78 | _make_commit(repo, "abc123", "snap1", "Second") # Should not overwrite |
| 79 | loaded = read_commit(repo, "abc123") |
| 80 | assert loaded is not None |
| 81 | assert loaded.message == "First" |
| 82 | |
| 83 | def test_metadata_preserved(self, repo: pathlib.Path) -> None: |
| 84 | c = CommitRecord( |
| 85 | commit_id="abc123", |
| 86 | repo_id="test-repo", |
| 87 | branch="main", |
| 88 | snapshot_id="snap1", |
| 89 | message="With metadata", |
| 90 | committed_at=datetime.datetime.now(datetime.timezone.utc), |
| 91 | metadata={"section": "chorus", "emotion": "joyful"}, |
| 92 | ) |
| 93 | write_commit(repo, c) |
| 94 | loaded = read_commit(repo, "abc123") |
| 95 | assert loaded is not None |
| 96 | assert loaded.metadata["section"] == "chorus" |
| 97 | assert loaded.metadata["emotion"] == "joyful" |
| 98 | |
| 99 | |
| 100 | class TestUpdateCommitMetadata: |
| 101 | def test_set_key(self, repo: pathlib.Path) -> None: |
| 102 | _make_commit(repo, "abc123", "snap1", "msg") |
| 103 | result = update_commit_metadata(repo, "abc123", "tempo_bpm", 120.0) |
| 104 | assert result is True |
| 105 | loaded = read_commit(repo, "abc123") |
| 106 | assert loaded is not None |
| 107 | assert loaded.metadata["tempo_bpm"] == 120.0 |
| 108 | |
| 109 | def test_missing_commit_returns_false(self, repo: pathlib.Path) -> None: |
| 110 | assert update_commit_metadata(repo, "missing", "k", "v") is False |
| 111 | |
| 112 | |
| 113 | class TestWriteReadSnapshot: |
| 114 | def test_roundtrip(self, repo: pathlib.Path) -> None: |
| 115 | s = _make_snapshot(repo, "snap1", {"tracks/drums.mid": "deadbeef"}) |
| 116 | loaded = read_snapshot(repo, "snap1") |
| 117 | assert loaded is not None |
| 118 | assert loaded.manifest == {"tracks/drums.mid": "deadbeef"} |
| 119 | |
| 120 | def test_read_missing_returns_none(self, repo: pathlib.Path) -> None: |
| 121 | assert read_snapshot(repo, "nonexistent") is None |
| 122 | |
| 123 | |
| 124 | class TestHeadQueries: |
| 125 | def test_get_head_commit_id_empty_branch(self, repo: pathlib.Path) -> None: |
| 126 | assert get_head_commit_id(repo, "main") is None |
| 127 | |
| 128 | def test_get_head_commit_id(self, repo: pathlib.Path) -> None: |
| 129 | (repo / ".muse" / "refs" / "heads" / "main").write_text("abc123") |
| 130 | assert get_head_commit_id(repo, "main") == "abc123" |
| 131 | |
| 132 | def test_get_head_snapshot_id(self, repo: pathlib.Path) -> None: |
| 133 | _make_commit(repo, "abc123", "snap1", "msg") |
| 134 | _make_snapshot(repo, "snap1", {"f.mid": "hash1"}) |
| 135 | (repo / ".muse" / "refs" / "heads" / "main").write_text("abc123") |
| 136 | assert get_head_snapshot_id(repo, "test-repo", "main") == "snap1" |
| 137 | |
| 138 | def test_get_head_snapshot_manifest(self, repo: pathlib.Path) -> None: |
| 139 | _make_commit(repo, "abc123", "snap1", "msg") |
| 140 | _make_snapshot(repo, "snap1", {"f.mid": "hash1"}) |
| 141 | (repo / ".muse" / "refs" / "heads" / "main").write_text("abc123") |
| 142 | manifest = get_head_snapshot_manifest(repo, "test-repo", "main") |
| 143 | assert manifest == {"f.mid": "hash1"} |
| 144 | |
| 145 | |
| 146 | class TestGetCommitsForBranch: |
| 147 | def test_chain(self, repo: pathlib.Path) -> None: |
| 148 | _make_commit(repo, "root", "snap0", "Root") |
| 149 | _make_commit(repo, "child", "snap1", "Child", parent="root") |
| 150 | _make_commit(repo, "grandchild", "snap2", "Grandchild", parent="child") |
| 151 | (repo / ".muse" / "refs" / "heads" / "main").write_text("grandchild") |
| 152 | |
| 153 | commits = get_commits_for_branch(repo, "test-repo", "main") |
| 154 | assert [c.commit_id for c in commits] == ["grandchild", "child", "root"] |
| 155 | |
| 156 | def test_empty_branch(self, repo: pathlib.Path) -> None: |
| 157 | assert get_commits_for_branch(repo, "test-repo", "main") == [] |
| 158 | |
| 159 | |
| 160 | class TestFindByPrefix: |
| 161 | def test_finds_match(self, repo: pathlib.Path) -> None: |
| 162 | _make_commit(repo, "abcdef1234", "snap1", "msg") |
| 163 | results = find_commits_by_prefix(repo, "abcdef") |
| 164 | assert len(results) == 1 |
| 165 | assert results[0].commit_id == "abcdef1234" |
| 166 | |
| 167 | def test_no_match(self, repo: pathlib.Path) -> None: |
| 168 | assert find_commits_by_prefix(repo, "zzz") == [] |
| 169 | |
| 170 | |
| 171 | class TestTags: |
| 172 | def test_write_and_read(self, repo: pathlib.Path) -> None: |
| 173 | _make_commit(repo, "abc123", "snap1", "msg") |
| 174 | write_tag(repo, TagRecord( |
| 175 | tag_id="tag1", |
| 176 | repo_id="test-repo", |
| 177 | commit_id="abc123", |
| 178 | tag="emotion:joyful", |
| 179 | )) |
| 180 | tags = get_tags_for_commit(repo, "test-repo", "abc123") |
| 181 | assert len(tags) == 1 |
| 182 | assert tags[0].tag == "emotion:joyful" |
| 183 | |
| 184 | def test_get_all_tags(self, repo: pathlib.Path) -> None: |
| 185 | _make_commit(repo, "abc123", "snap1", "msg") |
| 186 | write_tag(repo, TagRecord(tag_id="t1", repo_id="test-repo", commit_id="abc123", tag="stage:rough-mix")) |
| 187 | write_tag(repo, TagRecord(tag_id="t2", repo_id="test-repo", commit_id="abc123", tag="key:Am")) |
| 188 | all_tags = get_all_tags(repo, "test-repo") |
| 189 | assert len(all_tags) == 2 |