gabriel / muse public
test_cmd_tag.py python
305 lines 12.6 KB
Raw
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 7 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.commits import (
57 CommitRecord,
58 write_commit,
59 )
60 from muse.core.snapshots import (
61 SnapshotRecord,
62 write_snapshot,
63 )
64 from muse.core.ids import hash_snapshot, hash_commit
65
66 ref_file = ref_path(root, branch)
67 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
68 manifest: Manifest = {}
69 snap_id = hash_snapshot(manifest)
70 committed_at = datetime.datetime.now(datetime.timezone.utc)
71 commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [],
72 snapshot_id=snap_id, message=message,
73 committed_at_iso=committed_at.isoformat(),
74 )
75 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
76 write_commit(root, CommitRecord(
77 commit_id=commit_id, branch=branch,
78 snapshot_id=snap_id, message=message, committed_at=committed_at,
79 parent_commit_id=parent_id,
80 ))
81 ref_file.parent.mkdir(parents=True, exist_ok=True)
82 ref_file.write_text(commit_id, encoding="utf-8")
83 return commit_id
84
85
86 # ---------------------------------------------------------------------------
87 # Unit tests
88 # ---------------------------------------------------------------------------
89
90 class TestTagUnit:
91 def test_write_and_read_tag(self, tmp_path: pathlib.Path) -> None:
92 root, repo_id = _init_repo(tmp_path)
93 commit_id = _make_commit(root, repo_id)
94 from muse.core.tags import (
95 TagRecord,
96 get_tags_for_commit,
97 write_tag,
98 )
99 tag = TagRecord(tag_id=fake_id("tag"), repo_id=repo_id,
100 commit_id=commit_id, tag="emotion:joyful")
101 write_tag(root, tag)
102 tags = get_tags_for_commit(root, repo_id, commit_id)
103 assert len(tags) == 1
104 assert tags[0].tag == "emotion:joyful"
105
106 def test_delete_tag(self, tmp_path: pathlib.Path) -> None:
107 root, repo_id = _init_repo(tmp_path)
108 commit_id = _make_commit(root, repo_id)
109 from muse.core.tags import (
110 TagRecord,
111 delete_tag,
112 get_tags_for_commit,
113 write_tag,
114 )
115 tag_id = fake_id("tag")
116 write_tag(root, TagRecord(tag_id=tag_id, repo_id=repo_id,
117 commit_id=commit_id, tag="section:chorus"))
118 assert len(get_tags_for_commit(root, repo_id, commit_id)) == 1
119 assert delete_tag(root, repo_id, tag_id) is True
120 assert get_tags_for_commit(root, repo_id, commit_id) == []
121
122 def test_delete_nonexistent_tag_returns_false(self, tmp_path: pathlib.Path) -> None:
123 root, repo_id = _init_repo(tmp_path)
124 from muse.core.tags import delete_tag
125 assert delete_tag(root, repo_id, fake_id("tag")) is False
126
127 def test_get_all_tags_empty(self, tmp_path: pathlib.Path) -> None:
128 root, repo_id = _init_repo(tmp_path)
129 from muse.core.tags import get_all_tags
130 assert get_all_tags(root, repo_id) == []
131
132
133 # ---------------------------------------------------------------------------
134 # Content-addressed tag_id
135 # ---------------------------------------------------------------------------
136
137 class TestTagIdContentAddressed:
138 """tag_id must be sha256: of genesis content, not a UUID."""
139
140 def test_tag_id_is_sha256_prefixed(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.tags import compute_tag_id
144 tag_id = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="emotion:joyful")
145 assert tag_id.startswith("sha256:"), f"Expected sha256: prefix, got {tag_id!r}"
146 assert len(tag_id) == 71
147
148 def test_tag_id_is_deterministic(self, tmp_path: pathlib.Path) -> None:
149 root, repo_id = _init_repo(tmp_path)
150 commit_id = _make_commit(root, repo_id)
151 from muse.core.tags import compute_tag_id
152 id1 = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="v1.0")
153 id2 = compute_tag_id(repo_id=repo_id, commit_id=commit_id, tag="v1.0")
154 assert id1 == id2
155
156 def test_tag_id_differs_by_tag_name(self, tmp_path: pathlib.Path) -> None:
157 root, repo_id = _init_repo(tmp_path)
158 commit_id = _make_commit(root, repo_id)
159 from muse.core.tags import compute_tag_id
160 assert compute_tag_id(repo_id, commit_id, "v1.0") != compute_tag_id(repo_id, commit_id, "v2.0")
161
162 def test_tag_id_differs_by_commit(self, tmp_path: pathlib.Path) -> None:
163 root, repo_id = _init_repo(tmp_path)
164 c1 = _make_commit(root, repo_id, message="first")
165 c2 = _make_commit(root, repo_id, message="second")
166 from muse.core.tags import compute_tag_id
167 assert compute_tag_id(repo_id, c1, "v1.0") != compute_tag_id(repo_id, c2, "v1.0")
168
169 def test_tag_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None:
170 import re
171 root, repo_id = _init_repo(tmp_path)
172 commit_id = _make_commit(root, repo_id)
173 from muse.core.tags import compute_tag_id
174 tag_id = compute_tag_id(repo_id, commit_id, "release:1.0")
175 assert tag_id.startswith("sha256:")
176 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}$")
177 assert not uuid4_re.match(tag_id)
178
179 def test_cli_tag_add_returns_content_addressed_id(self, tmp_path: pathlib.Path) -> None:
180 root, repo_id = _init_repo(tmp_path)
181 commit_id = _make_commit(root, repo_id)
182 result = runner.invoke(cli, ["tag", "add", "v1.0", commit_id, "--json"], env=_env(root))
183 assert result.exit_code == 0
184 data = json.loads(result.output)
185 tag_id = data["tag_id"]
186 assert tag_id.startswith("sha256:"), f"Expected sha256: prefix, got {tag_id!r}"
187 assert len(tag_id) == 71
188
189 def test_write_tag_uses_content_addressed_id(self, tmp_path: pathlib.Path) -> None:
190 root, repo_id = _init_repo(tmp_path)
191 commit_id = _make_commit(root, repo_id)
192 from muse.core.tags import (
193 TagRecord,
194 compute_tag_id,
195 get_tags_for_commit,
196 write_tag,
197 )
198 expected_id = compute_tag_id(repo_id, commit_id, "emotion:sad")
199 tag = TagRecord(tag_id=expected_id, repo_id=repo_id, commit_id=commit_id, tag="emotion:sad")
200 write_tag(root, tag)
201 tags = get_tags_for_commit(root, repo_id, commit_id)
202 assert tags[0].tag_id == expected_id
203
204
205 # ---------------------------------------------------------------------------
206 # Integration tests
207 # ---------------------------------------------------------------------------
208
209 class TestTagIntegration:
210 def test_add_and_list_tag(self, tmp_path: pathlib.Path) -> None:
211 root, repo_id = _init_repo(tmp_path)
212 _make_commit(root, repo_id)
213 result = runner.invoke(cli, ["tag", "add", "emotion:joyful"], env=_env(root), catch_exceptions=False)
214 assert result.exit_code == 0
215 assert "Tagged" in result.output
216
217 result2 = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
218 assert "emotion:joyful" in result2.output
219
220 def test_list_tags_for_specific_commit(self, tmp_path: pathlib.Path) -> None:
221 root, repo_id = _init_repo(tmp_path)
222 commit_id = _make_commit(root, repo_id)
223 runner.invoke(cli, ["tag", "add", "section:verse"], env=_env(root), catch_exceptions=False)
224 result = runner.invoke(cli, ["tag", "list", short_id(commit_id)], env=_env(root), catch_exceptions=False)
225 assert "section:verse" in result.output
226
227 def test_remove_tag(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", "emotion:tense"], env=_env(root), catch_exceptions=False)
231 result = runner.invoke(cli, ["tag", "remove", "emotion:tense"], env=_env(root), catch_exceptions=False)
232 assert result.exit_code == 0
233 assert "Removed" in result.output
234 result2 = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
235 assert "emotion:tense" not in result2.output
236
237 def test_remove_nonexistent_tag_is_idempotent(self, tmp_path: pathlib.Path) -> None:
238 """Removing a tag that doesn't exist exits 0 (idempotent) with not_found status."""
239 root, repo_id = _init_repo(tmp_path)
240 _make_commit(root, repo_id)
241 result = runner.invoke(cli, ["tag", "remove", "ghost:tag", "--json"], env=_env(root))
242 assert result.exit_code == 0
243 import json as _json
244 d = _json.loads(result.output)
245 assert d["status"] == "not_found"
246 assert d["removed_count"] == 0
247
248 def test_add_multiple_tags_same_commit(self, tmp_path: pathlib.Path) -> None:
249 root, repo_id = _init_repo(tmp_path)
250 _make_commit(root, repo_id)
251 runner.invoke(cli, ["tag", "add", "key:Am"], env=_env(root), catch_exceptions=False)
252 runner.invoke(cli, ["tag", "add", "tempo:120bpm"], env=_env(root), catch_exceptions=False)
253 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
254 assert "key:Am" in result.output
255 assert "tempo:120bpm" in result.output
256
257 def test_tag_on_invalid_ref_fails(self, tmp_path: pathlib.Path) -> None:
258 root, repo_id = _init_repo(tmp_path)
259 _make_commit(root, repo_id)
260 result = runner.invoke(cli, ["tag", "add", "emotion:sad", "deadbeef" * 8], env=_env(root))
261 assert result.exit_code != 0
262
263
264 # ---------------------------------------------------------------------------
265 # Security tests
266 # ---------------------------------------------------------------------------
267
268 class TestTagSecurity:
269 def test_tag_with_control_characters_sanitized_in_output(
270 self, tmp_path: pathlib.Path
271 ) -> None:
272 root, repo_id = _init_repo(tmp_path)
273 _make_commit(root, repo_id)
274 malicious = "emotion:\x1b[31mred\x1b[0m"
275 runner.invoke(cli, ["tag", "add", malicious], env=_env(root), catch_exceptions=False)
276 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
277 assert result.exit_code == 0
278 assert "\x1b" not in result.output
279
280
281 # ---------------------------------------------------------------------------
282 # Stress tests
283 # ---------------------------------------------------------------------------
284
285 class TestTagStress:
286 def test_many_tags_on_many_commits(self, tmp_path: pathlib.Path) -> None:
287 root, repo_id = _init_repo(tmp_path)
288 commit_ids = [_make_commit(root, repo_id, message=f"commit {i}") for i in range(30)]
289 from muse.core.tags import (
290 TagRecord,
291 get_all_tags,
292 write_tag,
293 )
294 tag_types = ["emotion:joyful", "section:chorus", "key:Am", "tempo:120bpm", "stage:master"]
295 for i, cid in enumerate(commit_ids):
296 write_tag(root, TagRecord(
297 tag_id=fake_id(f"tag-{i}"), repo_id=repo_id,
298 commit_id=cid, tag=tag_types[i % len(tag_types)],
299 ))
300 all_tags = get_all_tags(root, repo_id)
301 assert len(all_tags) == 30
302 result = runner.invoke(cli, ["tag", "list"], env=_env(root), catch_exceptions=False)
303 assert result.exit_code == 0
304 for tag_type in tag_types:
305 assert tag_type in result.output
File History 1 commit
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 7 days ago