cgcardona / muse public
test_muse_release.py python
565 lines 18.5 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse release`` — export a tagged commit as release artifacts.
2
3 Verifies:
4 - ``build_release`` writes a release-manifest.json when called with no flags.
5 - ``build_release --render-midi`` produces a zip archive of all MIDI files.
6 - ``build_release --render-audio`` copies the MIDI as an audio stub.
7 - ``build_release --export-stems`` produces per-track audio stubs.
8 - ``build_release`` raises ``ValueError`` when no MIDI files exist in snapshot.
9 - ``build_release`` raises ``StorpheusReleaseUnavailableError`` when Storpheus
10 is down and audio rendering is requested.
11 - ``_resolve_tag_to_commit`` resolves a tag to the most recent commit.
12 - ``_resolve_tag_to_commit`` falls back to prefix lookup when tag not found.
13 - ``_release_async`` (regression): resolves tag, fetches manifest, delegates.
14 - Boundary seal (AST): ``from __future__ import annotations`` present.
15 """
16 from __future__ import annotations
17
18 import ast
19 import datetime
20 import json
21 import pathlib
22 import uuid
23 import zipfile
24 from collections.abc import AsyncGenerator
25 from unittest.mock import patch
26
27 import pytest
28 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
29
30 from maestro.db.database import Base
31 from maestro.muse_cli import models as cli_models # noqa: F401 — register tables
32 from maestro.muse_cli.models import MuseCliCommit, MuseCliObject, MuseCliSnapshot, MuseCliTag
33 from maestro.services.muse_release import (
34 ReleaseAudioFormat,
35 ReleaseResult,
36 StorpheusReleaseUnavailableError,
37 _collect_midi_paths,
38 _sha256_file,
39 build_release,
40 )
41
42
43 # ---------------------------------------------------------------------------
44 # Fixtures
45 # ---------------------------------------------------------------------------
46
47
48 @pytest.fixture
49 async def async_session() -> AsyncGenerator[AsyncSession, None]:
50 """In-memory SQLite session with all CLI tables created."""
51 engine = create_async_engine("sqlite+aiosqlite:///:memory:")
52 async with engine.begin() as conn:
53 await conn.run_sync(Base.metadata.create_all)
54 Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
55 async with Session() as session:
56 yield session
57 await engine.dispose()
58
59
60 @pytest.fixture
61 def repo_root(tmp_path: pathlib.Path) -> pathlib.Path:
62 """Create a minimal Muse repo structure under *tmp_path*."""
63 muse_dir = tmp_path / ".muse"
64 muse_dir.mkdir()
65 (muse_dir / "HEAD").write_text("refs/heads/main")
66 refs_dir = muse_dir / "refs" / "heads"
67 refs_dir.mkdir(parents=True)
68 return tmp_path
69
70
71 @pytest.fixture
72 def repo_id() -> str:
73 return str(uuid.uuid4())
74
75
76 @pytest.fixture
77 def write_repo_json(repo_root: pathlib.Path, repo_id: str) -> None:
78 """Write .muse/repo.json with a stable repo_id."""
79 (repo_root / ".muse" / "repo.json").write_text(json.dumps({"repo_id": repo_id}))
80
81
82 @pytest.fixture
83 def midi_repo(repo_root: pathlib.Path) -> dict[str, pathlib.Path]:
84 """Create a muse-work/ directory with two MIDI stub files.
85
86 Returns a dict mapping relative path strings to absolute Path objects.
87 """
88 workdir = repo_root / "muse-work"
89 workdir.mkdir()
90 paths: dict[str, pathlib.Path] = {}
91 for name in ("piano.mid", "bass.mid"):
92 p = workdir / name
93 p.write_bytes(b"MIDI")
94 paths[name] = p
95 return paths
96
97
98 async def _insert_commit_with_tag(
99 session: AsyncSession,
100 repo_id: str,
101 repo_root: pathlib.Path,
102 tag: str,
103 manifest: dict[str, str] | None = None,
104 commit_id_char: str = "b",
105 ) -> str:
106 """Insert a commit + snapshot + tag; return the commit_id."""
107 object_id = "c" * 64
108 snapshot_id = ("s" + commit_id_char) * 32
109 commit_id = commit_id_char * 64
110
111 if not manifest:
112 manifest = {"piano.mid": object_id}
113
114 session.add(MuseCliObject(object_id=object_id, size_bytes=4))
115 session.add(MuseCliSnapshot(snapshot_id=snapshot_id, manifest=manifest))
116 await session.flush()
117
118 committed_at = datetime.datetime.now(datetime.timezone.utc)
119 session.add(
120 MuseCliCommit(
121 commit_id=commit_id,
122 repo_id=repo_id,
123 branch="main",
124 parent_commit_id=None,
125 parent2_commit_id=None,
126 snapshot_id=snapshot_id,
127 message="tagged commit",
128 author="",
129 committed_at=committed_at,
130 )
131 )
132 await session.flush()
133
134 session.add(MuseCliTag(repo_id=repo_id, commit_id=commit_id, tag=tag))
135 await session.flush()
136
137 # Update HEAD pointer
138 ref_path = repo_root / ".muse" / "refs" / "heads" / "main"
139 ref_path.write_text(commit_id)
140 return commit_id
141
142
143 # ---------------------------------------------------------------------------
144 # Unit tests — service layer (build_release)
145 # ---------------------------------------------------------------------------
146
147
148 def test_build_release_writes_manifest_only(
149 repo_root: pathlib.Path,
150 midi_repo: dict[str, pathlib.Path],
151 tmp_path: pathlib.Path,
152 ) -> None:
153 """build_release writes release-manifest.json even when no flags are set."""
154 manifest = {name: "c" * 64 for name in midi_repo}
155 output_dir = tmp_path / "releases" / "v1.0"
156
157 with patch(
158 "maestro.services.muse_release._check_storpheus_reachable"
159 ): # not called — no audio flags
160 result = build_release(
161 tag="v1.0",
162 commit_id="b" * 64,
163 manifest=manifest,
164 root=repo_root,
165 output_dir=output_dir,
166 render_audio=False,
167 render_midi=False,
168 export_stems=False,
169 )
170
171 assert result.manifest_path.exists()
172 data = json.loads(result.manifest_path.read_text())
173 assert data["tag"] == "v1.0"
174 assert data["commit_id"] == "b" * 64
175 assert data["commit_short"] == "b" * 8
176 assert "released_at" in data
177 assert isinstance(data["files"], list)
178
179
180 def test_build_release_render_midi_produces_zip(
181 repo_root: pathlib.Path,
182 midi_repo: dict[str, pathlib.Path],
183 tmp_path: pathlib.Path,
184 ) -> None:
185 """build_release --render-midi produces a zip containing all MIDI files."""
186 manifest = {name: "c" * 64 for name in midi_repo}
187 output_dir = tmp_path / "releases" / "v1.0"
188
189 result = build_release(
190 tag="v1.0",
191 commit_id="b" * 64,
192 manifest=manifest,
193 root=repo_root,
194 output_dir=output_dir,
195 render_midi=True,
196 )
197
198 bundle_path = output_dir / "midi" / "midi-bundle.zip"
199 assert bundle_path.exists()
200 assert any(a.role == "midi-bundle" for a in result.artifacts)
201
202 with zipfile.ZipFile(bundle_path) as zf:
203 names = zf.namelist()
204 assert "piano.mid" in names
205 assert "bass.mid" in names
206
207
208 def test_build_release_render_audio_produces_stub(
209 repo_root: pathlib.Path,
210 midi_repo: dict[str, pathlib.Path],
211 tmp_path: pathlib.Path,
212 ) -> None:
213 """build_release --render-audio copies MIDI as audio stub when /render not deployed."""
214 manifest = {name: "c" * 64 for name in midi_repo}
215 output_dir = tmp_path / "releases" / "v1.0"
216
217 with patch(
218 "maestro.services.muse_release._check_storpheus_reachable"
219 ):
220 result = build_release(
221 tag="v1.0",
222 commit_id="b" * 64,
223 manifest=manifest,
224 root=repo_root,
225 output_dir=output_dir,
226 render_audio=True,
227 )
228
229 audio_artifact = next(a for a in result.artifacts if a.role == "audio")
230 assert audio_artifact.path.exists()
231 assert audio_artifact.path.suffix == ".wav"
232 assert result.stubbed is True
233
234
235 def test_build_release_export_stems_produces_per_track_files(
236 repo_root: pathlib.Path,
237 midi_repo: dict[str, pathlib.Path],
238 tmp_path: pathlib.Path,
239 ) -> None:
240 """build_release --export-stems writes one audio file per MIDI track."""
241 manifest = {name: "c" * 64 for name in midi_repo}
242 output_dir = tmp_path / "releases" / "v1.0"
243
244 with patch(
245 "maestro.services.muse_release._check_storpheus_reachable"
246 ):
247 result = build_release(
248 tag="v1.0",
249 commit_id="b" * 64,
250 manifest=manifest,
251 root=repo_root,
252 output_dir=output_dir,
253 export_stems=True,
254 audio_format=ReleaseAudioFormat.FLAC,
255 )
256
257 stem_artifacts = [a for a in result.artifacts if a.role == "stem"]
258 assert len(stem_artifacts) == 2
259 for a in stem_artifacts:
260 assert a.path.suffix == ".flac"
261 assert a.path.exists()
262
263
264 def test_build_release_manifest_contains_checksums(
265 repo_root: pathlib.Path,
266 midi_repo: dict[str, pathlib.Path],
267 tmp_path: pathlib.Path,
268 ) -> None:
269 """release-manifest.json includes sha256 checksums for every artifact."""
270 manifest = {name: "c" * 64 for name in midi_repo}
271 output_dir = tmp_path / "releases" / "v1.0"
272
273 result = build_release(
274 tag="v1.0",
275 commit_id="b" * 64,
276 manifest=manifest,
277 root=repo_root,
278 output_dir=output_dir,
279 render_midi=True,
280 )
281
282 data = json.loads(result.manifest_path.read_text())
283 for file_entry in data["files"]:
284 assert "sha256" in file_entry
285 assert len(file_entry["sha256"]) == 64
286 assert "size_bytes" in file_entry
287 assert "role" in file_entry
288
289
290 def test_build_release_raises_when_no_midi_files(
291 repo_root: pathlib.Path,
292 tmp_path: pathlib.Path,
293 ) -> None:
294 """build_release raises ValueError when no MIDI files exist in snapshot."""
295 workdir = repo_root / "muse-work"
296 workdir.mkdir()
297 (workdir / "notes.json").write_text("{}") # not a MIDI file
298
299 manifest = {"notes.json": "c" * 64}
300 output_dir = tmp_path / "releases" / "v1.0"
301
302 with pytest.raises(ValueError, match="No MIDI files found"):
303 build_release(
304 tag="v1.0",
305 commit_id="b" * 64,
306 manifest=manifest,
307 root=repo_root,
308 output_dir=output_dir,
309 render_audio=True,
310 )
311
312
313 def test_build_release_raises_when_storpheus_unreachable(
314 repo_root: pathlib.Path,
315 midi_repo: dict[str, pathlib.Path],
316 tmp_path: pathlib.Path,
317 ) -> None:
318 """build_release raises StorpheusReleaseUnavailableError when Storpheus is down."""
319 manifest = {name: "c" * 64 for name in midi_repo}
320 output_dir = tmp_path / "releases" / "v1.0"
321
322 with patch(
323 "maestro.services.muse_release._check_storpheus_reachable",
324 side_effect=StorpheusReleaseUnavailableError("Storpheus is down"),
325 ):
326 with pytest.raises(StorpheusReleaseUnavailableError):
327 build_release(
328 tag="v1.0",
329 commit_id="b" * 64,
330 manifest=manifest,
331 root=repo_root,
332 output_dir=output_dir,
333 render_audio=True,
334 )
335
336
337 # ---------------------------------------------------------------------------
338 # Unit tests — SHA-256 helper
339 # ---------------------------------------------------------------------------
340
341
342 def test_sha256_file_matches_known_digest(tmp_path: pathlib.Path) -> None:
343 """_sha256_file computes the correct SHA-256 for a known byte sequence."""
344 import hashlib
345
346 content = b"MIDI content for hashing"
347 p = tmp_path / "test.mid"
348 p.write_bytes(content)
349
350 expected = hashlib.sha256(content).hexdigest()
351 assert _sha256_file(p) == expected
352
353
354 # ---------------------------------------------------------------------------
355 # Unit tests — _collect_midi_paths
356 # ---------------------------------------------------------------------------
357
358
359 def test_collect_midi_paths_filters_by_track(
360 repo_root: pathlib.Path,
361 midi_repo: dict[str, pathlib.Path],
362 ) -> None:
363 """_collect_midi_paths returns only paths matching the track filter."""
364 manifest = {name: "c" * 64 for name in midi_repo}
365 paths, skipped = _collect_midi_paths(manifest, repo_root, track="piano")
366
367 assert len(paths) == 1
368 assert paths[0].name == "piano.mid"
369 assert skipped == 1
370
371
372 def test_collect_midi_paths_skips_missing_files(
373 repo_root: pathlib.Path,
374 ) -> None:
375 """_collect_midi_paths counts missing files in skipped_count."""
376 workdir = repo_root / "muse-work"
377 workdir.mkdir()
378 manifest = {"missing.mid": "c" * 64} # file does not exist on disk
379
380 paths, skipped = _collect_midi_paths(manifest, repo_root)
381 assert paths == []
382 assert skipped == 1
383
384
385 # ---------------------------------------------------------------------------
386 # Integration tests — _resolve_tag_to_commit
387 # ---------------------------------------------------------------------------
388
389
390 @pytest.mark.anyio
391 async def test_resolve_tag_to_commit_finds_tagged_commit(
392 async_session: AsyncSession,
393 repo_root: pathlib.Path,
394 repo_id: str,
395 write_repo_json: None,
396 ) -> None:
397 """_resolve_tag_to_commit resolves a tag string to the correct commit ID."""
398 from maestro.muse_cli.commands.release import _resolve_tag_to_commit
399
400 commit_id = await _insert_commit_with_tag(async_session, repo_id, repo_root, "v1.0")
401
402 resolved = await _resolve_tag_to_commit(async_session, repo_root, "v1.0")
403 assert resolved == commit_id
404
405
406 @pytest.mark.anyio
407 async def test_resolve_tag_to_commit_uses_most_recent_when_ambiguous(
408 async_session: AsyncSession,
409 repo_root: pathlib.Path,
410 repo_id: str,
411 write_repo_json: None,
412 ) -> None:
413 """When multiple commits share a tag, the most recently committed is returned."""
414 from maestro.muse_cli.commands.release import _resolve_tag_to_commit
415
416 object_id = "c" * 64
417 snap1_id = "s1" * 32
418 snap2_id = "s2" * 32
419 cid1 = "1" * 64
420 cid2 = "2" * 64
421
422 async_session.add(MuseCliObject(object_id=object_id, size_bytes=4))
423 async_session.add(MuseCliSnapshot(snapshot_id=snap1_id, manifest={"a.mid": object_id}))
424 async_session.add(MuseCliSnapshot(snapshot_id=snap2_id, manifest={"b.mid": object_id}))
425 await async_session.flush()
426
427 t1 = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
428 t2 = datetime.datetime(2024, 6, 1, tzinfo=datetime.timezone.utc) # more recent
429
430 async_session.add(
431 MuseCliCommit(
432 commit_id=cid1, repo_id=repo_id, branch="main",
433 parent_commit_id=None, parent2_commit_id=None,
434 snapshot_id=snap1_id, message="old", author="", committed_at=t1,
435 )
436 )
437 async_session.add(
438 MuseCliCommit(
439 commit_id=cid2, repo_id=repo_id, branch="main",
440 parent_commit_id=cid1, parent2_commit_id=None,
441 snapshot_id=snap2_id, message="newer", author="", committed_at=t2,
442 )
443 )
444 await async_session.flush()
445
446 async_session.add(MuseCliTag(repo_id=repo_id, commit_id=cid1, tag="v1.0"))
447 async_session.add(MuseCliTag(repo_id=repo_id, commit_id=cid2, tag="v1.0"))
448 await async_session.flush()
449
450 resolved = await _resolve_tag_to_commit(async_session, repo_root, "v1.0")
451 assert resolved == cid2
452
453
454 @pytest.mark.anyio
455 async def test_resolve_tag_to_commit_falls_back_to_prefix(
456 async_session: AsyncSession,
457 repo_root: pathlib.Path,
458 repo_id: str,
459 write_repo_json: None,
460 ) -> None:
461 """_resolve_tag_to_commit falls back to commit prefix lookup when tag absent."""
462 from maestro.muse_cli.commands.release import _resolve_tag_to_commit
463
464 commit_id = await _insert_commit_with_tag(async_session, repo_id, repo_root, "v2.0")
465 # Use the commit ID prefix directly (not the tag)
466 resolved = await _resolve_tag_to_commit(async_session, repo_root, commit_id[:8])
467 assert resolved == commit_id
468
469
470 # ---------------------------------------------------------------------------
471 # Regression test — test_release_resolves_tag_and_exports_manifest
472 # ---------------------------------------------------------------------------
473
474
475 @pytest.mark.anyio
476 async def test_release_resolves_tag_and_exports_manifest(
477 async_session: AsyncSession,
478 repo_root: pathlib.Path,
479 repo_id: str,
480 write_repo_json: None,
481 midi_repo: dict[str, pathlib.Path],
482 tmp_path: pathlib.Path,
483 ) -> None:
484 """Regression: _release_async resolves tag, fetches manifest, writes manifest.json.
485
486 This is the primary acceptance criterion: a producer runs
487 'muse release v1.0' and receives a release-manifest.json pinning the
488 tagged snapshot.
489 """
490 from maestro.muse_cli.commands.release import _release_async
491
492 manifest = {name: "c" * 64 for name in midi_repo}
493 commit_id = await _insert_commit_with_tag(
494 async_session, repo_id, repo_root, "v1.0", manifest=manifest
495 )
496
497 output_dir = tmp_path / "releases" / "v1.0"
498
499 with patch("maestro.config.settings") as mock_settings:
500 mock_settings.storpheus_base_url = "http://storpheus:10002"
501 result = await _release_async(
502 tag="v1.0",
503 audio_format=ReleaseAudioFormat.WAV,
504 output_dir=output_dir,
505 render_audio=False,
506 render_midi=True,
507 export_stems=False,
508 root=repo_root,
509 session=async_session,
510 )
511
512 assert result.tag == "v1.0"
513 assert result.commit_id == commit_id
514 assert result.manifest_path.exists()
515
516 data = json.loads(result.manifest_path.read_text())
517 assert data["tag"] == "v1.0"
518 assert data["commit_id"] == commit_id
519
520 # MIDI bundle should be present
521 bundle_artifact = next((a for a in result.artifacts if a.role == "midi-bundle"), None)
522 assert bundle_artifact is not None
523 assert bundle_artifact.path.exists()
524
525 with zipfile.ZipFile(bundle_artifact.path) as zf:
526 names = zf.namelist()
527 for midi_name in midi_repo:
528 assert midi_name in names
529
530
531 # ---------------------------------------------------------------------------
532 # Boundary seal
533 # ---------------------------------------------------------------------------
534
535
536 def test_future_annotations_in_service() -> None:
537 """``from __future__ import annotations`` is present in muse_release.py."""
538 import maestro.services.muse_release as mod
539
540 src = pathlib.Path(mod.__file__).read_text()
541 tree = ast.parse(src)
542 future_imports = [
543 node
544 for node in ast.walk(tree)
545 if isinstance(node, ast.ImportFrom)
546 and node.module == "__future__"
547 and any(alias.name == "annotations" for alias in node.names)
548 ]
549 assert future_imports, "from __future__ import annotations missing in muse_release.py"
550
551
552 def test_future_annotations_in_command() -> None:
553 """``from __future__ import annotations`` is present in commands/release.py."""
554 import maestro.muse_cli.commands.release as mod
555
556 src = pathlib.Path(mod.__file__).read_text()
557 tree = ast.parse(src)
558 future_imports = [
559 node
560 for node in ast.walk(tree)
561 if isinstance(node, ast.ImportFrom)
562 and node.module == "__future__"
563 and any(alias.name == "annotations" for alias in node.names)
564 ]
565 assert future_imports, "from __future__ import annotations missing in commands/release.py"