cgcardona / muse public
test_export.py python
503 lines 17.1 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse export`` command and export_engine module.
2
3 Test matrix:
4 - ``test_export_midi_writes_valid_midi_file``
5 - ``test_export_json_outputs_full_note_structure``
6 - ``test_export_musicxml_produces_valid_xml``
7 - ``test_export_track_scoped_midi``
8 - ``test_export_section_scoped_midi``
9 - ``test_export_split_tracks_creates_one_file_per_track``
10 - ``test_export_wav_raises_clear_error_when_storpheus_unavailable``
11 - ``test_export_no_commits_exits_user_error``
12 - ``test_export_commit_prefix_resolution``
13 - ``test_filter_manifest_track``
14 - ``test_filter_manifest_section``
15 - ``test_filter_manifest_both``
16 - ``test_filter_manifest_no_filter``
17 """
18 from __future__ import annotations
19
20 import json
21 import pathlib
22 import uuid
23 from typing import Any
24 from unittest.mock import MagicMock, patch
25
26 import pytest
27 import pytest_asyncio
28 from sqlalchemy.ext.asyncio import AsyncSession
29 from typer.testing import CliRunner
30
31 from maestro.muse_cli.app import cli
32 from maestro.muse_cli.commands.export import _export_async, _default_output_path
33 from maestro.muse_cli.export_engine import (
34 ExportFormat,
35 MuseExportOptions,
36 StorpheusUnavailableError,
37 export_json,
38 export_midi,
39 export_musicxml,
40 export_abc,
41 export_wav,
42 export_snapshot,
43 filter_manifest,
44 resolve_commit_id,
45 _midi_note_to_abc,
46 _midi_note_to_step_octave,
47 )
48 from maestro.muse_cli.snapshot import hash_file
49
50 runner = CliRunner()
51
52
53 # ---------------------------------------------------------------------------
54 # Fixtures / helpers
55 # ---------------------------------------------------------------------------
56
57
58 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
59 """Create a minimal .muse/ layout for CLI tests."""
60 rid = repo_id or str(uuid.uuid4())
61 muse = root / ".muse"
62 (muse / "refs" / "heads").mkdir(parents=True)
63 (muse / "repo.json").write_text(
64 json.dumps({"repo_id": rid, "schema_version": "1"})
65 )
66 (muse / "HEAD").write_text("refs/heads/main")
67 (muse / "refs" / "heads" / "main").write_text("")
68 return rid
69
70
71 def _set_head(root: pathlib.Path, commit_id: str) -> None:
72 """Point the HEAD of the main branch at commit_id."""
73 ref_path = root / ".muse" / "refs" / "heads" / "main"
74 ref_path.write_text(commit_id)
75
76
77 def _make_minimal_midi() -> bytes:
78 """Return a minimal, well-formed MIDI file (single note, type 0).
79
80 Format: Type 0, 1 track, 480 ticks/beat.
81 Track: Note On C4 (velocity 64) at tick 0, Note Off at tick 480.
82 """
83 # MIDI header: MThd + length(6) + format(0) + tracks(1) + tpb(480)
84 header = b"MThd\x00\x00\x00\x06\x00\x00\x00\x01\x01\xe0"
85 # Track: Note On C4, wait 480 ticks, Note Off C4, end-of-track
86 track_data = (
87 b"\x00\x90\x3c\x40" # delta=0, Note On ch=0, pitch=60, vel=64
88 b"\x81\x60\x80\x3c\x00" # delta=480 (var-len), Note Off ch=0, pitch=60, vel=0
89 b"\x00\xff\x2f\x00" # delta=0, End of Track
90 )
91 track_len = len(track_data).to_bytes(4, "big")
92 return header + b"MTrk" + track_len + track_data
93
94
95 def _make_manifest_with_midi(
96 tmp_path: pathlib.Path,
97 filenames: list[str] | None = None,
98 ) -> dict[str, str]:
99 """Write MIDI files to muse-work/ and return a manifest dict."""
100 workdir = tmp_path / "muse-work"
101 workdir.mkdir(exist_ok=True)
102 filenames = filenames or ["beat.mid"]
103 midi_bytes = _make_minimal_midi()
104 manifest: dict[str, str] = {}
105 for name in filenames:
106 p = workdir / name
107 p.write_bytes(midi_bytes)
108 manifest[name] = hash_file(p)
109 return manifest
110
111
112 # ---------------------------------------------------------------------------
113 # Unit tests — filter_manifest
114 # ---------------------------------------------------------------------------
115
116
117 def test_filter_manifest_no_filter() -> None:
118 """filter_manifest returns all entries when no filters are given."""
119 m = {"tracks/piano/take1.mid": "aaa", "tracks/bass/take1.mid": "bbb"}
120 result = filter_manifest(m, track=None, section=None)
121 assert result == m
122
123
124 def test_filter_manifest_track() -> None:
125 """filter_manifest keeps only entries matching the track substring."""
126 m = {
127 "tracks/piano/take1.mid": "aaa",
128 "tracks/bass/take1.mid": "bbb",
129 "tracks/piano/take2.mid": "ccc",
130 }
131 result = filter_manifest(m, track="piano", section=None)
132 assert set(result.keys()) == {"tracks/piano/take1.mid", "tracks/piano/take2.mid"}
133
134
135 def test_filter_manifest_section() -> None:
136 """filter_manifest keeps only entries matching the section substring."""
137 m = {
138 "chorus/piano.mid": "aaa",
139 "verse/piano.mid": "bbb",
140 "chorus/bass.mid": "ccc",
141 }
142 result = filter_manifest(m, track=None, section="chorus")
143 assert set(result.keys()) == {"chorus/piano.mid", "chorus/bass.mid"}
144
145
146 def test_filter_manifest_both() -> None:
147 """filter_manifest applies both track and section filters (AND semantics)."""
148 m = {
149 "chorus/piano.mid": "aaa",
150 "verse/piano.mid": "bbb",
151 "chorus/bass.mid": "ccc",
152 }
153 result = filter_manifest(m, track="piano", section="chorus")
154 assert set(result.keys()) == {"chorus/piano.mid"}
155
156
157 # ---------------------------------------------------------------------------
158 # Unit tests — export_midi
159 # ---------------------------------------------------------------------------
160
161
162 def test_export_midi_writes_valid_midi_file(tmp_path: pathlib.Path) -> None:
163 """export_midi copies a MIDI file to the output path."""
164 manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"])
165 out = tmp_path / "exports" / "out.mid"
166 opts = MuseExportOptions(
167 format=ExportFormat.MIDI,
168 commit_id="abc123",
169 output_path=out,
170 )
171
172 result = export_midi(manifest, tmp_path, opts)
173
174 assert result.paths_written == [out]
175 assert out.exists()
176 assert out.read_bytes() == _make_minimal_midi()
177
178
179 def test_export_track_scoped_midi(tmp_path: pathlib.Path) -> None:
180 """export_midi respects the track filter: only piano files are exported."""
181 (tmp_path / "muse-work").mkdir()
182 midi_bytes = _make_minimal_midi()
183 (tmp_path / "muse-work" / "piano.mid").write_bytes(midi_bytes)
184 (tmp_path / "muse-work" / "bass.mid").write_bytes(midi_bytes)
185 manifest = {
186 "piano.mid": hash_file(tmp_path / "muse-work" / "piano.mid"),
187 "bass.mid": hash_file(tmp_path / "muse-work" / "bass.mid"),
188 }
189
190 filtered = filter_manifest(manifest, track="piano", section=None)
191 out = tmp_path / "exports" / "out.mid"
192 opts = MuseExportOptions(
193 format=ExportFormat.MIDI,
194 commit_id="abc123",
195 output_path=out,
196 )
197 result = export_midi(filtered, tmp_path, opts)
198
199 assert len(result.paths_written) == 1
200 assert result.paths_written[0].name == "out.mid"
201
202
203 def test_export_section_scoped_midi(tmp_path: pathlib.Path) -> None:
204 """export_midi respects the section filter: only chorus files are exported."""
205 workdir = tmp_path / "muse-work"
206 (workdir / "chorus").mkdir(parents=True)
207 (workdir / "verse").mkdir(parents=True)
208 midi_bytes = _make_minimal_midi()
209 (workdir / "chorus" / "piano.mid").write_bytes(midi_bytes)
210 (workdir / "verse" / "piano.mid").write_bytes(midi_bytes)
211 manifest = {
212 "chorus/piano.mid": hash_file(workdir / "chorus" / "piano.mid"),
213 "verse/piano.mid": hash_file(workdir / "verse" / "piano.mid"),
214 }
215
216 filtered = filter_manifest(manifest, track=None, section="chorus")
217 out_dir = tmp_path / "exports"
218 opts = MuseExportOptions(
219 format=ExportFormat.MIDI,
220 commit_id="abc123",
221 output_path=out_dir,
222 split_tracks=True,
223 )
224 result = export_midi(filtered, tmp_path, opts)
225
226 assert len(result.paths_written) == 1
227 assert "chorus" not in result.paths_written[0].name # stem is "piano"
228 assert result.paths_written[0].name == "piano.mid"
229
230
231 def test_export_split_tracks_creates_one_file_per_track(tmp_path: pathlib.Path) -> None:
232 """--split-tracks writes one .mid file per MIDI entry in the manifest."""
233 manifest = _make_manifest_with_midi(tmp_path, ["drums.mid", "keys.mid", "bass.mid"])
234 out_dir = tmp_path / "exports"
235 opts = MuseExportOptions(
236 format=ExportFormat.MIDI,
237 commit_id="abc123",
238 output_path=out_dir,
239 split_tracks=True,
240 )
241
242 result = export_midi(manifest, tmp_path, opts)
243
244 assert len(result.paths_written) == 3
245 stems = {p.stem for p in result.paths_written}
246 assert stems == {"drums", "keys", "bass"}
247
248
249 # ---------------------------------------------------------------------------
250 # Unit tests — export_json
251 # ---------------------------------------------------------------------------
252
253
254 def test_export_json_outputs_full_note_structure(tmp_path: pathlib.Path) -> None:
255 """export_json writes a JSON file with commit_id, exported_at, and files array."""
256 manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"])
257 out = tmp_path / "exports" / "out.json"
258 commit_id = "deadbeef" * 8 # 64-char hex
259 opts = MuseExportOptions(
260 format=ExportFormat.JSON,
261 commit_id=commit_id,
262 output_path=out,
263 )
264
265 result = export_json(manifest, tmp_path, opts)
266
267 assert result.paths_written == [out]
268 data = json.loads(out.read_text())
269 assert data["commit_id"] == commit_id
270 assert "exported_at" in data
271 assert isinstance(data["files"], list)
272 assert len(data["files"]) == 1
273 assert data["files"][0]["path"] == "beat.mid"
274 assert "object_id" in data["files"][0]
275 assert data["files"][0]["exists_in_workdir"] is True
276
277
278 # ---------------------------------------------------------------------------
279 # Unit tests — export_musicxml
280 # ---------------------------------------------------------------------------
281
282
283 def test_export_musicxml_produces_valid_xml(tmp_path: pathlib.Path) -> None:
284 """export_musicxml writes a well-formed MusicXML file."""
285 manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"])
286 out = tmp_path / "exports" / "out.xml"
287 opts = MuseExportOptions(
288 format=ExportFormat.MUSICXML,
289 commit_id="abc123",
290 output_path=out,
291 )
292
293 result = export_musicxml(manifest, tmp_path, opts)
294
295 assert result.paths_written == [out]
296 content = out.read_text(encoding="utf-8")
297 assert '<?xml version="1.0"' in content
298 assert "<score-partwise" in content
299 assert "<part-list" in content
300
301
302 # ---------------------------------------------------------------------------
303 # Unit tests — WAV (Storpheus unavailable)
304 # ---------------------------------------------------------------------------
305
306
307 def test_export_wav_raises_clear_error_when_storpheus_unavailable(
308 tmp_path: pathlib.Path,
309 ) -> None:
310 """export_wav raises StorpheusUnavailableError when Storpheus is not reachable."""
311 manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"])
312 out = tmp_path / "exports" / "out.wav"
313 opts = MuseExportOptions(
314 format=ExportFormat.WAV,
315 commit_id="abc123",
316 output_path=out,
317 )
318
319 with patch("maestro.muse_cli.export_engine.httpx.Client") as mock_client_cls:
320 mock_client_cls.return_value.__enter__.side_effect = ConnectionRefusedError(
321 "Connection refused"
322 )
323 with pytest.raises(StorpheusUnavailableError) as exc_info:
324 export_wav(manifest, tmp_path, opts, storpheus_url="http://localhost:10002")
325
326 assert "not reachable" in str(exc_info.value).lower() or "storpheus" in str(exc_info.value).lower()
327
328
329 def test_export_wav_non_200_raises_unavailable(tmp_path: pathlib.Path) -> None:
330 """export_wav raises StorpheusUnavailableError on non-200 health response."""
331 manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"])
332 out = tmp_path / "exports" / "out.wav"
333 opts = MuseExportOptions(
334 format=ExportFormat.WAV,
335 commit_id="abc123",
336 output_path=out,
337 )
338
339 mock_resp = MagicMock()
340 mock_resp.status_code = 503
341
342 with patch("maestro.muse_cli.export_engine.httpx.Client") as mock_client_cls:
343 mock_client_cls.return_value.__enter__.return_value.get.return_value = mock_resp
344 with pytest.raises(StorpheusUnavailableError):
345 export_wav(manifest, tmp_path, opts, storpheus_url="http://localhost:10002")
346
347
348 # ---------------------------------------------------------------------------
349 # Unit tests — CLI integration
350 # ---------------------------------------------------------------------------
351
352
353 def test_export_no_commits_exits_user_error(tmp_path: pathlib.Path) -> None:
354 """``muse export --format json`` exits 1 when there are no commits (HEAD empty)."""
355 _init_muse_repo(tmp_path) # HEAD ref file is empty
356
357 with patch.dict("os.environ", {"MUSE_REPO_ROOT": str(tmp_path)}):
358 result = runner.invoke(cli, ["export", "--format", "json"])
359
360 # Exit code 1 (USER_ERROR) expected because HEAD is empty.
361 assert result.exit_code != 0
362
363
364 def test_export_cli_json_format(tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession) -> None:
365 """``muse export --format json`` writes a JSON file to the default path."""
366 import asyncio
367 from maestro.muse_cli.commands.commit import _commit_async
368
369 _init_muse_repo(tmp_path)
370 workdir = tmp_path / "muse-work"
371 workdir.mkdir()
372 (workdir / "beat.mid").write_bytes(_make_minimal_midi())
373
374 commit_id = asyncio.get_event_loop().run_until_complete(
375 _commit_async(
376 message="test commit",
377 root=tmp_path,
378 session=muse_cli_db_session,
379 )
380 )
381 _set_head(tmp_path, commit_id)
382
383 # Use _export_async directly to avoid the DB session bootstrapping in the CLI.
384 async def _run() -> None:
385 result = await _export_async(
386 commit_ref=None,
387 fmt=ExportFormat.JSON,
388 output=tmp_path / "out.json",
389 track=None,
390 section=None,
391 split_tracks=False,
392 root=tmp_path,
393 session=muse_cli_db_session,
394 )
395 assert result.paths_written
396 data = json.loads((tmp_path / "out.json").read_text())
397 assert data["commit_id"] == commit_id
398 assert len(data["files"]) == 1
399
400 asyncio.get_event_loop().run_until_complete(_run())
401
402
403 @pytest.mark.anyio
404 async def test_export_commit_prefix_resolution(
405 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
406 ) -> None:
407 """_export_async resolves a short commit prefix to the full commit ID."""
408 from maestro.muse_cli.commands.commit import _commit_async
409
410 _init_muse_repo(tmp_path)
411 workdir = tmp_path / "muse-work"
412 workdir.mkdir()
413 (workdir / "melody.mid").write_bytes(_make_minimal_midi())
414
415 commit_id = await _commit_async(
416 message="prefix resolution test",
417 root=tmp_path,
418 session=muse_cli_db_session,
419 )
420 _set_head(tmp_path, commit_id)
421
422 result = await _export_async(
423 commit_ref=commit_id[:8], # use short prefix
424 fmt=ExportFormat.JSON,
425 output=tmp_path / "prefix_out.json",
426 track=None,
427 section=None,
428 split_tracks=False,
429 root=tmp_path,
430 session=muse_cli_db_session,
431 )
432
433 assert result.commit_id == commit_id
434 assert result.paths_written
435
436
437 # ---------------------------------------------------------------------------
438 # Unit tests — helper functions
439 # ---------------------------------------------------------------------------
440
441
442 def test_midi_note_to_step_octave_middle_c() -> None:
443 """MIDI note 60 (middle C) maps to step='C', octave=4."""
444 step, octave = _midi_note_to_step_octave(60)
445 assert step == "C"
446 assert octave == 4
447
448
449 def test_midi_note_to_step_octave_sharp() -> None:
450 """MIDI note 61 (C#4) maps to step='C#', octave=4."""
451 step, octave = _midi_note_to_step_octave(61)
452 assert step == "C#"
453 assert octave == 4
454
455
456 def test_midi_note_to_abc_middle_c() -> None:
457 """MIDI note 60 (C4) maps to ABC 'C'."""
458 assert _midi_note_to_abc(60) == "C"
459
460
461 def test_midi_note_to_abc_c5() -> None:
462 """MIDI note 72 (C5) maps to ABC 'c' (lowercase)."""
463 assert _midi_note_to_abc(72) == "c"
464
465
466 def test_midi_note_to_abc_c3() -> None:
467 """MIDI note 48 (C3) maps to ABC 'C,' (comma suffix)."""
468 assert _midi_note_to_abc(48) == "C,"
469
470
471 def test_resolve_commit_id_returns_head(tmp_path: pathlib.Path) -> None:
472 """resolve_commit_id returns the HEAD commit ID when prefix is None."""
473 _init_muse_repo(tmp_path)
474 commit_id = "a" * 64
475 _set_head(tmp_path, commit_id)
476
477 result = resolve_commit_id(tmp_path, None)
478 assert result == commit_id
479
480
481 def test_resolve_commit_id_returns_prefix(tmp_path: pathlib.Path) -> None:
482 """resolve_commit_id returns the prefix unchanged when provided."""
483 _init_muse_repo(tmp_path)
484 prefix = "abcd1234"
485
486 result = resolve_commit_id(tmp_path, prefix)
487 assert result == prefix
488
489
490 def test_resolve_commit_id_raises_when_no_commits(tmp_path: pathlib.Path) -> None:
491 """resolve_commit_id raises ValueError when HEAD has no commits."""
492 _init_muse_repo(tmp_path) # HEAD ref is empty
493
494 with pytest.raises(ValueError, match="No commits yet"):
495 resolve_commit_id(tmp_path, None)
496
497
498 def test_default_output_path() -> None:
499 """_default_output_path uses first 8 chars of commit_id and format extension."""
500 commit_id = "abcdef12" + "0" * 56
501 path = _default_output_path(commit_id, ExportFormat.MIDI)
502 assert path.name == "abcdef12.midi"
503 assert path.parent.name == "exports"