gabriel / muse public
test_cmd_tag.py python
280 lines 12.5 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 125 days ago
1 """Comprehensive tests for ``muse tag``.
2
3 Covers:
4 - Unit: write_tag, delete_tag, get_tags_for_commit, get_all_tags
5 - Integration: add → list → remove round-trip
6 - E2E: full CLI via CliRunner
7 - Security: tag names sanitized, ref validation
8 - Stress: many tags on many commits
9 """
10
11 from __future__ import annotations
12
13 import datetime
14 import json
15 import pathlib
16
17 import pytest
18 from tests.cli_test_helper import CliRunner
19 from muse.core.types import fake_id, short_id
20 from muse.core.paths import muse_dir, ref_path
21
22 cli = None # argparse migration — CliRunner ignores this arg
23
24 runner = CliRunner()
25
26
27 # ---------------------------------------------------------------------------
28 # Helpers
29 # ---------------------------------------------------------------------------
30
31 def _env(root: pathlib.Path) -> Manifest:
32 return {"MUSE_REPO_ROOT": str(root)}
33
34
35 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
36 dot_muse = muse_dir(tmp_path)
37 dot_muse.mkdir()
38 repo_id = fake_id("repo")
39 (dot_muse / "repo.json").write_text(json.dumps({
40 "repo_id": repo_id,
41 "domain": "midi",
42 "default_branch": "main",
43 "created_at": "2025-01-01T00:00:00+00:00",
44 }), encoding="utf-8")
45 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
46 (dot_muse / "refs" / "heads").mkdir(parents=True)
47 (dot_muse / "snapshots").mkdir()
48 (dot_muse / "commits").mkdir()
49 (dot_muse / "objects").mkdir()
50 return tmp_path, repo_id
51
52
53 def _make_commit(
54 root: pathlib.Path, repo_id: str, branch: str = "main", message: str = "test"
55 ) -> str:
56 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
57 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
58
59 ref_file = ref_path(root, branch)
60 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
61 manifest: Manifest = {}
62 snap_id = compute_snapshot_id(manifest)
63 committed_at = datetime.datetime.now(datetime.timezone.utc)
64 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
65 snapshot_id=snap_id, message=message,
66 committed_at_iso=committed_at.isoformat(),
67 )
68 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
69 write_commit(root, CommitRecord(
70 commit_id=commit_id, repo_id=repo_id, branch=branch,
71 snapshot_id=snap_id, message=message, committed_at=committed_at,
72 parent_commit_id=parent_id,
73 ))
74 ref_file.parent.mkdir(parents=True, exist_ok=True)
75 ref_file.write_text(commit_id, encoding="utf-8")
76 return commit_id
77
78
79 # ---------------------------------------------------------------------------
80 # Unit tests
81 # ---------------------------------------------------------------------------
82
83 class TestTagUnit:
84 def test_write_and_read_tag(self, tmp_path: pathlib.Path) -> None:
85 root, repo_id = _init_repo(tmp_path)
86 commit_id = _make_commit(root, repo_id)
87 from muse.core.store import TagRecord, write_tag, get_tags_for_commit
88 tag = TagRecord(tag_id=fake_id("tag"), repo_id=repo_id,
89 commit_id=commit_id, tag="emotion:joyful")
90 write_tag(root, tag)
91 tags = get_tags_for_commit(root, repo_id, commit_id)
92 assert len(tags) == 1
93 assert tags[0].tag == "emotion:joyful"
94
95 def test_delete_tag(self, tmp_path: pathlib.Path) -> None:
96 root, repo_id = _init_repo(tmp_path)
97 commit_id = _make_commit(root, repo_id)
98 from muse.core.store import TagRecord, write_tag, get_tags_for_commit, delete_tag
99 tag_id = fake_id("tag")
100 write_tag(root, TagRecord(tag_id=tag_id, repo_id=repo_id,
101 commit_id=commit_id, tag="section:chorus"))
102 assert len(get_tags_for_commit(root, repo_id, commit_id)) == 1
103 assert delete_tag(root, repo_id, tag_id) is True
104 assert get_tags_for_commit(root, repo_id, commit_id) == []
105
106 def test_delete_nonexistent_tag_returns_false(self, tmp_path: pathlib.Path) -> None:
107 root, repo_id = _init_repo(tmp_path)
108 from muse.core.store import delete_tag
109 assert delete_tag(root, repo_id, fake_id("tag")) is False
110
111 def test_get_all_tags_empty(self, tmp_path: pathlib.Path) -> None:
112 root, repo_id = _init_repo(tmp_path)
113 from muse.core.store import get_all_tags
114 assert get_all_tags(root, repo_id) == []
115
116
117 # ---------------------------------------------------------------------------
118 # Content-addressed tag_id
119 # ---------------------------------------------------------------------------
120
121 class TestTagIdContentAddressed:
122 """tag_id must be sha256: of genesis content, not a UUID."""
123
124 def test_tag_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
125 root, repo_id = _init_repo(tmp_path)
126 commit_id = _make_commit(root, repo_id)
127 from muse.core.store import compute_tag_id
128 tag_id = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="emotion:joyful")
129 assert tag_id.startswith("sha256:"), f"Expected sha256: prefix, got {tag_id!r}"
130 assert len(tag_id) == 71
131
132 def test_tag_id_is_deterministic(self, tmp_path: pathlib.Path) -> None:
133 root, repo_id = _init_repo(tmp_path)
134 commit_id = _make_commit(root, repo_id)
135 from muse.core.store import compute_tag_id
136 id1 = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="v1.0")
137 id2 = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="v1.0")
138 assert id1 == id2
139
140 def test_tag_id_differs_by_tag_name(self, tmp_path: pathlib.Path) -> None:
141 root, repo_id = _init_repo(tmp_path)
142 commit_id = _make_commit(root, repo_id)
143 from muse.core.store import compute_tag_id
144 assert compute_tag_id(repo_id, commit_id, "v1.0") != compute_tag_id(repo_id, commit_id, "v2.0")
145
146 def test_tag_id_differs_by_commit(self, tmp_path: pathlib.Path) -> None:
147 root, repo_id = _init_repo(tmp_path)
148 c1 = _make_commit(root, repo_id, message="first")
149 c2 = _make_commit(root, repo_id, message="second")
150 from muse.core.store import compute_tag_id
151 assert compute_tag_id(repo_id, c1, "v1.0") != compute_tag_id(repo_id, c2, "v1.0")
152
153 def test_tag_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None:
154 import re
155 root, repo_id = _init_repo(tmp_path)
156 commit_id = _make_commit(root, repo_id)
157 from muse.core.store import compute_tag_id
158 tag_id = compute_tag_id(repo_id, commit_id, "release:1.0")
159 assert tag_id.startswith("sha256:")
160 uuid4_re = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
161 assert not uuid4_re.match(tag_id)
162
163 def test_cli_tag_add_returns_content_addressed_id(self, tmp_path: pathlib.Path) -> None:
164 root, repo_id = _init_repo(tmp_path)
165 commit_id = _make_commit(root, repo_id)
166 result = runner.invoke(cli, ["tag", "add", "v1.0", commit_id, "--json"], env=_env(root))
167 assert result.exit_code == 0
168 data = json.loads(result.output)
169 tag_id = data["tag_id"]
170 assert tag_id.startswith("sha256:"), f"Expected sha256: prefix, got {tag_id!r}"
171 assert len(tag_id) == 71
172
173 def test_write_tag_uses_content_addressed_id(self, tmp_path: pathlib.Path) -> None:
174 root, repo_id = _init_repo(tmp_path)
175 commit_id = _make_commit(root, repo_id)
176 from muse.core.store import TagRecord, write_tag, get_tags_for_commit, compute_tag_id
177 expected_id = compute_tag_id(repo_id, commit_id, "emotion:sad")
178 tag = TagRecord(tag_id=expected_id, repo_id=repo_id, commit_id=commit_id, tag="emotion:sad")
179 write_tag(root, tag)
180 tags = get_tags_for_commit(root, repo_id, commit_id)
181 assert tags[0].tag_id == expected_id
182
183
184 # ---------------------------------------------------------------------------
185 # Integration tests
186 # ---------------------------------------------------------------------------
187
188 class TestTagIntegration:
189 def test_add_and_list_tag(self, tmp_path: pathlib.Path) -> None:
190 root, repo_id = _init_repo(tmp_path)
191 _make_commit(root, repo_id)
192 result = runner.invoke(cli, ["tag", "add", "emotion:joyful"], env=_env(root), catch_exceptions=False)
193 assert result.exit_code == 0
194 assert "Tagged" in result.output
195
196 result2 = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
197 assert "emotion:joyful" in result2.output
198
199 def test_list_tags_for_specific_commit(self, tmp_path: pathlib.Path) -> None:
200 root, repo_id = _init_repo(tmp_path)
201 commit_id = _make_commit(root, repo_id)
202 runner.invoke(cli, ["tag", "add", "section:verse"], env=_env(root), catch_exceptions=False)
203 result = runner.invoke(cli, ["tag", "list", short_id(commit_id)], env=_env(root), catch_exceptions=False)
204 assert "section:verse" in result.output
205
206 def test_remove_tag(self, tmp_path: pathlib.Path) -> None:
207 root, repo_id = _init_repo(tmp_path)
208 _make_commit(root, repo_id)
209 runner.invoke(cli, ["tag", "add", "emotion:tense"], env=_env(root), catch_exceptions=False)
210 result = runner.invoke(cli, ["tag", "remove", "emotion:tense"], env=_env(root), catch_exceptions=False)
211 assert result.exit_code == 0
212 assert "Removed" in result.output
213 result2 = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
214 assert "emotion:tense" not in result2.output
215
216 def test_remove_nonexistent_tag_is_idempotent(self, tmp_path: pathlib.Path) -> None:
217 """Removing a tag that doesn't exist exits 0 (idempotent) with not_found status."""
218 root, repo_id = _init_repo(tmp_path)
219 _make_commit(root, repo_id)
220 result = runner.invoke(cli, ["tag", "remove", "ghost:tag", "--json"], env=_env(root))
221 assert result.exit_code == 0
222 import json as _json
223 d = _json.loads(result.output)
224 assert d["status"] == "not_found"
225 assert d["removed_count"] == 0
226
227 def test_add_multiple_tags_same_commit(self, tmp_path: pathlib.Path) -> None:
228 root, repo_id = _init_repo(tmp_path)
229 _make_commit(root, repo_id)
230 runner.invoke(cli, ["tag", "add", "key:Am"], env=_env(root), catch_exceptions=False)
231 runner.invoke(cli, ["tag", "add", "tempo:120bpm"], env=_env(root), catch_exceptions=False)
232 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
233 assert "key:Am" in result.output
234 assert "tempo:120bpm" in result.output
235
236 def test_tag_on_invalid_ref_fails(self, tmp_path: pathlib.Path) -> None:
237 root, repo_id = _init_repo(tmp_path)
238 _make_commit(root, repo_id)
239 result = runner.invoke(cli, ["tag", "add", "emotion:sad", "deadbeef" * 8], env=_env(root))
240 assert result.exit_code != 0
241
242
243 # ---------------------------------------------------------------------------
244 # Security tests
245 # ---------------------------------------------------------------------------
246
247 class TestTagSecurity:
248 def test_tag_with_control_characters_sanitized_in_output(
249 self, tmp_path: pathlib.Path
250 ) -> None:
251 root, repo_id = _init_repo(tmp_path)
252 _make_commit(root, repo_id)
253 malicious = "emotion:\x1b[31mred\x1b[0m"
254 runner.invoke(cli, ["tag", "add", malicious], env=_env(root), catch_exceptions=False)
255 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
256 assert result.exit_code == 0
257 assert "\x1b" not in result.output
258
259
260 # ---------------------------------------------------------------------------
261 # Stress tests
262 # ---------------------------------------------------------------------------
263
264 class TestTagStress:
265 def test_many_tags_on_many_commits(self, tmp_path: pathlib.Path) -> None:
266 root, repo_id = _init_repo(tmp_path)
267 commit_ids = [_make_commit(root, repo_id, message=f"commit {i}") for i in range(30)]
268 from muse.core.store import TagRecord, write_tag, get_all_tags
269 tag_types = ["emotion:joyful", "section:chorus", "key:Am", "tempo:120bpm", "stage:master"]
270 for i, cid in enumerate(commit_ids):
271 write_tag(root, TagRecord(
272 tag_id=fake_id(f"tag-{i}"), repo_id=repo_id,
273 commit_id=cid, tag=tag_types[i % len(tag_types)],
274 ))
275 all_tags = get_all_tags(root, repo_id)
276 assert len(all_tags) == 30
277 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
278 assert result.exit_code == 0
279 for tag_type in tag_types:
280 assert tag_type in result.output
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 125 days ago