cgcardona / muse public
test_transpose.py python
617 lines 21.4 KB
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse transpose`` — CLI interface, service layer, and integration.
2
3 Coverage:
4 - ``parse_interval``: signed integers, named intervals, error cases.
5 - ``update_key_metadata``: key transposition, edge cases.
6 - ``transpose_midi_bytes``: valid MIDI, drum exclusion, track filter, identity.
7 - ``apply_transpose_to_workdir``: file discovery, modification, dry-run.
8 - ``_transpose_async``: end-to-end with in-memory DB (creates real commits).
9 - CLI via CliRunner: argument parsing, flag handling, exit codes.
10
11 Regression test naming follows the issue specification:
12 - ``test_transpose_excludes_drum_channel_from_pitch_shift``
13 - ``test_transpose_semitones_updates_key_metadata``
14 - ``test_transpose_named_interval_down_perfect_fifth``
15 - ``test_transpose_scoped_to_track``
16 - ``test_transpose_dry_run_no_commit_created``
17 """
18 from __future__ import annotations
19
20 import json
21 import pathlib
22 import struct
23 import uuid
24
25 import pytest
26 import typer
27 from sqlalchemy.ext.asyncio import AsyncSession
28
29 from maestro.muse_cli.commands.commit import _commit_async
30 from maestro.muse_cli.commands.transpose import _transpose_async
31 from maestro.muse_cli.errors import ExitCode
32 from maestro.muse_cli.models import MuseCliCommit
33 from maestro.services.muse_transpose import (
34 TransposeResult,
35 apply_transpose_to_workdir,
36 parse_interval,
37 transpose_midi_bytes,
38 update_key_metadata,
39 )
40
41 # ---------------------------------------------------------------------------
42 # Repo + workdir helpers
43 # ---------------------------------------------------------------------------
44
45
46 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
47 """Initialise a minimal .muse/ layout for testing."""
48 rid = str(uuid.uuid4())
49 muse = root / ".muse"
50 (muse / "refs" / "heads").mkdir(parents=True)
51 (muse / "repo.json").write_text(
52 json.dumps({"repo_id": rid, "schema_version": "1"})
53 )
54 (muse / "HEAD").write_text(f"refs/heads/{branch}")
55 (muse / "refs" / "heads" / branch).write_text("")
56 return rid
57
58
59 def _write_workdir(root: pathlib.Path, files: dict[str, bytes]) -> None:
60 workdir = root / "muse-work"
61 workdir.mkdir(exist_ok=True)
62 for name, data in files.items():
63 (workdir / name).write_bytes(data)
64
65
66 async def _make_commit(
67 root: pathlib.Path,
68 session: AsyncSession,
69 message: str = "initial",
70 files: dict[str, bytes] | None = None,
71 ) -> str:
72 """Write files to muse-work/ and commit."""
73 if files is None:
74 files = {"track.mid": b"MIDI-placeholder"}
75 _write_workdir(root, files)
76 return await _commit_async(message=message, root=root, session=session)
77
78
79 # ---------------------------------------------------------------------------
80 # MIDI helpers
81 # ---------------------------------------------------------------------------
82
83
84 def _build_minimal_midi(
85 note: int = 60,
86 velocity: int = 80,
87 channel: int = 0,
88 ) -> bytes:
89 """Build a minimal Type-0 MIDI file with one note-on + note-off pair.
90
91 Produces a syntactically valid MIDI file with:
92 - MThd header (format 0, 1 track, 480 ticks/quarter)
93 - One MTrk chunk with:
94 - delta(0) Note-On channel note velocity
95 - delta(480) Note-Off channel note 0
96 - delta(0) End-of-Track meta
97
98 Channel is 0-indexed (use 9 for drums).
99 """
100 note_on_status = 0x90 | (channel & 0x0F)
101 note_off_status = 0x80 | (channel & 0x0F)
102
103 track_events = bytes([
104 0x00, note_on_status, note, velocity, # delta=0, note on
105 0x83, 0x60, note_off_status, note, 0x00, # delta=480 (VLQ), note off
106 0x00, 0xFF, 0x2F, 0x00, # delta=0, end of track
107 ])
108
109 header = b"MThd" + struct.pack(">I", 6) + struct.pack(">HHH", 0, 1, 480)
110 track = b"MTrk" + struct.pack(">I", len(track_events)) + track_events
111 return header + track
112
113
114 def _build_midi_with_track_name(
115 track_name: str,
116 note: int = 60,
117 channel: int = 0,
118 ) -> bytes:
119 """Build a MIDI file with a Track Name meta-event followed by a note.
120
121 The Track Name is embedded as meta-event 0xFF 0x03 at the start of the
122 track so ``_get_track_name`` and the ``--track`` filter can detect it.
123 """
124 name_bytes = track_name.encode("latin-1")
125 name_event = bytes([
126 0x00, 0xFF, 0x03, len(name_bytes)
127 ]) + name_bytes
128
129 note_on_status = 0x90 | (channel & 0x0F)
130 note_off_status = 0x80 | (channel & 0x0F)
131 note_events = bytes([
132 0x00, note_on_status, note, 80,
133 0x83, 0x60, note_off_status, note, 0x00,
134 0x00, 0xFF, 0x2F, 0x00,
135 ])
136
137 track_events = name_event + note_events
138 header = b"MThd" + struct.pack(">I", 6) + struct.pack(">HHH", 0, 1, 480)
139 track = b"MTrk" + struct.pack(">I", len(track_events)) + track_events
140 return header + track
141
142
143 # ---------------------------------------------------------------------------
144 # parse_interval tests
145 # ---------------------------------------------------------------------------
146
147
148 @pytest.mark.parametrize("interval_str,expected", [
149 ("+3", 3),
150 ("-5", -5),
151 ("+12", 12),
152 ("12", 12),
153 ("-12", -12),
154 ("0", 0),
155 ("+0", 0),
156 ])
157 def test_parse_interval_integers(interval_str: str, expected: int) -> None:
158 """Signed integers are parsed correctly to semitone counts."""
159 assert parse_interval(interval_str) == expected
160
161
162 def test_transpose_named_interval_down_perfect_fifth() -> None:
163 """``down-perfect5th`` resolves to -7 semitones (regression test)."""
164 assert parse_interval("down-perfect5th") == -7
165
166
167 @pytest.mark.parametrize("interval_str,expected", [
168 ("up-minor3rd", 3),
169 ("up-major3rd", 4),
170 ("up-perfect4th", 5),
171 ("up-perfect5th", 7),
172 ("up-octave", 12),
173 ("down-minor3rd", -3),
174 ("down-major3rd", -4),
175 ("down-perfect5th", -7),
176 ("down-octave", -12),
177 ("up-unison", 0),
178 ])
179 def test_parse_interval_named(interval_str: str, expected: int) -> None:
180 """Named intervals with up-/down- prefix resolve to correct semitone counts."""
181 assert parse_interval(interval_str) == expected
182
183
184 def test_parse_interval_invalid_string_raises_value_error() -> None:
185 """Unparseable strings raise ValueError with a descriptive message."""
186 with pytest.raises(ValueError, match="Cannot parse interval"):
187 parse_interval("jump-high")
188
189
190 def test_parse_interval_unknown_name_raises_value_error() -> None:
191 """Known direction but unknown interval name raises ValueError."""
192 with pytest.raises(ValueError, match="Unknown interval name"):
193 parse_interval("up-ultrawide9th")
194
195
196 # ---------------------------------------------------------------------------
197 # update_key_metadata tests
198 # ---------------------------------------------------------------------------
199
200
201 def test_transpose_semitones_updates_key_metadata() -> None:
202 """Transposing 'Eb major' by +2 semitones yields 'F major' (regression test)."""
203 assert update_key_metadata("Eb major", 2) == "F major"
204
205
206 @pytest.mark.parametrize("key_str,semitones,expected", [
207 ("C major", 2, "D major"),
208 ("C major", -1, "B major"),
209 ("F# minor", 2, "Ab minor"), # G#/Ab are enharmonic; service uses flat names
210 ("Bb major", 3, "Db major"),
211 ("A minor", 12, "A minor"), # octave → same key
212 ("G major", -7, "C major"), # down perfect 5th
213 ("Eb major", 2, "F major"),
214 ("Eb major", 3, "F# major"),
215 ])
216 def test_update_key_metadata_parametrized(
217 key_str: str, semitones: int, expected: str
218 ) -> None:
219 """Key metadata is updated correctly for common transpositions."""
220 assert update_key_metadata(key_str, semitones) == expected
221
222
223 def test_update_key_metadata_unknown_root_returns_unchanged() -> None:
224 """An unrecognized root note in the key string is returned unchanged."""
225 assert update_key_metadata("X# major", 3) == "X# major"
226
227
228 def test_update_key_metadata_empty_string() -> None:
229 """An empty key string is returned unchanged."""
230 assert update_key_metadata("", 3) == ""
231
232
233 def test_update_key_metadata_preserves_mode() -> None:
234 """Mode string (major, minor, dorian, etc.) is preserved verbatim."""
235 assert update_key_metadata("D dorian", 2) == "E dorian"
236
237
238 # ---------------------------------------------------------------------------
239 # transpose_midi_bytes tests
240 # ---------------------------------------------------------------------------
241
242
243 def test_transpose_midi_bytes_transposes_note() -> None:
244 """A note-on event on a pitched channel is shifted by the semitone offset."""
245 midi = _build_minimal_midi(note=60, channel=0)
246 transposed, count = transpose_midi_bytes(midi, semitones=3)
247 assert count > 0, "Expected at least one note byte to change"
248 # The transposed file should differ from the original
249 assert transposed != midi
250
251
252 def test_transpose_excludes_drum_channel_from_pitch_shift() -> None:
253 """Note-on events on channel 9 (drums) must NOT be transposed (regression test)."""
254 drum_midi = _build_minimal_midi(note=36, channel=9) # channel 9 = drums
255 transposed, count = transpose_midi_bytes(drum_midi, semitones=5)
256 assert count == 0, "Drum notes must not be transposed"
257 assert transposed == drum_midi, "Drum MIDI bytes must be identical after transpose"
258
259
260 def test_transpose_midi_bytes_zero_semitones_identity() -> None:
261 """Zero semitones applied to a MIDI file returns the file unchanged."""
262 midi = _build_minimal_midi(note=60, channel=0)
263 transposed, count = transpose_midi_bytes(midi, semitones=0)
264 assert count == 0
265 assert transposed == midi
266
267
268 def test_transpose_midi_bytes_clamps_note_at_max() -> None:
269 """Notes shifted beyond 127 are clamped to 127."""
270 midi = _build_minimal_midi(note=126, channel=0)
271 transposed, _ = transpose_midi_bytes(midi, semitones=10)
272 # We can't easily introspect the note value without re-parsing,
273 # but we verify the file is structurally valid (same length, different bytes)
274 assert len(transposed) == len(midi)
275 assert transposed != midi
276
277
278 def test_transpose_midi_bytes_clamps_note_at_min() -> None:
279 """Notes shifted below 0 are clamped to 0."""
280 midi = _build_minimal_midi(note=1, channel=0)
281 transposed, _ = transpose_midi_bytes(midi, semitones=-10)
282 assert len(transposed) == len(midi)
283
284
285 def test_transpose_midi_bytes_invalid_file_returns_unchanged() -> None:
286 """Non-MIDI bytes are returned unchanged with 0 notes changed."""
287 data = b"not a midi file at all"
288 transposed, count = transpose_midi_bytes(data, semitones=5)
289 assert transposed == data
290 assert count == 0
291
292
293 def test_transpose_scoped_to_track() -> None:
294 """``--track`` filter only transposes the matching track; non-matching tracks are unchanged (regression test)."""
295 melody_midi = _build_midi_with_track_name("melody", note=60, channel=0)
296 bass_midi = _build_midi_with_track_name("bass", note=36, channel=1)
297
298 melody_transposed, melody_count = transpose_midi_bytes(melody_midi, semitones=3, track_filter="melody")
299 bass_transposed, bass_count = transpose_midi_bytes(bass_midi, semitones=3, track_filter="melody")
300
301 assert melody_count > 0, "Melody track should be transposed"
302 assert bass_count == 0, "Bass track should be skipped (name doesn't match 'melody')"
303 assert bass_transposed == bass_midi, "Bass MIDI bytes should be identical"
304
305
306 def test_transpose_track_filter_case_insensitive() -> None:
307 """Track name filter matching is case-insensitive."""
308 midi = _build_midi_with_track_name("Lead Guitar", note=60, channel=0)
309 _, count = transpose_midi_bytes(midi, semitones=2, track_filter="lead guitar")
310 assert count > 0
311
312
313 def test_transpose_track_filter_substring_match() -> None:
314 """Track name filter matches as a substring."""
315 midi = _build_midi_with_track_name("Piano Lead", note=60, channel=0)
316 _, count = transpose_midi_bytes(midi, semitones=2, track_filter="lead")
317 assert count > 0
318
319
320 # ---------------------------------------------------------------------------
321 # apply_transpose_to_workdir tests
322 # ---------------------------------------------------------------------------
323
324
325 def test_apply_transpose_to_workdir_modifies_midi_files(
326 tmp_path: pathlib.Path,
327 ) -> None:
328 """MIDI files in workdir are transposed and written back."""
329 workdir = tmp_path / "muse-work"
330 workdir.mkdir()
331 midi_path = workdir / "track.mid"
332 midi_path.write_bytes(_build_minimal_midi(note=60, channel=0))
333
334 modified, skipped = apply_transpose_to_workdir(workdir, semitones=3)
335
336 assert "track.mid" in modified
337 assert len(skipped) == 0
338 # File should be modified on disk
339 new_bytes = midi_path.read_bytes()
340 assert new_bytes != _build_minimal_midi(note=60, channel=0)
341
342
343 def test_apply_transpose_to_workdir_skips_non_midi(
344 tmp_path: pathlib.Path,
345 ) -> None:
346 """Non-MIDI files (e.g. JSON, WAV) are skipped entirely."""
347 workdir = tmp_path / "muse-work"
348 workdir.mkdir()
349 (workdir / "notes.json").write_text('{"key": "C"}')
350 (workdir / "track.mid").write_bytes(_build_minimal_midi(note=60, channel=0))
351
352 modified, skipped = apply_transpose_to_workdir(workdir, semitones=2)
353
354 assert "track.mid" in modified
355 assert "notes.json" not in modified
356
357
358 def test_apply_transpose_to_workdir_dry_run_does_not_write(
359 tmp_path: pathlib.Path,
360 ) -> None:
361 """Dry-run mode reports what would change without writing files (regression test)."""
362 workdir = tmp_path / "muse-work"
363 workdir.mkdir()
364 original_bytes = _build_minimal_midi(note=60, channel=0)
365 midi_path = workdir / "track.mid"
366 midi_path.write_bytes(original_bytes)
367
368 modified, _ = apply_transpose_to_workdir(workdir, semitones=3, dry_run=True)
369
370 assert "track.mid" in modified, "Dry-run should still report files that would be modified"
371 assert midi_path.read_bytes() == original_bytes, "File must not be written in dry-run mode"
372
373
374 def test_apply_transpose_to_workdir_missing_workdir(
375 tmp_path: pathlib.Path,
376 ) -> None:
377 """Missing muse-work/ directory returns empty lists without raising."""
378 workdir = tmp_path / "muse-work" # intentionally not created
379 modified, skipped = apply_transpose_to_workdir(workdir, semitones=3)
380 assert modified == []
381 assert skipped == []
382
383
384 def test_apply_transpose_dry_run_no_commit_created(
385 tmp_path: pathlib.Path,
386 ) -> None:
387 """Dry-run flag does not write files — ``files_modified`` are reported only (regression test)."""
388 workdir = tmp_path / "muse-work"
389 workdir.mkdir()
390 original = _build_minimal_midi(note=60, channel=0)
391 (workdir / "song.mid").write_bytes(original)
392
393 modified, _ = apply_transpose_to_workdir(workdir, semitones=5, dry_run=True)
394
395 assert "song.mid" in modified
396 assert (workdir / "song.mid").read_bytes() == original
397
398
399 # ---------------------------------------------------------------------------
400 # _transpose_async integration tests (require DB session)
401 # ---------------------------------------------------------------------------
402
403
404 @pytest.mark.anyio
405 async def test_transpose_async_creates_commit(
406 tmp_path: pathlib.Path,
407 muse_cli_db_session: AsyncSession,
408 ) -> None:
409 """_transpose_async creates a new commit pointing to the transposed snapshot."""
410 _init_muse_repo(tmp_path)
411 midi = _build_minimal_midi(note=60, channel=0)
412 await _make_commit(tmp_path, muse_cli_db_session, files={"beat.mid": midi})
413
414 result = await _transpose_async(
415 root=tmp_path,
416 session=muse_cli_db_session,
417 semitones=2,
418 commit_ref=None,
419 track_filter=None,
420 section_filter=None,
421 message=None,
422 dry_run=False,
423 as_json=False,
424 )
425
426 assert result.new_commit_id is not None
427 assert len(result.files_modified) > 0
428 assert result.semitones == 2
429 assert not result.dry_run
430
431 # Verify commit was persisted
432 new_commit = await muse_cli_db_session.get(MuseCliCommit, result.new_commit_id)
433 assert new_commit is not None
434 assert new_commit.parent_commit_id == result.source_commit_id
435
436
437 @pytest.mark.anyio
438 async def test_transpose_async_dry_run_creates_no_commit(
439 tmp_path: pathlib.Path,
440 muse_cli_db_session: AsyncSession,
441 ) -> None:
442 """_transpose_async with dry_run=True does not create a commit or write files."""
443 _init_muse_repo(tmp_path)
444 midi = _build_minimal_midi(note=60, channel=0)
445 await _make_commit(tmp_path, muse_cli_db_session, files={"melody.mid": midi})
446
447 original_bytes = (tmp_path / "muse-work" / "melody.mid").read_bytes()
448
449 result = await _transpose_async(
450 root=tmp_path,
451 session=muse_cli_db_session,
452 semitones=3,
453 commit_ref=None,
454 track_filter=None,
455 section_filter=None,
456 message=None,
457 dry_run=True,
458 as_json=False,
459 )
460
461 assert result.new_commit_id is None
462 assert result.dry_run is True
463 # File must not have been written
464 assert (tmp_path / "muse-work" / "melody.mid").read_bytes() == original_bytes
465
466
467 @pytest.mark.anyio
468 async def test_transpose_async_updates_key_metadata(
469 tmp_path: pathlib.Path,
470 muse_cli_db_session: AsyncSession,
471 ) -> None:
472 """Key metadata in commit is updated when source commit has a key annotation."""
473 from maestro.muse_cli.db import resolve_commit_ref
474
475 _init_muse_repo(tmp_path)
476 midi = _build_minimal_midi(note=60, channel=0)
477 source_id = await _make_commit(tmp_path, muse_cli_db_session, files={"track.mid": midi})
478
479 # Manually annotate source commit with a key
480 source_commit = await muse_cli_db_session.get(MuseCliCommit, source_id)
481 assert source_commit is not None
482 source_commit.commit_metadata = {"key": "Eb major"}
483 muse_cli_db_session.add(source_commit)
484 await muse_cli_db_session.flush()
485
486 result = await _transpose_async(
487 root=tmp_path,
488 session=muse_cli_db_session,
489 semitones=2,
490 commit_ref=None,
491 track_filter=None,
492 section_filter=None,
493 message=None,
494 dry_run=False,
495 as_json=False,
496 )
497
498 assert result.original_key == "Eb major"
499 assert result.new_key == "F major"
500
501 # Verify persisted on the new commit
502 new_commit = await muse_cli_db_session.get(MuseCliCommit, result.new_commit_id)
503 assert new_commit is not None
504 assert new_commit.commit_metadata is not None
505 assert new_commit.commit_metadata.get("key") == "F major"
506
507
508 @pytest.mark.anyio
509 async def test_transpose_async_missing_commit_ref_exits(
510 tmp_path: pathlib.Path,
511 muse_cli_db_session: AsyncSession,
512 ) -> None:
513 """_transpose_async raises typer.Exit(USER_ERROR) when commit ref not found."""
514 _init_muse_repo(tmp_path)
515 # No commits in the repo
516
517 with pytest.raises(typer.Exit) as exc_info:
518 await _transpose_async(
519 root=tmp_path,
520 session=muse_cli_db_session,
521 semitones=2,
522 commit_ref="nonexistent",
523 track_filter=None,
524 section_filter=None,
525 message=None,
526 dry_run=False,
527 as_json=False,
528 )
529
530 assert exc_info.value.exit_code == ExitCode.USER_ERROR
531
532
533 @pytest.mark.anyio
534 async def test_transpose_async_custom_message(
535 tmp_path: pathlib.Path,
536 muse_cli_db_session: AsyncSession,
537 ) -> None:
538 """Custom ``--message`` is used as the commit message for the transposed commit."""
539 _init_muse_repo(tmp_path)
540 midi = _build_minimal_midi(note=60, channel=0)
541 await _make_commit(tmp_path, muse_cli_db_session, files={"track.mid": midi})
542
543 result = await _transpose_async(
544 root=tmp_path,
545 session=muse_cli_db_session,
546 semitones=5,
547 commit_ref=None,
548 track_filter=None,
549 section_filter=None,
550 message="My custom transpose message",
551 dry_run=False,
552 as_json=False,
553 )
554
555 assert result.new_commit_id is not None
556 new_commit = await muse_cli_db_session.get(MuseCliCommit, result.new_commit_id)
557 assert new_commit is not None
558 assert new_commit.message == "My custom transpose message"
559
560
561 @pytest.mark.anyio
562 async def test_transpose_async_json_output(
563 tmp_path: pathlib.Path,
564 muse_cli_db_session: AsyncSession,
565 ) -> None:
566 """``--json`` flag returns a result with all required fields populated."""
567 _init_muse_repo(tmp_path)
568 midi = _build_minimal_midi(note=60, channel=0)
569 await _make_commit(tmp_path, muse_cli_db_session, files={"track.mid": midi})
570
571 result = await _transpose_async(
572 root=tmp_path,
573 session=muse_cli_db_session,
574 semitones=3,
575 commit_ref=None,
576 track_filter=None,
577 section_filter=None,
578 message=None,
579 dry_run=False,
580 as_json=True,
581 )
582
583 # Verify the result object has all expected fields (JSON rendering is tested via CLI runner)
584 assert result.source_commit_id != ""
585 assert result.semitones == 3
586 assert result.new_commit_id is not None
587 assert isinstance(result.files_modified, list)
588 assert result.dry_run is False
589
590
591 # ---------------------------------------------------------------------------
592 # CLI CliRunner tests
593 # ---------------------------------------------------------------------------
594
595
596 def test_cli_transpose_bad_interval_exits_with_user_error(
597 tmp_path: pathlib.Path,
598 ) -> None:
599 """Invalid interval string exits with code 1 (USER_ERROR) without crashing."""
600 from typer.testing import CliRunner
601 from maestro.muse_cli.app import cli
602
603 runner = CliRunner()
604 # We need a repo to get past require_repo, but the interval is checked first
605 result = runner.invoke(cli, ["transpose", "jump-high"])
606 assert result.exit_code == ExitCode.USER_ERROR
607
608
609 def test_cli_transpose_help() -> None:
610 """``muse transpose --help`` exits cleanly and contains interval description."""
611 from typer.testing import CliRunner
612 from maestro.muse_cli.app import cli
613
614 runner = CliRunner()
615 result = runner.invoke(cli, ["transpose", "--help"])
616 assert result.exit_code == 0
617 assert "interval" in result.output.lower() or "transpose" in result.output.lower()