cgcardona / muse public
test_muse_humanize.py python
647 lines 18.1 KB
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse humanize`` — CLI interface, flag parsing, and stub output.
2
3 All CLI-level tests use ``typer.testing.CliRunner`` against the full ``muse``
4 app. Async core tests call ``_humanize_async`` directly with an in-memory
5 SQLite session — the stub does not query the DB, so the session satisfies only
6 the signature contract.
7
8 Covered acceptance criteria:
9 - ``--seed 42`` produces identical results every time (deterministic)
10 - ``--natural`` increases timing variance vs no humanization
11 - ``--tight`` stays within its documented bounds
12 - ``--timing-only`` preserves velocity (velocity_range == 0 for all tracks)
13 - ``--velocity-only`` preserves timing (timing_range_ms == 0 for all tracks)
14 - ``--track bass`` leaves other tracks unchanged
15 - Drum channel is excluded from timing variation (drum_channel_excluded=True)
16 """
17 from __future__ import annotations
18
19 import json
20 import os
21 import pathlib
22 import random
23 import uuid
24 from collections.abc import AsyncGenerator
25
26 import pytest
27 import pytest_asyncio
28 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
29 from sqlalchemy.pool import StaticPool
30 from typer.testing import CliRunner
31
32 from maestro.db.database import Base
33 import maestro.muse_cli.models # noqa: F401 — registers MuseCli* ORM models
34 from maestro.muse_cli.app import cli
35 from maestro.muse_cli.commands.humanize import (
36 LOOSE_TIMING_MS,
37 LOOSE_VELOCITY,
38 NATURAL_TIMING_MS,
39 NATURAL_VELOCITY,
40 TIGHT_TIMING_MS,
41 TIGHT_VELOCITY,
42 HumanizeResult,
43 TrackHumanizeResult,
44 _apply_humanization,
45 _humanize_async,
46 _render_json,
47 _render_table,
48 _resolve_preset,
49 _timing_ms_for_factor,
50 _velocity_range_for_factor,
51 )
52 from maestro.muse_cli.errors import ExitCode
53
54 runner = CliRunner()
55
56
57 # ---------------------------------------------------------------------------
58 # Helpers / fixtures
59 # ---------------------------------------------------------------------------
60
61
62 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
63 rid = str(uuid.uuid4())
64 muse = root / ".muse"
65 (muse / "refs" / "heads").mkdir(parents=True)
66 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
67 (muse / "HEAD").write_text(f"refs/heads/{branch}")
68 (muse / "refs" / "heads" / branch).write_text("")
69 return rid
70
71
72 def _commit_ref(root: pathlib.Path, branch: str = "main") -> None:
73 muse = root / ".muse"
74 (muse / "refs" / "heads" / branch).write_text(
75 "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
76 )
77
78
79 @pytest_asyncio.fixture
80 async def db_session() -> AsyncGenerator[AsyncSession, None]:
81 """In-memory SQLite session (stub humanize does not query it)."""
82 engine = create_async_engine(
83 "sqlite+aiosqlite:///:memory:",
84 connect_args={"check_same_thread": False},
85 poolclass=StaticPool,
86 )
87 async with engine.begin() as conn:
88 await conn.run_sync(Base.metadata.create_all)
89 factory = async_sessionmaker(bind=engine, expire_on_commit=False)
90 async with factory() as session:
91 yield session
92 async with engine.begin() as conn:
93 await conn.run_sync(Base.metadata.drop_all)
94 await engine.dispose()
95
96
97 # ---------------------------------------------------------------------------
98 # Unit — _resolve_preset
99 # ---------------------------------------------------------------------------
100
101
102 def test_resolve_preset_default_is_natural() -> None:
103 """With no flags set, default preset is 'natural'."""
104 label, factor = _resolve_preset(tight=False, natural=False, loose=False, factor=None)
105 assert label == "natural"
106 assert 0.0 < factor < 1.0
107
108
109 def test_resolve_preset_tight() -> None:
110 """--tight maps to a factor < natural."""
111 label, factor = _resolve_preset(tight=True, natural=False, loose=False, factor=None)
112 assert label == "tight"
113 assert factor < 0.6
114
115
116 def test_resolve_preset_loose() -> None:
117 """--loose maps to factor=1.0."""
118 label, factor = _resolve_preset(tight=False, natural=False, loose=True, factor=None)
119 assert label == "loose"
120 assert factor == 1.0
121
122
123 def test_resolve_preset_custom_factor() -> None:
124 """--factor overrides preset flags."""
125 label, factor = _resolve_preset(tight=False, natural=False, loose=False, factor=0.42)
126 assert label == "custom"
127 assert factor == 0.42
128
129
130 def test_resolve_preset_multiple_presets_raises() -> None:
131 """Specifying two preset flags simultaneously raises ValueError."""
132 with pytest.raises(ValueError, match="Only one"):
133 _resolve_preset(tight=True, natural=False, loose=True, factor=None)
134
135
136 # ---------------------------------------------------------------------------
137 # Unit — timing / velocity helpers
138 # ---------------------------------------------------------------------------
139
140
141 def test_timing_ms_for_factor_zero() -> None:
142 assert _timing_ms_for_factor(0.0) == 0
143
144
145 def test_timing_ms_for_factor_one() -> None:
146 assert _timing_ms_for_factor(1.0) == LOOSE_TIMING_MS
147
148
149 def test_timing_ms_for_natural_within_ceiling() -> None:
150 assert _timing_ms_for_factor(0.6) <= NATURAL_TIMING_MS
151
152
153 def test_velocity_range_for_factor_zero() -> None:
154 assert _velocity_range_for_factor(0.0) == 0
155
156
157 def test_velocity_range_for_factor_one() -> None:
158 assert _velocity_range_for_factor(1.0) == LOOSE_VELOCITY
159
160
161 # ---------------------------------------------------------------------------
162 # Unit — _apply_humanization
163 # ---------------------------------------------------------------------------
164
165
166 def test_humanize_seed_produces_deterministic_output() -> None:
167 """Regression: identical seeds produce identical TrackHumanizeResult values."""
168 rng_a = random.Random(42)
169 rng_b = random.Random(42)
170 result_a = _apply_humanization(
171 track_name="bass",
172 timing_ms=12,
173 velocity_range=10,
174 timing_only=False,
175 velocity_only=False,
176 rng=rng_a,
177 )
178 result_b = _apply_humanization(
179 track_name="bass",
180 timing_ms=12,
181 velocity_range=10,
182 timing_only=False,
183 velocity_only=False,
184 rng=rng_b,
185 )
186 assert result_a == result_b
187
188
189 def test_humanize_natural_increases_timing_variance() -> None:
190 """--natural produces non-zero timing range for non-drum tracks."""
191 rng = random.Random(7)
192 result = _apply_humanization(
193 track_name="bass",
194 timing_ms=NATURAL_TIMING_MS,
195 velocity_range=NATURAL_VELOCITY,
196 timing_only=False,
197 velocity_only=False,
198 rng=rng,
199 )
200 assert result["timing_range_ms"] > 0
201
202
203 def test_humanize_tight_stays_within_bounds() -> None:
204 """--tight result stays within TIGHT_TIMING_MS and TIGHT_VELOCITY."""
205 rng = random.Random(1)
206 timing_ms = _timing_ms_for_factor(0.25)
207 vel_range = _velocity_range_for_factor(0.25)
208 result = _apply_humanization(
209 track_name="keys",
210 timing_ms=timing_ms,
211 velocity_range=vel_range,
212 timing_only=False,
213 velocity_only=False,
214 rng=rng,
215 )
216 assert result["timing_range_ms"] <= TIGHT_TIMING_MS
217 assert result["velocity_range"] <= TIGHT_VELOCITY
218
219
220 def test_humanize_timing_only_preserves_velocity() -> None:
221 """--timing-only sets velocity_range to 0."""
222 rng = random.Random(2)
223 result = _apply_humanization(
224 track_name="lead",
225 timing_ms=12,
226 velocity_range=10,
227 timing_only=True,
228 velocity_only=False,
229 rng=rng,
230 )
231 assert result["velocity_range"] == 0
232
233
234 def test_humanize_velocity_only_preserves_timing() -> None:
235 """--velocity-only sets timing_range_ms to 0."""
236 rng = random.Random(3)
237 result = _apply_humanization(
238 track_name="lead",
239 timing_ms=12,
240 velocity_range=10,
241 timing_only=False,
242 velocity_only=True,
243 rng=rng,
244 )
245 assert result["timing_range_ms"] == 0
246
247
248 def test_humanize_drum_channel_excluded_from_timing_variation() -> None:
249 """Drum track is excluded from timing variation."""
250 rng = random.Random(4)
251 result = _apply_humanization(
252 track_name="drums",
253 timing_ms=12,
254 velocity_range=10,
255 timing_only=False,
256 velocity_only=False,
257 rng=rng,
258 )
259 assert result["drum_channel_excluded"] is True
260 assert result["timing_range_ms"] == 0
261
262
263 def test_humanize_drum_channel_velocity_applied() -> None:
264 """Drum track still receives velocity humanization."""
265 rng = random.Random(5)
266 result = _apply_humanization(
267 track_name="drums",
268 timing_ms=12,
269 velocity_range=10,
270 timing_only=False,
271 velocity_only=False,
272 rng=rng,
273 )
274 assert result["velocity_range"] > 0
275
276
277 # ---------------------------------------------------------------------------
278 # Async core — _humanize_async
279 # ---------------------------------------------------------------------------
280
281
282 @pytest.mark.anyio
283 async def test_humanize_async_default_output(
284 tmp_path: pathlib.Path,
285 db_session: AsyncSession,
286 capsys: pytest.CaptureFixture[str],
287 ) -> None:
288 """_humanize_async with defaults renders a table with all four stub tracks."""
289 _init_muse_repo(tmp_path)
290 _commit_ref(tmp_path)
291
292 result = await _humanize_async(
293 root=tmp_path,
294 session=db_session,
295 source_commit=None,
296 preset="natural",
297 factor=0.6,
298 seed=None,
299 timing_only=False,
300 velocity_only=False,
301 track=None,
302 section=None,
303 message=None,
304 as_json=False,
305 )
306
307 out = capsys.readouterr().out
308 assert "Humanize" in out
309 assert "drums" in out
310 assert "bass" in out
311 assert result["preset"] == "natural"
312 assert len(result["tracks"]) == 4
313
314
315 @pytest.mark.anyio
316 async def test_humanize_async_seed_deterministic(
317 tmp_path: pathlib.Path,
318 db_session: AsyncSession,
319 capsys: pytest.CaptureFixture[str],
320 ) -> None:
321 """--seed 42 produces identical commit IDs across two invocations."""
322 _init_muse_repo(tmp_path)
323 _commit_ref(tmp_path)
324
325 result_a = await _humanize_async(
326 root=tmp_path,
327 session=db_session,
328 source_commit=None,
329 preset="natural",
330 factor=0.6,
331 seed=42,
332 timing_only=False,
333 velocity_only=False,
334 track=None,
335 section=None,
336 message=None,
337 as_json=False,
338 )
339 capsys.readouterr()
340
341 result_b = await _humanize_async(
342 root=tmp_path,
343 session=db_session,
344 source_commit=None,
345 preset="natural",
346 factor=0.6,
347 seed=42,
348 timing_only=False,
349 velocity_only=False,
350 track=None,
351 section=None,
352 message=None,
353 as_json=False,
354 )
355
356 assert result_a["new_commit_id"] == result_b["new_commit_id"]
357 assert result_a["tracks"] == result_b["tracks"]
358
359
360 @pytest.mark.anyio
361 async def test_humanize_async_track_scoped_leaves_other_tracks_unchanged(
362 tmp_path: pathlib.Path,
363 db_session: AsyncSession,
364 capsys: pytest.CaptureFixture[str],
365 ) -> None:
366 """--track bass: only the bass track appears in the result."""
367 _init_muse_repo(tmp_path)
368 _commit_ref(tmp_path)
369
370 result = await _humanize_async(
371 root=tmp_path,
372 session=db_session,
373 source_commit=None,
374 preset="natural",
375 factor=0.6,
376 seed=10,
377 timing_only=False,
378 velocity_only=False,
379 track="bass",
380 section=None,
381 message=None,
382 as_json=False,
383 )
384
385 track_names = [t["track"] for t in result["tracks"]]
386 assert track_names == ["bass"]
387
388
389 @pytest.mark.anyio
390 async def test_humanize_async_json_mode(
391 tmp_path: pathlib.Path,
392 db_session: AsyncSession,
393 capsys: pytest.CaptureFixture[str],
394 ) -> None:
395 """--json emits parseable JSON with the expected top-level keys."""
396 _init_muse_repo(tmp_path)
397 _commit_ref(tmp_path)
398
399 await _humanize_async(
400 root=tmp_path,
401 session=db_session,
402 source_commit=None,
403 preset="tight",
404 factor=0.25,
405 seed=99,
406 timing_only=False,
407 velocity_only=False,
408 track=None,
409 section=None,
410 message=None,
411 as_json=True,
412 )
413
414 raw = capsys.readouterr().out
415 payload = json.loads(raw)
416 for key in ("commit", "branch", "preset", "factor", "seed", "tracks", "new_commit_id"):
417 assert key in payload, f"Missing key: {key}"
418 assert payload["preset"] == "tight"
419 assert isinstance(payload["tracks"], list)
420
421
422 @pytest.mark.anyio
423 async def test_humanize_async_timing_only_all_velocities_zero(
424 tmp_path: pathlib.Path,
425 db_session: AsyncSession,
426 capsys: pytest.CaptureFixture[str],
427 ) -> None:
428 """--timing-only: all track results have velocity_range == 0."""
429 _init_muse_repo(tmp_path)
430 _commit_ref(tmp_path)
431
432 result = await _humanize_async(
433 root=tmp_path,
434 session=db_session,
435 source_commit=None,
436 preset="natural",
437 factor=0.6,
438 seed=1,
439 timing_only=True,
440 velocity_only=False,
441 track=None,
442 section=None,
443 message=None,
444 as_json=False,
445 )
446
447 for tr in result["tracks"]:
448 assert tr["velocity_range"] == 0, f"{tr['track']}: expected velocity_range=0"
449
450
451 @pytest.mark.anyio
452 async def test_humanize_async_velocity_only_all_timing_zero(
453 tmp_path: pathlib.Path,
454 db_session: AsyncSession,
455 capsys: pytest.CaptureFixture[str],
456 ) -> None:
457 """--velocity-only: all track results have timing_range_ms == 0."""
458 _init_muse_repo(tmp_path)
459 _commit_ref(tmp_path)
460
461 result = await _humanize_async(
462 root=tmp_path,
463 session=db_session,
464 source_commit=None,
465 preset="natural",
466 factor=0.6,
467 seed=1,
468 timing_only=False,
469 velocity_only=True,
470 track=None,
471 section=None,
472 message=None,
473 as_json=False,
474 )
475
476 for tr in result["tracks"]:
477 assert tr["timing_range_ms"] == 0, f"{tr['track']}: expected timing_range_ms=0"
478
479
480 @pytest.mark.anyio
481 async def test_humanize_async_explicit_source_commit_in_output(
482 tmp_path: pathlib.Path,
483 db_session: AsyncSession,
484 capsys: pytest.CaptureFixture[str],
485 ) -> None:
486 """An explicit source commit ref appears in the rendered output."""
487 _init_muse_repo(tmp_path)
488 _commit_ref(tmp_path)
489
490 await _humanize_async(
491 root=tmp_path,
492 session=db_session,
493 source_commit="deadbeef",
494 preset="natural",
495 factor=0.6,
496 seed=None,
497 timing_only=False,
498 velocity_only=False,
499 track=None,
500 section=None,
501 message=None,
502 as_json=False,
503 )
504
505 out = capsys.readouterr().out
506 assert "deadbeef" in out
507
508
509 # ---------------------------------------------------------------------------
510 # Renderer unit tests
511 # ---------------------------------------------------------------------------
512
513
514 def _make_result() -> HumanizeResult:
515 tracks: list[TrackHumanizeResult] = [
516 TrackHumanizeResult(
517 track="bass",
518 timing_range_ms=12,
519 velocity_range=10,
520 notes_affected=64,
521 drum_channel_excluded=False,
522 ),
523 TrackHumanizeResult(
524 track="drums",
525 timing_range_ms=0,
526 velocity_range=10,
527 notes_affected=48,
528 drum_channel_excluded=True,
529 ),
530 ]
531 return HumanizeResult(
532 commit="a1b2c3d4",
533 branch="main",
534 source_commit="a1b2c3d4",
535 preset="natural",
536 factor=0.6,
537 seed=None,
538 timing_only=False,
539 velocity_only=False,
540 track_filter=None,
541 section_filter=None,
542 tracks=tracks,
543 new_commit_id="deadbeef",
544 )
545
546
547 def test_render_table_includes_commit_and_preset(
548 capsys: pytest.CaptureFixture[str],
549 ) -> None:
550 _render_table(_make_result())
551 out = capsys.readouterr().out
552 assert "natural" in out
553 assert "a1b2c3d4" in out
554
555
556 def test_render_table_shows_drum_excluded(
557 capsys: pytest.CaptureFixture[str],
558 ) -> None:
559 _render_table(_make_result())
560 out = capsys.readouterr().out
561 assert "yes" in out
562
563
564 def test_render_json_is_valid(capsys: pytest.CaptureFixture[str]) -> None:
565 _render_json(_make_result())
566 raw = capsys.readouterr().out
567 payload = json.loads(raw)
568 assert payload["preset"] == "natural"
569 assert payload["tracks"][0]["track"] == "bass"
570
571
572 # ---------------------------------------------------------------------------
573 # CLI integration — CliRunner
574 # ---------------------------------------------------------------------------
575
576
577 def test_cli_humanize_appears_in_muse_help() -> None:
578 result = runner.invoke(cli, ["--help"])
579 assert result.exit_code == 0
580 assert "humanize" in result.output
581
582
583 def test_cli_humanize_help_lists_flags() -> None:
584 result = runner.invoke(cli, ["humanize", "--help"])
585 assert result.exit_code == 0
586 for flag in (
587 "--tight",
588 "--natural",
589 "--loose",
590 "--factor",
591 "--timing-only",
592 "--velocity-only",
593 "--track",
594 "--section",
595 "--seed",
596 "--message",
597 "--json",
598 ):
599 assert flag in result.output, f"Flag '{flag}' missing from help"
600
601
602 def test_cli_humanize_outside_repo_exits_repo_not_found(tmp_path: pathlib.Path) -> None:
603 prev = os.getcwd()
604 try:
605 os.chdir(tmp_path)
606 result = runner.invoke(cli, ["humanize"], catch_exceptions=False)
607 finally:
608 os.chdir(prev)
609
610 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND)
611 assert "not a muse repository" in result.output.lower()
612
613
614 def test_cli_humanize_timing_and_velocity_only_mutually_exclusive(
615 tmp_path: pathlib.Path,
616 ) -> None:
617 _init_muse_repo(tmp_path)
618 _commit_ref(tmp_path)
619 prev = os.getcwd()
620 try:
621 os.chdir(tmp_path)
622 result = runner.invoke(
623 cli,
624 ["humanize", "--timing-only", "--velocity-only"],
625 catch_exceptions=False,
626 )
627 finally:
628 os.chdir(prev)
629
630 assert result.exit_code == int(ExitCode.USER_ERROR)
631
632
633 def test_cli_humanize_two_presets_exits_user_error(tmp_path: pathlib.Path) -> None:
634 _init_muse_repo(tmp_path)
635 _commit_ref(tmp_path)
636 prev = os.getcwd()
637 try:
638 os.chdir(tmp_path)
639 result = runner.invoke(
640 cli,
641 ["humanize", "--tight", "--loose"],
642 catch_exceptions=False,
643 )
644 finally:
645 os.chdir(prev)
646
647 assert result.exit_code == int(ExitCode.USER_ERROR)