cgcardona / muse public
test_meter.py python
488 lines 18.8 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse meter`` — time-signature read/set/detect/history/polyrhythm.
2
3 All async tests use the ``muse_cli_db_session`` fixture (in-memory SQLite).
4 Pure-logic tests (MIDI parsing, validation) are synchronous.
5 """
6 from __future__ import annotations
7
8 import json
9 import pathlib
10 import struct
11 import uuid
12
13 import pytest
14 import typer
15 from sqlalchemy.ext.asyncio import AsyncSession
16
17 from maestro.muse_cli.commands.commit import _commit_async
18 from maestro.muse_cli.commands.meter import (
19 MuseMeterHistoryEntry,
20 MuseMeterReadResult,
21 MusePolyrhythmResult,
22 _meter_history_async,
23 _meter_polyrhythm_async,
24 _meter_read_async,
25 _meter_set_async,
26 detect_midi_time_signature,
27 scan_workdir_for_time_signatures,
28 validate_time_signature,
29 )
30 from maestro.muse_cli.errors import ExitCode
31
32
33 # ──────────────────────────────────────────────────────────────────────────────
34 # Test helpers
35 # ──────────────────────────────────────────────────────────────────────────────
36
37
38 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
39 """Create a minimal .muse/ layout."""
40 rid = repo_id or str(uuid.uuid4())
41 muse = root / ".muse"
42 (muse / "refs" / "heads").mkdir(parents=True)
43 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
44 (muse / "HEAD").write_text("refs/heads/main")
45 (muse / "refs" / "heads" / "main").write_text("")
46 return rid
47
48
49 def _populate_workdir(root: pathlib.Path, files: dict[str, bytes] | None = None) -> None:
50 workdir = root / "muse-work"
51 workdir.mkdir(exist_ok=True)
52 if files is None:
53 files = {"beat.mid": b"MIDI-DATA"}
54 for name, content in files.items():
55 (workdir / name).write_bytes(content)
56
57
58 def _make_midi_with_time_sig(numerator: int, denominator_exp: int) -> bytes:
59 """Build a minimal valid MIDI file containing a time-signature meta event."""
60 # Time-signature meta event: FF 58 04 nn dd cc bb
61 time_sig_event = bytes([
62 0x00, # delta time (0)
63 0xFF, 0x58, 0x04, # meta type, length
64 numerator, denominator_exp, # numerator, denominator exponent
65 0x18, 0x08, # clocks/tick, 32nds/quarter
66 ])
67 # End-of-track event
68 eot = bytes([0x00, 0xFF, 0x2F, 0x00])
69
70 track_data = time_sig_event + eot
71 track_length = len(track_data)
72
73 # MThd header: type=1, ntracks=1, division=480
74 header = b"MThd" + struct.pack(">I", 6) + struct.pack(">HHH", 1, 1, 480)
75 # MTrk chunk
76 track = b"MTrk" + struct.pack(">I", track_length) + track_data
77 return header + track
78
79
80 # ──────────────────────────────────────────────────────────────────────────────
81 # validate_time_signature — pure logic
82 # ──────────────────────────────────────────────────────────────────────────────
83
84
85 def test_validate_time_signature_accepts_4_4() -> None:
86 assert validate_time_signature("4/4") == "4/4"
87
88
89 def test_validate_time_signature_accepts_7_8() -> None:
90 assert validate_time_signature("7/8") == "7/8"
91
92
93 def test_validate_time_signature_accepts_5_4() -> None:
94 assert validate_time_signature("5/4") == "5/4"
95
96
97 def test_validate_time_signature_accepts_3_4() -> None:
98 assert validate_time_signature("3/4") == "3/4"
99
100
101 def test_validate_time_signature_accepts_12_8() -> None:
102 assert validate_time_signature("12/8") == "12/8"
103
104
105 def test_validate_time_signature_strips_whitespace() -> None:
106 assert validate_time_signature(" 4/4 ") == "4/4"
107
108
109 def test_validate_time_signature_rejects_non_power_of_two_denominator() -> None:
110 with pytest.raises(ValueError, match="power of 2"):
111 validate_time_signature("4/3")
112
113
114 def test_validate_time_signature_rejects_zero_numerator() -> None:
115 with pytest.raises(ValueError, match="[Nn]umerator"):
116 validate_time_signature("0/4")
117
118
119 def test_validate_time_signature_rejects_malformed() -> None:
120 with pytest.raises(ValueError):
121 validate_time_signature("four-four")
122
123
124 def test_validate_time_signature_rejects_missing_slash() -> None:
125 with pytest.raises(ValueError):
126 validate_time_signature("44")
127
128
129 # ──────────────────────────────────────────────────────────────────────────────
130 # detect_midi_time_signature — MIDI parsing
131 # ──────────────────────────────────────────────────────────────────────────────
132
133
134 def test_detect_midi_time_signature_4_4() -> None:
135 midi = _make_midi_with_time_sig(numerator=4, denominator_exp=2) # 2^2=4
136 assert detect_midi_time_signature(midi) == "4/4"
137
138
139 def test_detect_midi_time_signature_3_4() -> None:
140 midi = _make_midi_with_time_sig(numerator=3, denominator_exp=2)
141 assert detect_midi_time_signature(midi) == "3/4"
142
143
144 def test_detect_midi_time_signature_7_8() -> None:
145 midi = _make_midi_with_time_sig(numerator=7, denominator_exp=3) # 2^3=8
146 assert detect_midi_time_signature(midi) == "7/8"
147
148
149 def test_detect_midi_time_signature_returns_none_for_empty_bytes() -> None:
150 assert detect_midi_time_signature(b"") is None
151
152
153 def test_detect_midi_time_signature_returns_none_for_no_event() -> None:
154 # Random bytes with no FF 58 sequence
155 assert detect_midi_time_signature(b"\x00\x90\x3C\x7F\x00\x80\x3C\x00") is None
156
157
158 def test_detect_midi_time_signature_12_8() -> None:
159 midi = _make_midi_with_time_sig(numerator=12, denominator_exp=3) # 2^3=8
160 assert detect_midi_time_signature(midi) == "12/8"
161
162
163 # ──────────────────────────────────────────────────────────────────────────────
164 # scan_workdir_for_time_signatures
165 # ──────────────────────────────────────────────────────────────────────────────
166
167
168 def test_scan_workdir_finds_time_signature_in_midi(tmp_path: pathlib.Path) -> None:
169 workdir = tmp_path / "muse-work"
170 workdir.mkdir()
171 (workdir / "beat.mid").write_bytes(_make_midi_with_time_sig(4, 2)) # 4/4
172
173 sigs = scan_workdir_for_time_signatures(workdir)
174 assert sigs == {"beat.mid": "4/4"}
175
176
177 def test_scan_workdir_returns_question_mark_for_unknown(tmp_path: pathlib.Path) -> None:
178 workdir = tmp_path / "muse-work"
179 workdir.mkdir()
180 (workdir / "no-sig.mid").write_bytes(b"\x00\x90\x3C\x7F")
181
182 sigs = scan_workdir_for_time_signatures(workdir)
183 assert sigs == {"no-sig.mid": "?"}
184
185
186 def test_scan_workdir_returns_empty_for_missing_workdir(tmp_path: pathlib.Path) -> None:
187 sigs = scan_workdir_for_time_signatures(tmp_path / "muse-work")
188 assert sigs == {}
189
190
191 def test_scan_workdir_ignores_non_midi_files(tmp_path: pathlib.Path) -> None:
192 workdir = tmp_path / "muse-work"
193 workdir.mkdir()
194 (workdir / "render.mp3").write_bytes(b"MP3-DATA")
195 (workdir / "beat.mid").write_bytes(_make_midi_with_time_sig(3, 2)) # 3/4
196
197 sigs = scan_workdir_for_time_signatures(workdir)
198 assert "render.mp3" not in sigs
199 assert "beat.mid" in sigs
200
201
202 def test_scan_workdir_multiple_midi_files(tmp_path: pathlib.Path) -> None:
203 workdir = tmp_path / "muse-work"
204 workdir.mkdir()
205 (workdir / "drums.mid").write_bytes(_make_midi_with_time_sig(4, 2)) # 4/4
206 (workdir / "bass.mid").write_bytes(_make_midi_with_time_sig(4, 2)) # 4/4
207
208 sigs = scan_workdir_for_time_signatures(workdir)
209 assert len(sigs) == 2
210 assert all(s == "4/4" for s in sigs.values())
211
212
213 # ──────────────────────────────────────────────────────────────────────────────
214 # _meter_read_async / _meter_set_async — DB integration
215 # ──────────────────────────────────────────────────────────────────────────────
216
217
218 @pytest.mark.anyio
219 async def test_meter_read_returns_none_when_not_set(
220 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
221 ) -> None:
222 """Reading meter on an uncommitted repo raises USER_ERROR (no HEAD)."""
223 _init_muse_repo(tmp_path)
224 _populate_workdir(tmp_path)
225
226 commit_id = await _commit_async(
227 message="bare commit", root=tmp_path, session=muse_cli_db_session
228 )
229
230 result = await _meter_read_async(
231 session=muse_cli_db_session, root=tmp_path, commit_ref=None
232 )
233 assert isinstance(result, MuseMeterReadResult)
234 assert result.commit_id == commit_id
235 assert result.time_signature is None
236
237
238 @pytest.mark.anyio
239 async def test_meter_set_and_read_roundtrip(
240 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
241 ) -> None:
242 """Setting a meter annotation and reading it back returns the same value."""
243 _init_muse_repo(tmp_path)
244 _populate_workdir(tmp_path)
245
246 commit_id = await _commit_async(
247 message="jazz take", root=tmp_path, session=muse_cli_db_session
248 )
249
250 await _meter_set_async(
251 session=muse_cli_db_session,
252 root=tmp_path,
253 commit_ref=None,
254 time_signature="7/8",
255 )
256 await muse_cli_db_session.flush()
257
258 result = await _meter_read_async(
259 session=muse_cli_db_session, root=tmp_path, commit_ref=None
260 )
261 assert result.commit_id == commit_id
262 assert result.time_signature == "7/8"
263
264
265 @pytest.mark.anyio
266 async def test_meter_set_by_abbreviated_commit_id(
267 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
268 ) -> None:
269 """--set works when an abbreviated commit ID is passed."""
270 _init_muse_repo(tmp_path)
271 _populate_workdir(tmp_path)
272
273 commit_id = await _commit_async(
274 message="boom bap", root=tmp_path, session=muse_cli_db_session
275 )
276
277 set_commit_id = await _meter_set_async(
278 session=muse_cli_db_session,
279 root=tmp_path,
280 commit_ref=commit_id[:8],
281 time_signature="4/4",
282 )
283 assert set_commit_id == commit_id
284
285
286 @pytest.mark.anyio
287 async def test_meter_read_no_commits_raises_exit(
288 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
289 ) -> None:
290 """Reading meter when there are no commits raises typer.Exit(USER_ERROR)."""
291 _init_muse_repo(tmp_path)
292
293 with pytest.raises(typer.Exit) as exc_info:
294 await _meter_read_async(
295 session=muse_cli_db_session, root=tmp_path, commit_ref=None
296 )
297 assert exc_info.value.exit_code == ExitCode.USER_ERROR
298
299
300 @pytest.mark.anyio
301 async def test_meter_set_unknown_commit_raises_exit(
302 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
303 ) -> None:
304 """Setting meter on an unknown commit ref raises typer.Exit(USER_ERROR)."""
305 _init_muse_repo(tmp_path)
306 _populate_workdir(tmp_path)
307 await _commit_async(message="init", root=tmp_path, session=muse_cli_db_session)
308
309 with pytest.raises(typer.Exit) as exc_info:
310 await _meter_set_async(
311 session=muse_cli_db_session,
312 root=tmp_path,
313 commit_ref="deadbeef",
314 time_signature="4/4",
315 )
316 assert exc_info.value.exit_code == ExitCode.USER_ERROR
317
318
319 @pytest.mark.anyio
320 async def test_meter_set_overwrites_previous_annotation(
321 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
322 ) -> None:
323 """Setting meter twice on the same commit overwrites the first annotation."""
324 _init_muse_repo(tmp_path)
325 _populate_workdir(tmp_path)
326 await _commit_async(message="v1", root=tmp_path, session=muse_cli_db_session)
327
328 await _meter_set_async(
329 session=muse_cli_db_session, root=tmp_path, commit_ref=None, time_signature="4/4"
330 )
331 await muse_cli_db_session.flush()
332 await _meter_set_async(
333 session=muse_cli_db_session, root=tmp_path, commit_ref=None, time_signature="3/4"
334 )
335 await muse_cli_db_session.flush()
336
337 result = await _meter_read_async(
338 session=muse_cli_db_session, root=tmp_path, commit_ref=None
339 )
340 assert result.time_signature == "3/4"
341
342
343 # ──────────────────────────────────────────────────────────────────────────────
344 # _meter_history_async
345 # ──────────────────────────────────────────────────────────────────────────────
346
347
348 @pytest.mark.anyio
349 async def test_meter_history_returns_empty_for_no_commits(
350 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
351 ) -> None:
352 _init_muse_repo(tmp_path)
353 entries = await _meter_history_async(session=muse_cli_db_session, root=tmp_path)
354 assert entries == []
355
356
357 @pytest.mark.anyio
358 async def test_meter_history_shows_annotated_and_unannotated_commits(
359 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
360 ) -> None:
361 """History walks the full chain, returning None for unannotated commits."""
362 _init_muse_repo(tmp_path)
363 _populate_workdir(tmp_path, {"beat.mid": b"V1"})
364
365 cid1 = await _commit_async(message="v1", root=tmp_path, session=muse_cli_db_session)
366 await _meter_set_async(
367 session=muse_cli_db_session,
368 root=tmp_path,
369 commit_ref=None,
370 time_signature="4/4",
371 )
372 await muse_cli_db_session.flush()
373
374 (tmp_path / "muse-work" / "beat.mid").write_bytes(b"V2")
375 cid2 = await _commit_async(message="v2", root=tmp_path, session=muse_cli_db_session)
376
377 entries = await _meter_history_async(session=muse_cli_db_session, root=tmp_path)
378
379 assert len(entries) == 2
380 # Newest-first: v2 has no annotation, v1 has 4/4
381 assert entries[0].commit_id == cid2
382 assert entries[0].time_signature is None
383 assert entries[1].commit_id == cid1
384 assert entries[1].time_signature == "4/4"
385
386
387 @pytest.mark.anyio
388 async def test_meter_history_entries_are_muse_meter_history_entry(
389 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
390 ) -> None:
391 _init_muse_repo(tmp_path)
392 _populate_workdir(tmp_path)
393 await _commit_async(message="only commit", root=tmp_path, session=muse_cli_db_session)
394
395 entries = await _meter_history_async(session=muse_cli_db_session, root=tmp_path)
396 assert len(entries) == 1
397 assert isinstance(entries[0], MuseMeterHistoryEntry)
398 assert entries[0].message == "only commit"
399
400
401 # ──────────────────────────────────────────────────────────────────────────────
402 # _meter_polyrhythm_async
403 # ──────────────────────────────────────────────────────────────────────────────
404
405
406 @pytest.mark.anyio
407 async def test_meter_polyrhythm_no_polyrhythm_when_same_signature(
408 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
409 ) -> None:
410 _init_muse_repo(tmp_path)
411 workdir = tmp_path / "muse-work"
412 workdir.mkdir()
413 (workdir / "drums.mid").write_bytes(_make_midi_with_time_sig(4, 2))
414 (workdir / "bass.mid").write_bytes(_make_midi_with_time_sig(4, 2))
415 await _commit_async(message="4/4 all", root=tmp_path, session=muse_cli_db_session)
416
417 result = await _meter_polyrhythm_async(
418 session=muse_cli_db_session, root=tmp_path, commit_ref=None
419 )
420 assert isinstance(result, MusePolyrhythmResult)
421 assert result.is_polyrhythmic is False
422
423
424 @pytest.mark.anyio
425 async def test_meter_polyrhythm_detected_when_mixed_signatures(
426 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
427 ) -> None:
428 _init_muse_repo(tmp_path)
429 workdir = tmp_path / "muse-work"
430 workdir.mkdir()
431 (workdir / "drums.mid").write_bytes(_make_midi_with_time_sig(4, 2)) # 4/4
432 (workdir / "melody.mid").write_bytes(_make_midi_with_time_sig(7, 3)) # 7/8
433 await _commit_async(message="polyrhythm", root=tmp_path, session=muse_cli_db_session)
434
435 result = await _meter_polyrhythm_async(
436 session=muse_cli_db_session, root=tmp_path, commit_ref=None
437 )
438 assert result.is_polyrhythmic is True
439 assert "drums.mid" in result.signatures_by_file
440 assert "melody.mid" in result.signatures_by_file
441 assert result.signatures_by_file["drums.mid"] == "4/4"
442 assert result.signatures_by_file["melody.mid"] == "7/8"
443
444
445 @pytest.mark.anyio
446 async def test_meter_polyrhythm_not_polyrhythmic_when_unknown_only(
447 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
448 ) -> None:
449 """Files with no time-sig meta events are '?' — not considered for polyrhythm."""
450 _init_muse_repo(tmp_path)
451 workdir = tmp_path / "muse-work"
452 workdir.mkdir()
453 (workdir / "a.mid").write_bytes(b"\x00\x90\x3C\x7F")
454 (workdir / "b.mid").write_bytes(b"\x00\x90\x3C\x7F")
455 await _commit_async(message="unknown sigs", root=tmp_path, session=muse_cli_db_session)
456
457 result = await _meter_polyrhythm_async(
458 session=muse_cli_db_session, root=tmp_path, commit_ref=None
459 )
460 assert result.is_polyrhythmic is False
461
462
463 # ──────────────────────────────────────────────────────────────────────────────
464 # CLI integration (Typer runner)
465 # ──────────────────────────────────────────────────────────────────────────────
466
467
468 def test_meter_no_repo_exits_2(tmp_path: pathlib.Path) -> None:
469 """Running muse meter outside a repo exits with REPO_NOT_FOUND (2)."""
470 import os
471
472 from typer.testing import CliRunner
473
474 from maestro.muse_cli.app import cli
475
476 runner = CliRunner()
477 orig = os.getcwd()
478 try:
479 os.chdir(tmp_path)
480 result = runner.invoke(cli, ["meter"], catch_exceptions=False)
481 finally:
482 os.chdir(orig)
483 assert result.exit_code == ExitCode.REPO_NOT_FOUND
484
485
486 def test_validate_time_signature_denominator_zero_raises() -> None:
487 with pytest.raises(ValueError):
488 validate_time_signature("4/0")