cgcardona / muse public
test_tempo_scale.py python
619 lines 19.5 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse tempo-scale`` — timing stretch/compress command.
2
3 Covers:
4 - compute_factor_from_bpm: factor computation, edge cases, errors
5 - apply_factor: correct BPM calculation
6 - _tempo_scale_async: result schema, factor resolution, determinism
7 - _format_result: text and JSON output modes
8 - CLI flag parsing via CliRunner (factor, commit, --bpm, --track,
9 --preserve-expressions, --message, --json)
10 - Validation errors: no args, mutual exclusion, out-of-range factor,
11 non-positive BPM
12 - Outside-repo invocation exits 2
13
14 All async tests use @pytest.mark.anyio with the shared muse_cli_db_session
15 fixture from tests/muse_cli/conftest.py.
16 """
17 from __future__ import annotations
18
19 import json
20 import os
21 import pathlib
22 import uuid
23
24 import pytest
25 from sqlalchemy.ext.asyncio import AsyncSession
26 from typer.testing import CliRunner
27
28 from maestro.muse_cli.app import cli
29 from maestro.muse_cli.commands.tempo_scale import (
30 FACTOR_MAX,
31 FACTOR_MIN,
32 TempoScaleResult,
33 _format_result,
34 _tempo_scale_async,
35 apply_factor,
36 compute_factor_from_bpm,
37 )
38 from maestro.muse_cli.errors import ExitCode
39
40 runner = CliRunner()
41
42
43 # ---------------------------------------------------------------------------
44 # Helpers
45 # ---------------------------------------------------------------------------
46
47
48 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
49 """Create a minimal .muse/ layout with one commit ref and return repo_id."""
50 rid = str(uuid.uuid4())
51 muse = root / ".muse"
52 (muse / "refs" / "heads").mkdir(parents=True)
53 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
54 (muse / "HEAD").write_text(f"refs/heads/{branch}")
55 (muse / "refs" / "heads" / branch).write_text("abc12345")
56 return rid
57
58
59 # ---------------------------------------------------------------------------
60 # compute_factor_from_bpm — pure function
61 # ---------------------------------------------------------------------------
62
63
64 def test_compute_factor_from_bpm_120_to_128() -> None:
65 """120 BPM to 128 BPM yields factor 128/120."""
66 factor = compute_factor_from_bpm(120.0, 128.0)
67 assert abs(factor - 128.0 / 120.0) < 1e-9
68
69
70 def test_tempo_scale_bpm_target_computes_correct_factor() -> None:
71 """Regression: --bpm 128 starting from 120 BPM produces factor ~1.0667."""
72 factor = compute_factor_from_bpm(120.0, 128.0)
73 assert abs(factor - (128.0 / 120.0)) < 1e-9, (
74 f"Expected {128.0 / 120.0}, got {factor}"
75 )
76
77
78 def test_compute_factor_from_bpm_half_time() -> None:
79 """Targeting 60 BPM from 120 yields 0.5 (half-time)."""
80 factor = compute_factor_from_bpm(120.0, 60.0)
81 assert abs(factor - 0.5) < 1e-9
82
83
84 def test_compute_factor_from_bpm_double_time() -> None:
85 """Targeting 240 BPM from 120 yields 2.0 (double-time)."""
86 factor = compute_factor_from_bpm(120.0, 240.0)
87 assert abs(factor - 2.0) < 1e-9
88
89
90 def test_compute_factor_from_bpm_same_bpm_yields_one() -> None:
91 """No change: source == target yields factor 1.0."""
92 factor = compute_factor_from_bpm(120.0, 120.0)
93 assert abs(factor - 1.0) < 1e-9
94
95
96 def test_compute_factor_from_bpm_raises_on_zero_source() -> None:
97 """Zero source BPM raises ValueError."""
98 with pytest.raises(ValueError, match="source_bpm must be positive"):
99 compute_factor_from_bpm(0.0, 128.0)
100
101
102 def test_compute_factor_from_bpm_raises_on_negative_source() -> None:
103 """Negative source BPM raises ValueError."""
104 with pytest.raises(ValueError, match="source_bpm must be positive"):
105 compute_factor_from_bpm(-10.0, 128.0)
106
107
108 def test_compute_factor_from_bpm_raises_on_zero_target() -> None:
109 """Zero target BPM raises ValueError."""
110 with pytest.raises(ValueError, match="target_bpm must be positive"):
111 compute_factor_from_bpm(120.0, 0.0)
112
113
114 def test_compute_factor_from_bpm_raises_on_negative_target() -> None:
115 """Negative target BPM raises ValueError."""
116 with pytest.raises(ValueError, match="target_bpm must be positive"):
117 compute_factor_from_bpm(120.0, -1.0)
118
119
120 # ---------------------------------------------------------------------------
121 # apply_factor — pure function
122 # ---------------------------------------------------------------------------
123
124
125 def test_apply_factor_double_time() -> None:
126 """Factor 2.0 doubles the BPM."""
127 assert apply_factor(120.0, 2.0) == 240.0
128
129
130 def test_apply_factor_half_time() -> None:
131 """Factor 0.5 halves the BPM."""
132 assert apply_factor(120.0, 0.5) == 60.0
133
134
135 def test_apply_factor_identity() -> None:
136 """Factor 1.0 leaves BPM unchanged."""
137 assert apply_factor(120.0, 1.0) == 120.0
138
139
140 def test_tempo_scale_half_time_doubles_all_offsets() -> None:
141 """Regression: factor 0.5 halves BPM (half-time = twice as slow)."""
142 result = apply_factor(120.0, 0.5)
143 assert result == 60.0, f"Expected 60.0, got {result}"
144
145
146 def test_apply_factor_128_bpm() -> None:
147 """Factor 128/120 starting from 120 BPM yields exactly 128 BPM."""
148 factor = compute_factor_from_bpm(120.0, 128.0)
149 new_bpm = apply_factor(120.0, factor)
150 assert abs(new_bpm - 128.0) < 0.0001
151
152
153 # ---------------------------------------------------------------------------
154 # _tempo_scale_async — schema and behaviour
155 # ---------------------------------------------------------------------------
156
157
158 @pytest.mark.anyio
159 async def test_tempo_scale_factor_updates_note_timings(
160 tmp_path: pathlib.Path,
161 muse_cli_db_session: AsyncSession,
162 ) -> None:
163 """_tempo_scale_async returns a result with the correct factor applied."""
164 _init_muse_repo(tmp_path)
165 result = await _tempo_scale_async(
166 root=tmp_path,
167 session=muse_cli_db_session,
168 commit=None,
169 factor=2.0,
170 bpm=None,
171 track=None,
172 preserve_expressions=False,
173 message=None,
174 )
175 assert result["factor"] == 2.0
176 assert result["new_bpm"] == apply_factor(result["source_bpm"], 2.0)
177
178
179 @pytest.mark.anyio
180 async def test_tempo_scale_bpm_128_from_120_bpm(
181 tmp_path: pathlib.Path,
182 muse_cli_db_session: AsyncSession,
183 ) -> None:
184 """--bpm 128 from a 120 BPM source produces new_bpm == 128."""
185 _init_muse_repo(tmp_path)
186 result = await _tempo_scale_async(
187 root=tmp_path,
188 session=muse_cli_db_session,
189 commit=None,
190 factor=None,
191 bpm=128.0,
192 track=None,
193 preserve_expressions=False,
194 message=None,
195 )
196 assert abs(result["new_bpm"] - 128.0) < 0.0001
197 assert result["source_bpm"] == 120.0
198
199
200 @pytest.mark.anyio
201 async def test_tempo_scale_updates_commit_tempo_metadata(
202 tmp_path: pathlib.Path,
203 muse_cli_db_session: AsyncSession,
204 ) -> None:
205 """Result contains source_commit, new_commit, and they differ."""
206 _init_muse_repo(tmp_path)
207 result = await _tempo_scale_async(
208 root=tmp_path,
209 session=muse_cli_db_session,
210 commit=None,
211 factor=1.5,
212 bpm=None,
213 track=None,
214 preserve_expressions=False,
215 message=None,
216 )
217 assert result["source_commit"]
218 assert result["new_commit"]
219 assert result["source_commit"] != result["new_commit"]
220
221
222 @pytest.mark.anyio
223 async def test_tempo_scale_preserve_expressions_scales_cc_events(
224 tmp_path: pathlib.Path,
225 muse_cli_db_session: AsyncSession,
226 ) -> None:
227 """preserve_expressions flag is reflected in the result."""
228 _init_muse_repo(tmp_path)
229 result = await _tempo_scale_async(
230 root=tmp_path,
231 session=muse_cli_db_session,
232 commit=None,
233 factor=0.5,
234 bpm=None,
235 track=None,
236 preserve_expressions=True,
237 message=None,
238 )
239 assert result["preserve_expressions"] is True
240
241
242 @pytest.mark.anyio
243 async def test_tempo_scale_returns_all_schema_keys(
244 tmp_path: pathlib.Path,
245 muse_cli_db_session: AsyncSession,
246 ) -> None:
247 """TempoScaleResult contains all expected keys."""
248 _init_muse_repo(tmp_path)
249 result = await _tempo_scale_async(
250 root=tmp_path,
251 session=muse_cli_db_session,
252 commit=None,
253 factor=1.0,
254 bpm=None,
255 track=None,
256 preserve_expressions=False,
257 message=None,
258 )
259 for key in (
260 "source_commit",
261 "new_commit",
262 "factor",
263 "source_bpm",
264 "new_bpm",
265 "track",
266 "preserve_expressions",
267 "message",
268 ):
269 assert key in result, f"Missing key: {key}"
270
271
272 @pytest.mark.anyio
273 async def test_tempo_scale_deterministic_output(
274 tmp_path: pathlib.Path,
275 muse_cli_db_session: AsyncSession,
276 ) -> None:
277 """Same inputs always produce the same new_commit (deterministic)."""
278 _init_muse_repo(tmp_path)
279 result_a = await _tempo_scale_async(
280 root=tmp_path,
281 session=muse_cli_db_session,
282 commit="abc12345",
283 factor=0.5,
284 bpm=None,
285 track="bass",
286 preserve_expressions=False,
287 message=None,
288 )
289 result_b = await _tempo_scale_async(
290 root=tmp_path,
291 session=muse_cli_db_session,
292 commit="abc12345",
293 factor=0.5,
294 bpm=None,
295 track="bass",
296 preserve_expressions=False,
297 message=None,
298 )
299 assert result_a["new_commit"] == result_b["new_commit"]
300
301
302 @pytest.mark.anyio
303 async def test_tempo_scale_custom_message_reflected(
304 tmp_path: pathlib.Path,
305 muse_cli_db_session: AsyncSession,
306 ) -> None:
307 """Custom --message is reflected in the result."""
308 _init_muse_repo(tmp_path)
309 result = await _tempo_scale_async(
310 root=tmp_path,
311 session=muse_cli_db_session,
312 commit=None,
313 factor=2.0,
314 bpm=None,
315 track=None,
316 preserve_expressions=False,
317 message="my custom message",
318 )
319 assert result["message"] == "my custom message"
320
321
322 @pytest.mark.anyio
323 async def test_tempo_scale_track_reflected(
324 tmp_path: pathlib.Path,
325 muse_cli_db_session: AsyncSession,
326 ) -> None:
327 """--track filter is reflected in the result."""
328 _init_muse_repo(tmp_path)
329 result = await _tempo_scale_async(
330 root=tmp_path,
331 session=muse_cli_db_session,
332 commit=None,
333 factor=2.0,
334 bpm=None,
335 track="keys",
336 preserve_expressions=False,
337 message=None,
338 )
339 assert result["track"] == "keys"
340
341
342 @pytest.mark.anyio
343 async def test_tempo_scale_no_track_defaults_to_all(
344 tmp_path: pathlib.Path,
345 muse_cli_db_session: AsyncSession,
346 ) -> None:
347 """Without --track, the result track field is 'all'."""
348 _init_muse_repo(tmp_path)
349 result = await _tempo_scale_async(
350 root=tmp_path,
351 session=muse_cli_db_session,
352 commit=None,
353 factor=1.0,
354 bpm=None,
355 track=None,
356 preserve_expressions=False,
357 message=None,
358 )
359 assert result["track"] == "all"
360
361
362 @pytest.mark.anyio
363 async def test_tempo_scale_raises_if_no_factor_and_no_bpm(
364 tmp_path: pathlib.Path,
365 muse_cli_db_session: AsyncSession,
366 ) -> None:
367 """ValueError raised when neither factor nor --bpm is provided."""
368 _init_muse_repo(tmp_path)
369 with pytest.raises(ValueError, match="Either factor or --bpm must be provided"):
370 await _tempo_scale_async(
371 root=tmp_path,
372 session=muse_cli_db_session,
373 commit=None,
374 factor=None,
375 bpm=None,
376 track=None,
377 preserve_expressions=False,
378 message=None,
379 )
380
381
382 # ---------------------------------------------------------------------------
383 # _format_result — output formatters
384 # ---------------------------------------------------------------------------
385
386
387 def _make_result(
388 factor: float = 2.0,
389 source_bpm: float = 120.0,
390 preserve_expressions: bool = False,
391 ) -> TempoScaleResult:
392 return TempoScaleResult(
393 source_commit="abc12345",
394 new_commit="deadbeef",
395 factor=factor,
396 source_bpm=source_bpm,
397 new_bpm=apply_factor(source_bpm, factor),
398 track="all",
399 preserve_expressions=preserve_expressions,
400 message="test message",
401 )
402
403
404 def test_format_result_json_is_valid() -> None:
405 """JSON output is parseable and contains all expected keys."""
406 result = _make_result()
407 output = _format_result(result, as_json=True)
408 parsed = json.loads(output)
409 for key in TempoScaleResult.__annotations__:
410 assert key in parsed, f"Missing key in JSON: {key}"
411
412
413 def test_format_result_text_contains_source_and_new_commit() -> None:
414 """Text output mentions both the source and new commit."""
415 result = _make_result()
416 output = _format_result(result, as_json=False)
417 assert "abc12345" in output
418 assert "deadbeef" in output
419
420
421 def test_format_result_text_contains_bpm_values() -> None:
422 """Text output includes both source and new BPM."""
423 result = _make_result(factor=2.0, source_bpm=120.0)
424 output = _format_result(result, as_json=False)
425 assert "120.0" in output
426 assert "240.0" in output
427
428
429 def test_format_result_text_shows_preserve_expressions() -> None:
430 """Text output notes expression scaling when the flag is set."""
431 result = _make_result(preserve_expressions=True)
432 output = _format_result(result, as_json=False)
433 assert "Expressions" in output or "expression" in output.lower()
434
435
436 def test_format_result_text_no_preserve_expressions_not_shown() -> None:
437 """Text output does NOT mention expressions when flag is off."""
438 result = _make_result(preserve_expressions=False)
439 output = _format_result(result, as_json=False)
440 assert "expression" not in output.lower()
441
442
443 # ---------------------------------------------------------------------------
444 # CLI integration — CliRunner
445 # ---------------------------------------------------------------------------
446
447
448 def test_cli_tempo_scale_factor_basic(tmp_path: pathlib.Path) -> None:
449 """``muse tempo-scale 2.0`` with a valid repo exits 0 and shows output."""
450 _init_muse_repo(tmp_path)
451 result = runner.invoke(cli, ["tempo-scale", "2.0"], env={"MUSE_REPO_ROOT": str(tmp_path)})
452 assert result.exit_code == ExitCode.SUCCESS, result.output
453 assert "Tempo scaled" in result.output
454
455
456 def test_cli_tempo_scale_half_time(tmp_path: pathlib.Path) -> None:
457 """``muse tempo-scale 0.5`` creates a half-time feel commit."""
458 _init_muse_repo(tmp_path)
459 result = runner.invoke(cli, ["tempo-scale", "0.5"], env={"MUSE_REPO_ROOT": str(tmp_path)})
460 assert result.exit_code == ExitCode.SUCCESS, result.output
461 assert "60.0 BPM" in result.output
462
463
464 def test_cli_tempo_scale_bpm_128(tmp_path: pathlib.Path) -> None:
465 """``muse tempo-scale --bpm 128`` scales to 128 BPM."""
466 _init_muse_repo(tmp_path)
467 result = runner.invoke(
468 cli,
469 ["tempo-scale", "--bpm", "128"],
470 env={"MUSE_REPO_ROOT": str(tmp_path)},
471 )
472 assert result.exit_code == ExitCode.SUCCESS, result.output
473 assert "128.0 BPM" in result.output
474
475
476 def test_cli_tempo_scale_json_output(tmp_path: pathlib.Path) -> None:
477 """``--json`` flag emits valid JSON with all expected keys.
478
479 Options are placed before the positional <factor> argument because Click
480 Groups disable interspersed-args parsing by default.
481 """
482 _init_muse_repo(tmp_path)
483 result = runner.invoke(
484 cli,
485 ["tempo-scale", "--json", "2.0"],
486 env={"MUSE_REPO_ROOT": str(tmp_path)},
487 )
488 assert result.exit_code == ExitCode.SUCCESS, result.output
489 parsed = json.loads(result.output)
490 assert "source_commit" in parsed
491 assert "new_commit" in parsed
492 assert "factor" in parsed
493 assert "new_bpm" in parsed
494
495
496 def test_cli_tempo_scale_with_commit_sha(tmp_path: pathlib.Path) -> None:
497 """Passing an explicit commit SHA is accepted."""
498 _init_muse_repo(tmp_path)
499 result = runner.invoke(
500 cli,
501 ["tempo-scale", "1.5", "deadbeef"],
502 env={"MUSE_REPO_ROOT": str(tmp_path)},
503 )
504 assert result.exit_code == ExitCode.SUCCESS, result.output
505
506
507 def test_cli_tempo_scale_with_track(tmp_path: pathlib.Path) -> None:
508 """``--track bass`` is accepted and reflected in JSON output.
509
510 Options precede the positional factor to satisfy Click Group parsing.
511 """
512 _init_muse_repo(tmp_path)
513 result = runner.invoke(
514 cli,
515 ["tempo-scale", "--track", "bass", "--json", "2.0"],
516 env={"MUSE_REPO_ROOT": str(tmp_path)},
517 )
518 assert result.exit_code == ExitCode.SUCCESS, result.output
519 parsed = json.loads(result.output)
520 assert parsed["track"] == "bass"
521
522
523 def test_cli_tempo_scale_preserve_expressions(tmp_path: pathlib.Path) -> None:
524 """``--preserve-expressions`` flag sets the flag in JSON output.
525
526 Options precede the positional factor to satisfy Click Group parsing.
527 """
528 _init_muse_repo(tmp_path)
529 result = runner.invoke(
530 cli,
531 ["tempo-scale", "--preserve-expressions", "--json", "0.5"],
532 env={"MUSE_REPO_ROOT": str(tmp_path)},
533 )
534 assert result.exit_code == ExitCode.SUCCESS, result.output
535 parsed = json.loads(result.output)
536 assert parsed["preserve_expressions"] is True
537
538
539 def test_cli_tempo_scale_custom_message(tmp_path: pathlib.Path) -> None:
540 """``--message`` is stored in JSON output.
541
542 Options precede the positional factor to satisfy Click Group parsing.
543 """
544 _init_muse_repo(tmp_path)
545 result = runner.invoke(
546 cli,
547 ["tempo-scale", "--message", "half-time remix", "--json", "2.0"],
548 env={"MUSE_REPO_ROOT": str(tmp_path)},
549 )
550 assert result.exit_code == ExitCode.SUCCESS, result.output
551 parsed = json.loads(result.output)
552 assert parsed["message"] == "half-time remix"
553
554
555 def test_cli_tempo_scale_no_args_exits_user_error(tmp_path: pathlib.Path) -> None:
556 """No factor and no --bpm exits with USER_ERROR (1)."""
557 _init_muse_repo(tmp_path)
558 result = runner.invoke(cli, ["tempo-scale"], env={"MUSE_REPO_ROOT": str(tmp_path)})
559 assert result.exit_code == ExitCode.USER_ERROR
560
561
562 def test_cli_tempo_scale_factor_and_bpm_mutually_exclusive(tmp_path: pathlib.Path) -> None:
563 """Providing both <factor> and --bpm exits with USER_ERROR (1).
564
565 ``--bpm`` is placed before the positional factor so Click parses it as
566 an option (not a positional arg) and our mutual-exclusion check fires.
567 """
568 _init_muse_repo(tmp_path)
569 result = runner.invoke(
570 cli,
571 ["tempo-scale", "--bpm", "128", "2.0"],
572 env={"MUSE_REPO_ROOT": str(tmp_path)},
573 )
574 assert result.exit_code == ExitCode.USER_ERROR
575
576
577 def test_cli_tempo_scale_out_of_range_factor(tmp_path: pathlib.Path) -> None:
578 """Factor outside [FACTOR_MIN, FACTOR_MAX] exits USER_ERROR (1)."""
579 _init_muse_repo(tmp_path)
580 result = runner.invoke(
581 cli,
582 ["tempo-scale", "999999"],
583 env={"MUSE_REPO_ROOT": str(tmp_path)},
584 )
585 assert result.exit_code == ExitCode.USER_ERROR
586
587
588 def test_cli_tempo_scale_zero_factor_exits_user_error(tmp_path: pathlib.Path) -> None:
589 """Factor of 0 (below FACTOR_MIN) exits USER_ERROR (1)."""
590 _init_muse_repo(tmp_path)
591 result = runner.invoke(
592 cli,
593 ["tempo-scale", "0.0"],
594 env={"MUSE_REPO_ROOT": str(tmp_path)},
595 )
596 assert result.exit_code == ExitCode.USER_ERROR
597
598
599 def test_cli_tempo_scale_non_positive_bpm_exits_user_error(tmp_path: pathlib.Path) -> None:
600 """--bpm 0 exits USER_ERROR (1)."""
601 _init_muse_repo(tmp_path)
602 result = runner.invoke(
603 cli,
604 ["tempo-scale", "--bpm", "0"],
605 env={"MUSE_REPO_ROOT": str(tmp_path)},
606 )
607 assert result.exit_code == ExitCode.USER_ERROR
608
609
610 def test_cli_tempo_scale_outside_repo_exits_repo_not_found(tmp_path: pathlib.Path) -> None:
611 """Running outside a Muse repo exits REPO_NOT_FOUND (2)."""
612 empty = tmp_path / "no_muse_here"
613 empty.mkdir()
614 result = runner.invoke(
615 cli,
616 ["tempo-scale", "2.0"],
617 env={"MUSE_REPO_ROOT": str(empty)},
618 )
619 assert result.exit_code == ExitCode.REPO_NOT_FOUND