cgcardona / muse public
test_harmony.py python
734 lines 22.0 KB
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse harmony`` — CLI interface, flag parsing, and stub output format.
2
3 All CLI-level tests use ``typer.testing.CliRunner`` against the full ``muse``
4 app so that argument parsing, flag handling, and exit codes are exercised
5 end-to-end.
6
7 Async core tests call ``_harmony_analyze_async`` directly with an in-memory
8 SQLite session (the stub does not query the DB, so the session satisfies
9 the signature contract only).
10 """
11 from __future__ import annotations
12
13 import json
14 import os
15 import pathlib
16 import uuid
17
18 import pytest
19 import pytest_asyncio
20 from collections.abc import AsyncGenerator
21 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
22 from sqlalchemy.pool import StaticPool
23 from typer.testing import CliRunner
24
25 from maestro.db.database import Base
26 import maestro.muse_cli.models # noqa: F401 — registers MuseCli* with Base.metadata
27 from maestro.muse_cli.app import cli
28 from maestro.muse_cli.commands.harmony import (
29 KNOWN_MODES,
30 KNOWN_MODES_SET,
31 HarmonyCompareResult,
32 HarmonyResult,
33 _harmony_analyze_async,
34 _render_compare_human,
35 _render_compare_json,
36 _render_result_human,
37 _render_result_json,
38 _stub_harmony,
39 _tension_label,
40 )
41 from maestro.muse_cli.errors import ExitCode
42
43 runner = CliRunner()
44
45
46 # ---------------------------------------------------------------------------
47 # Fixtures
48 # ---------------------------------------------------------------------------
49
50
51 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
52 """Create a minimal .muse/ layout with one empty commit ref."""
53 rid = str(uuid.uuid4())
54 muse = root / ".muse"
55 (muse / "refs" / "heads").mkdir(parents=True)
56 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
57 (muse / "HEAD").write_text(f"refs/heads/{branch}")
58 (muse / "refs" / "heads" / branch).write_text("")
59 return rid
60
61
62 def _commit_ref(root: pathlib.Path, branch: str = "main") -> None:
63 """Write a fake commit ID into the branch ref so HEAD is non-empty."""
64 muse = root / ".muse"
65 (muse / "refs" / "heads" / branch).write_text("a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2")
66
67
68 @pytest_asyncio.fixture
69 async def db_session() -> AsyncGenerator[AsyncSession, None]:
70 """In-memory SQLite session (stub harmony does not actually query it)."""
71 engine = create_async_engine(
72 "sqlite+aiosqlite:///:memory:",
73 connect_args={"check_same_thread": False},
74 poolclass=StaticPool,
75 )
76 async with engine.begin() as conn:
77 await conn.run_sync(Base.metadata.create_all)
78 factory = async_sessionmaker(bind=engine, expire_on_commit=False)
79 async with factory() as session:
80 yield session
81 async with engine.begin() as conn:
82 await conn.run_sync(Base.metadata.drop_all)
83 await engine.dispose()
84
85
86 # ---------------------------------------------------------------------------
87 # Unit — constants and helpers
88 # ---------------------------------------------------------------------------
89
90
91 def test_known_modes_set_matches_tuple() -> None:
92 """KNOWN_MODES_SET and KNOWN_MODES tuple are in sync."""
93 assert set(KNOWN_MODES) == KNOWN_MODES_SET
94
95
96 def test_known_modes_contains_standard_modes() -> None:
97 """All seven standard modes plus major/minor are present."""
98 for mode in ("major", "minor", "dorian", "mixolydian", "lydian"):
99 assert mode in KNOWN_MODES_SET
100
101
102 def test_tension_label_rising() -> None:
103 """Monotonically rising profile is labeled as 'Rising'."""
104 label = _tension_label([0.1, 0.3, 0.6, 0.9])
105 assert "Rising" in label
106
107
108 def test_tension_label_falling() -> None:
109 """Monotonically falling profile is labeled as 'Falling'."""
110 label = _tension_label([0.9, 0.6, 0.3, 0.1])
111 assert "Falling" in label
112
113
114 def test_tension_label_arch() -> None:
115 """Low-high-low arc produces a 'tension-release' label."""
116 label = _tension_label([0.2, 0.4, 0.8, 0.3])
117 assert "Resolution" in label or "release" in label.lower()
118
119
120 def test_tension_label_single_value_low() -> None:
121 label = _tension_label([0.1])
122 assert label == "Low"
123
124
125 def test_tension_label_single_value_high() -> None:
126 label = _tension_label([0.8])
127 assert label == "High"
128
129
130 def test_tension_label_empty() -> None:
131 assert _tension_label([]) == "unknown"
132
133
134 # ---------------------------------------------------------------------------
135 # Unit — stub data
136 # ---------------------------------------------------------------------------
137
138
139 def test_stub_harmony_returns_harmony_result() -> None:
140 """_stub_harmony returns a HarmonyResult with the expected fields."""
141 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
142 assert result["commit_id"] == "a1b2c3d4"
143 assert result["branch"] == "main"
144 assert result["key"] is not None
145 assert result["mode"] in KNOWN_MODES_SET
146 assert 0.0 <= result["confidence"] <= 1.0
147 assert isinstance(result["chord_progression"], list)
148 assert len(result["chord_progression"]) > 0
149 assert result["harmonic_rhythm_avg"] > 0
150 assert isinstance(result["tension_profile"], list)
151 assert result["track"] == "all"
152 assert result["source"] == "stub"
153
154
155 def test_stub_harmony_track_scope() -> None:
156 """_stub_harmony respects the track argument."""
157 result = _stub_harmony(commit_id="a1b2c3d4", branch="main", track="keys")
158 assert result["track"] == "keys"
159
160
161 def test_stub_harmony_chord_progression_are_strings() -> None:
162 """Every chord in the stub progression is a non-empty string."""
163 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
164 for chord in result["chord_progression"]:
165 assert isinstance(chord, str)
166 assert len(chord) > 0
167
168
169 # ---------------------------------------------------------------------------
170 # Unit — renderers
171 # ---------------------------------------------------------------------------
172
173
174 def test_render_result_human_full(capsys: pytest.CaptureFixture[str]) -> None:
175 """_render_result_human with no flags shows key, mode, chords, and tension."""
176 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
177 _render_result_human(result, False, False, False, False)
178 out = capsys.readouterr().out
179 assert "Harmonic Analysis" in out
180 assert "Key:" in out
181 assert "Mode:" in out
182 assert "Chord progression:" in out
183 assert "Tension profile:" in out
184
185
186 def test_render_result_human_key_only(capsys: pytest.CaptureFixture[str]) -> None:
187 """--key shows only the key line."""
188 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
189 _render_result_human(result, False, True, False, False)
190 out = capsys.readouterr().out
191 assert "Key:" in out
192 assert "Mode:" not in out
193 assert "Chord progression:" not in out
194 assert "Tension profile:" not in out
195
196
197 def test_render_result_human_mode_only(capsys: pytest.CaptureFixture[str]) -> None:
198 """--mode shows only the mode line."""
199 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
200 _render_result_human(result, False, False, True, False)
201 out = capsys.readouterr().out
202 assert "Mode:" in out
203 assert "Key:" not in out
204
205
206 def test_render_result_human_progression_only(capsys: pytest.CaptureFixture[str]) -> None:
207 """--progression shows only the chord progression line."""
208 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
209 _render_result_human(result, True, False, False, False)
210 out = capsys.readouterr().out
211 assert "Chord progression:" in out
212 assert "Key:" not in out
213 assert "Tension profile:" not in out
214
215
216 def test_render_result_human_tension_only(capsys: pytest.CaptureFixture[str]) -> None:
217 """--tension shows only the tension profile line."""
218 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
219 _render_result_human(result, False, False, False, True)
220 out = capsys.readouterr().out
221 assert "Tension profile:" in out
222 assert "Key:" not in out
223 assert "Chord progression:" not in out
224
225
226 def test_render_result_json_full(capsys: pytest.CaptureFixture[str]) -> None:
227 """_render_result_json with no flags emits all HarmonyResult fields."""
228 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
229 _render_result_json(result, False, False, False, False)
230 raw = capsys.readouterr().out
231 payload = json.loads(raw)
232 for field in ("commit_id", "branch", "key", "mode", "confidence",
233 "chord_progression", "harmonic_rhythm_avg", "tension_profile"):
234 assert field in payload, f"Missing field: {field}"
235
236
237 def test_render_result_json_key_only(capsys: pytest.CaptureFixture[str]) -> None:
238 """--key JSON includes only key and confidence."""
239 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
240 _render_result_json(result, False, True, False, False)
241 raw = capsys.readouterr().out
242 payload = json.loads(raw)
243 assert "key" in payload
244 assert "confidence" in payload
245 assert "mode" not in payload
246 assert "chord_progression" not in payload
247
248
249 def test_render_result_json_progression_only(capsys: pytest.CaptureFixture[str]) -> None:
250 """--progression JSON includes only chord_progression."""
251 result = _stub_harmony(commit_id="a1b2c3d4", branch="main")
252 _render_result_json(result, True, False, False, False)
253 raw = capsys.readouterr().out
254 payload = json.loads(raw)
255 assert "chord_progression" in payload
256 assert "key" not in payload
257
258
259 def test_render_compare_human(capsys: pytest.CaptureFixture[str]) -> None:
260 """_render_compare_human shows both commits and change flags."""
261 head = _stub_harmony(commit_id="a1b2c3d4", branch="main")
262 ref = _stub_harmony(commit_id="deadbeef", branch="main")
263 cmp: HarmonyCompareResult = HarmonyCompareResult(
264 head=head,
265 compare=ref,
266 key_changed=False,
267 mode_changed=False,
268 chord_progression_delta=[],
269 )
270 _render_compare_human(cmp)
271 out = capsys.readouterr().out
272 assert "a1b2c3d4" in out
273 assert "deadbeef" in out
274 assert "Key changed" in out
275
276
277 def test_render_compare_json(capsys: pytest.CaptureFixture[str]) -> None:
278 """_render_compare_json emits valid JSON with head/compare/delta keys."""
279 head = _stub_harmony(commit_id="a1b2c3d4", branch="main")
280 ref = _stub_harmony(commit_id="deadbeef", branch="main")
281 cmp: HarmonyCompareResult = HarmonyCompareResult(
282 head=head,
283 compare=ref,
284 key_changed=False,
285 mode_changed=False,
286 chord_progression_delta=[],
287 )
288 _render_compare_json(cmp)
289 raw = capsys.readouterr().out
290 payload = json.loads(raw)
291 assert "head" in payload
292 assert "compare" in payload
293 assert "key_changed" in payload
294 assert "chord_progression_delta" in payload
295
296
297 # ---------------------------------------------------------------------------
298 # Async core — _harmony_analyze_async
299 # ---------------------------------------------------------------------------
300
301
302 @pytest.mark.anyio
303 async def test_harmony_async_default_output(
304 tmp_path: pathlib.Path,
305 db_session: AsyncSession,
306 capsys: pytest.CaptureFixture[str],
307 ) -> None:
308 """_harmony_analyze_async with no flags shows full harmonic summary."""
309 _init_muse_repo(tmp_path)
310 _commit_ref(tmp_path)
311
312 result = await _harmony_analyze_async(
313 root=tmp_path,
314 session=db_session,
315 commit=None,
316 track=None,
317 section=None,
318 compare=None,
319 commit_range=None,
320 show_progression=False,
321 show_key=False,
322 show_mode=False,
323 show_tension=False,
324 as_json=False,
325 )
326
327 out = capsys.readouterr().out
328 assert "Harmonic Analysis" in out
329 assert "Key:" in out
330 assert "Mode:" in out
331 assert "Chord progression:" in out
332 assert "Tension profile:" in out
333 assert result["source"] == "stub"
334
335
336 @pytest.mark.anyio
337 async def test_harmony_async_json_mode(
338 tmp_path: pathlib.Path,
339 db_session: AsyncSession,
340 capsys: pytest.CaptureFixture[str],
341 ) -> None:
342 """_harmony_analyze_async --json emits valid JSON with all HarmonyResult fields."""
343 _init_muse_repo(tmp_path)
344 _commit_ref(tmp_path)
345
346 await _harmony_analyze_async(
347 root=tmp_path,
348 session=db_session,
349 commit=None,
350 track=None,
351 section=None,
352 compare=None,
353 commit_range=None,
354 show_progression=False,
355 show_key=False,
356 show_mode=False,
357 show_tension=False,
358 as_json=True,
359 )
360
361 raw = capsys.readouterr().out
362 payload = json.loads(raw)
363 for field in ("commit_id", "branch", "key", "mode", "confidence",
364 "chord_progression", "harmonic_rhythm_avg", "tension_profile"):
365 assert field in payload, f"Missing field: {field}"
366
367
368 @pytest.mark.anyio
369 async def test_harmony_async_no_commits_exits_success(
370 tmp_path: pathlib.Path,
371 db_session: AsyncSession,
372 capsys: pytest.CaptureFixture[str],
373 ) -> None:
374 """With no commits and no explicit commit arg, exits 0 with informative message."""
375 _init_muse_repo(tmp_path)
376 # No _commit_ref call — branch ref is empty.
377
378 import typer
379
380 with pytest.raises(typer.Exit) as exc_info:
381 await _harmony_analyze_async(
382 root=tmp_path,
383 session=db_session,
384 commit=None,
385 track=None,
386 section=None,
387 compare=None,
388 commit_range=None,
389 show_progression=False,
390 show_key=False,
391 show_mode=False,
392 show_tension=False,
393 as_json=False,
394 )
395 assert exc_info.value.exit_code == int(ExitCode.SUCCESS)
396 out = capsys.readouterr().out
397 assert "No commits yet" in out
398
399
400 @pytest.mark.anyio
401 async def test_harmony_async_explicit_commit_ref(
402 tmp_path: pathlib.Path,
403 db_session: AsyncSession,
404 capsys: pytest.CaptureFixture[str],
405 ) -> None:
406 """An explicit commit ref appears in the output."""
407 _init_muse_repo(tmp_path)
408 _commit_ref(tmp_path)
409
410 result = await _harmony_analyze_async(
411 root=tmp_path,
412 session=db_session,
413 commit="deadbeef",
414 track=None,
415 section=None,
416 compare=None,
417 commit_range=None,
418 show_progression=False,
419 show_key=False,
420 show_mode=False,
421 show_tension=False,
422 as_json=False,
423 )
424
425 out = capsys.readouterr().out
426 assert "deadbeef" in out
427 assert result["commit_id"] == "deadbeef"
428
429
430 @pytest.mark.anyio
431 async def test_harmony_async_track_scoped(
432 tmp_path: pathlib.Path,
433 db_session: AsyncSession,
434 capsys: pytest.CaptureFixture[str],
435 ) -> None:
436 """--track is reflected in the result track field."""
437 _init_muse_repo(tmp_path)
438 _commit_ref(tmp_path)
439
440 result = await _harmony_analyze_async(
441 root=tmp_path,
442 session=db_session,
443 commit=None,
444 track="keys",
445 section=None,
446 compare=None,
447 commit_range=None,
448 show_progression=False,
449 show_key=False,
450 show_mode=False,
451 show_tension=False,
452 as_json=False,
453 )
454
455 assert result["track"] == "keys"
456
457
458 @pytest.mark.anyio
459 async def test_harmony_async_progression_flag(
460 tmp_path: pathlib.Path,
461 db_session: AsyncSession,
462 capsys: pytest.CaptureFixture[str],
463 ) -> None:
464 """--progression shows chord progression and suppresses other fields."""
465 _init_muse_repo(tmp_path)
466 _commit_ref(tmp_path)
467
468 await _harmony_analyze_async(
469 root=tmp_path,
470 session=db_session,
471 commit=None,
472 track=None,
473 section=None,
474 compare=None,
475 commit_range=None,
476 show_progression=True,
477 show_key=False,
478 show_mode=False,
479 show_tension=False,
480 as_json=False,
481 )
482
483 out = capsys.readouterr().out
484 assert "Chord progression:" in out
485 assert "Key:" not in out
486 assert "Tension profile:" not in out
487
488
489 @pytest.mark.anyio
490 async def test_harmony_async_key_flag(
491 tmp_path: pathlib.Path,
492 db_session: AsyncSession,
493 capsys: pytest.CaptureFixture[str],
494 ) -> None:
495 """--key shows key center and suppresses other fields."""
496 _init_muse_repo(tmp_path)
497 _commit_ref(tmp_path)
498
499 await _harmony_analyze_async(
500 root=tmp_path,
501 session=db_session,
502 commit=None,
503 track=None,
504 section=None,
505 compare=None,
506 commit_range=None,
507 show_progression=False,
508 show_key=True,
509 show_mode=False,
510 show_tension=False,
511 as_json=False,
512 )
513
514 out = capsys.readouterr().out
515 assert "Key:" in out
516 assert "Mode:" not in out
517
518
519 @pytest.mark.anyio
520 async def test_harmony_async_mode_flag(
521 tmp_path: pathlib.Path,
522 db_session: AsyncSession,
523 capsys: pytest.CaptureFixture[str],
524 ) -> None:
525 """--mode shows mode and suppresses other fields."""
526 _init_muse_repo(tmp_path)
527 _commit_ref(tmp_path)
528
529 await _harmony_analyze_async(
530 root=tmp_path,
531 session=db_session,
532 commit=None,
533 track=None,
534 section=None,
535 compare=None,
536 commit_range=None,
537 show_progression=False,
538 show_key=False,
539 show_mode=True,
540 show_tension=False,
541 as_json=False,
542 )
543
544 out = capsys.readouterr().out
545 assert "Mode:" in out
546 assert "Key:" not in out
547
548
549 @pytest.mark.anyio
550 async def test_harmony_async_tension_flag(
551 tmp_path: pathlib.Path,
552 db_session: AsyncSession,
553 capsys: pytest.CaptureFixture[str],
554 ) -> None:
555 """--tension shows tension profile and suppresses other fields."""
556 _init_muse_repo(tmp_path)
557 _commit_ref(tmp_path)
558
559 await _harmony_analyze_async(
560 root=tmp_path,
561 session=db_session,
562 commit=None,
563 track=None,
564 section=None,
565 compare=None,
566 commit_range=None,
567 show_progression=False,
568 show_key=False,
569 show_mode=False,
570 show_tension=True,
571 as_json=False,
572 )
573
574 out = capsys.readouterr().out
575 assert "Tension profile:" in out
576 assert "Key:" not in out
577
578
579 @pytest.mark.anyio
580 async def test_harmony_async_compare_mode(
581 tmp_path: pathlib.Path,
582 db_session: AsyncSession,
583 capsys: pytest.CaptureFixture[str],
584 ) -> None:
585 """--compare renders a comparison between HEAD and the reference commit."""
586 _init_muse_repo(tmp_path)
587 _commit_ref(tmp_path)
588
589 await _harmony_analyze_async(
590 root=tmp_path,
591 session=db_session,
592 commit=None,
593 track=None,
594 section=None,
595 compare="deadbeef",
596 commit_range=None,
597 show_progression=False,
598 show_key=False,
599 show_mode=False,
600 show_tension=False,
601 as_json=False,
602 )
603
604 out = capsys.readouterr().out
605 assert "deadbeef" in out
606 assert "Harmonic Comparison" in out
607
608
609 @pytest.mark.anyio
610 async def test_harmony_async_compare_json(
611 tmp_path: pathlib.Path,
612 db_session: AsyncSession,
613 capsys: pytest.CaptureFixture[str],
614 ) -> None:
615 """--compare --json emits a HarmonyCompareResult as JSON."""
616 _init_muse_repo(tmp_path)
617 _commit_ref(tmp_path)
618
619 await _harmony_analyze_async(
620 root=tmp_path,
621 session=db_session,
622 commit=None,
623 track=None,
624 section=None,
625 compare="deadbeef",
626 commit_range=None,
627 show_progression=False,
628 show_key=False,
629 show_mode=False,
630 show_tension=False,
631 as_json=True,
632 )
633
634 raw = capsys.readouterr().out
635 payload = json.loads(raw)
636 assert "head" in payload
637 assert "compare" in payload
638 assert "key_changed" in payload
639 assert "mode_changed" in payload
640 assert "chord_progression_delta" in payload
641
642
643 @pytest.mark.anyio
644 async def test_harmony_async_range_flag_warns(
645 tmp_path: pathlib.Path,
646 db_session: AsyncSession,
647 capsys: pytest.CaptureFixture[str],
648 ) -> None:
649 """--range emits a stub boundary warning but still renders HEAD result."""
650 _init_muse_repo(tmp_path)
651 _commit_ref(tmp_path)
652
653 await _harmony_analyze_async(
654 root=tmp_path,
655 session=db_session,
656 commit=None,
657 track=None,
658 section=None,
659 compare=None,
660 commit_range="HEAD~10..HEAD",
661 show_progression=False,
662 show_key=False,
663 show_mode=False,
664 show_tension=False,
665 as_json=False,
666 )
667
668 out = capsys.readouterr().out
669 assert "--range" in out
670 assert "Harmonic Analysis" in out
671
672
673 @pytest.mark.anyio
674 async def test_harmony_async_section_flag_warns(
675 tmp_path: pathlib.Path,
676 db_session: AsyncSession,
677 capsys: pytest.CaptureFixture[str],
678 ) -> None:
679 """--section emits a stub boundary warning but still renders HEAD result."""
680 _init_muse_repo(tmp_path)
681 _commit_ref(tmp_path)
682
683 await _harmony_analyze_async(
684 root=tmp_path,
685 session=db_session,
686 commit=None,
687 track=None,
688 section="verse",
689 compare=None,
690 commit_range=None,
691 show_progression=False,
692 show_key=False,
693 show_mode=False,
694 show_tension=False,
695 as_json=False,
696 )
697
698 out = capsys.readouterr().out
699 assert "--section" in out
700 assert "Harmonic Analysis" in out
701
702
703 # ---------------------------------------------------------------------------
704 # CLI integration — CliRunner
705 # ---------------------------------------------------------------------------
706
707
708 def test_cli_harmony_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
709 """``muse harmony`` exits 2 when invoked outside a Muse repository."""
710 prev = os.getcwd()
711 try:
712 os.chdir(tmp_path)
713 result = runner.invoke(cli, ["harmony"], catch_exceptions=False)
714 finally:
715 os.chdir(prev)
716
717 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND)
718 assert "not a muse repository" in result.output.lower()
719
720
721 def test_cli_harmony_help_lists_flags() -> None:
722 """``muse harmony --help`` shows all documented flags."""
723 result = runner.invoke(cli, ["harmony", "--help"])
724 assert result.exit_code == 0
725 for flag in ("--track", "--section", "--compare", "--range", "--progression",
726 "--key", "--mode", "--tension", "--json"):
727 assert flag in result.output, f"Flag '{flag}' not found in help output"
728
729
730 def test_cli_harmony_appears_in_muse_help() -> None:
731 """``muse --help`` lists the harmony subcommand."""
732 result = runner.invoke(cli, ["--help"])
733 assert result.exit_code == 0
734 assert "harmony" in result.output