cgcardona / muse public
test_arrange.py python
528 lines 18.3 KB
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse arrange`` — arrangement map display.
2
3 Tests cover:
4 - ``test_arrange_renders_matrix_for_commit`` — basic matrix output for a commit
5 - ``test_arrange_compare_shows_diff`` — diff between two commits
6 - ``test_arrange_density_mode`` — density (byte-size) mode
7 - Additional: JSON format, CSV format, section/track filtering, empty snapshot
8 """
9 from __future__ import annotations
10
11 import json
12 import pathlib
13 import uuid
14
15 import pytest
16 import pytest_asyncio
17 from sqlalchemy.ext.asyncio import AsyncSession
18
19 from maestro.muse_cli.commands.arrange import _arrange_async, _load_matrix
20 from maestro.muse_cli.commands.commit import _commit_async
21 from maestro.muse_cli.errors import ExitCode
22 from maestro.services.muse_arrange import (
23 ArrangementCell,
24 ArrangementMatrix,
25 build_arrangement_diff,
26 build_arrangement_matrix,
27 extract_section_instrument,
28 render_diff_json,
29 render_diff_text,
30 render_matrix_csv,
31 render_matrix_json,
32 render_matrix_text,
33 )
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
42 """Create a minimal .muse/ layout."""
43 rid = repo_id or str(uuid.uuid4())
44 muse = root / ".muse"
45 (muse / "refs" / "heads").mkdir(parents=True)
46 (muse / "repo.json").write_text(
47 json.dumps({"repo_id": rid, "schema_version": "1"})
48 )
49 (muse / "HEAD").write_text("refs/heads/main")
50 (muse / "refs" / "heads" / "main").write_text("")
51 return rid
52
53
54 def _populate_arrangement(root: pathlib.Path, layout: dict[str, bytes]) -> None:
55 """Populate muse-work/ with a section/instrument file layout.
56
57 *layout* maps relative paths (e.g. ``"intro/drums/beat.mid"``) to bytes.
58 """
59 workdir = root / "muse-work"
60 for rel_path, content in layout.items():
61 abs_path = workdir / rel_path
62 abs_path.parent.mkdir(parents=True, exist_ok=True)
63 abs_path.write_bytes(content)
64
65
66 _BASIC_LAYOUT: dict[str, bytes] = {
67 "intro/drums/beat.mid": b"MIDI" * 100,
68 "intro/bass/line.mid": b"MIDI" * 50,
69 "verse/drums/beat.mid": b"MIDI" * 120,
70 "verse/bass/line.mid": b"MIDI" * 60,
71 "verse/strings/pad.mid": b"MIDI" * 80,
72 "chorus/drums/beat.mid": b"MIDI" * 150,
73 "chorus/bass/line.mid": b"MIDI" * 70,
74 "chorus/strings/pad.mid": b"MIDI" * 90,
75 "chorus/piano/chords.mid": b"MIDI" * 110,
76 }
77
78
79 # ---------------------------------------------------------------------------
80 # Unit tests — pure service functions (no DB required)
81 # ---------------------------------------------------------------------------
82
83
84 class TestExtractSectionInstrument:
85 """Tests for the path-parsing function."""
86
87 def test_three_component_path(self) -> None:
88 assert extract_section_instrument("intro/drums/beat.mid") == ("intro", "drums")
89
90 def test_deep_path_uses_first_two_components(self) -> None:
91 assert extract_section_instrument("chorus/strings/sub/pad.mid") == (
92 "chorus",
93 "strings",
94 )
95
96 def test_two_component_path_returns_none(self) -> None:
97 assert extract_section_instrument("drums/beat.mid") is None
98
99 def test_flat_file_returns_none(self) -> None:
100 assert extract_section_instrument("beat.mid") is None
101
102 def test_section_is_normalised_lowercase(self) -> None:
103 result = extract_section_instrument("CHORUS/Piano/chords.mid")
104 assert result == ("chorus", "piano")
105
106 def test_prechorus_alias(self) -> None:
107 result = extract_section_instrument("pre-chorus/violin/part.mid")
108 assert result is not None
109 assert result[0] == "prechorus"
110
111
112 class TestBuildArrangementMatrix:
113 """Tests for the matrix builder."""
114
115 def test_basic_matrix_has_correct_sections_and_instruments(self) -> None:
116 manifest = {
117 "intro/drums/beat.mid": "oid1",
118 "verse/drums/beat.mid": "oid2",
119 "verse/bass/line.mid": "oid3",
120 "chorus/strings/pad.mid": "oid4",
121 }
122 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
123
124 assert set(matrix.sections) == {"intro", "verse", "chorus"}
125 assert set(matrix.instruments) == {"drums", "bass", "strings"}
126
127 def test_active_cells_are_correct(self) -> None:
128 manifest = {
129 "intro/drums/beat.mid": "oid1",
130 "chorus/strings/pad.mid": "oid2",
131 }
132 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
133
134 assert matrix.get_cell("intro", "drums").active is True
135 assert matrix.get_cell("chorus", "strings").active is True
136 assert matrix.get_cell("intro", "strings").active is False
137 assert matrix.get_cell("chorus", "drums").active is False
138
139 def test_file_count_accumulated_per_cell(self) -> None:
140 manifest = {
141 "verse/drums/take1.mid": "oid1",
142 "verse/drums/take2.mid": "oid2",
143 }
144 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
145 assert matrix.get_cell("verse", "drums").file_count == 2
146
147 def test_density_accumulates_bytes(self) -> None:
148 manifest = {
149 "chorus/bass/line.mid": "oid1",
150 "chorus/bass/alt.mid": "oid2",
151 }
152 sizes = {"oid1": 1000, "oid2": 2000}
153 matrix = build_arrangement_matrix("abcd1234" * 8, manifest, object_sizes=sizes)
154 assert matrix.get_cell("chorus", "bass").total_bytes == 3000
155
156 def test_files_without_section_structure_are_ignored(self) -> None:
157 manifest = {
158 "drums/beat.mid": "oid1", # 1 dir + filename = 2 parts, ignored (< 3)
159 "solo.mid": "oid2", # flat file = ignored
160 "intro/drums/beat.mid": "oid3", # valid: section/instrument/filename
161 }
162 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
163 assert matrix.instruments == ["drums"]
164 assert matrix.sections == ["intro"]
165
166 def test_section_ordering_follows_canonical_order(self) -> None:
167 manifest = {
168 "outro/drums/beat.mid": "oid1",
169 "intro/drums/beat.mid": "oid2",
170 "chorus/drums/beat.mid": "oid3",
171 "verse/drums/beat.mid": "oid4",
172 }
173 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
174 assert matrix.sections == ["intro", "verse", "chorus", "outro"]
175
176
177 class TestRenderMatrixText:
178 """Tests for text rendering."""
179
180 def test_renders_header_and_rows(self) -> None:
181 manifest = {
182 "intro/drums/beat.mid": "oid1",
183 "verse/bass/line.mid": "oid2",
184 }
185 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
186 output = render_matrix_text(matrix)
187
188 assert "Arrangement Map" in output
189 assert "abcd1234" in output
190 assert "drums" in output
191 assert "bass" in output
192 assert "Intro" in output
193 assert "Verse" in output
194
195 def test_active_cell_shows_block_char(self) -> None:
196 manifest = {"chorus/piano/chords.mid": "oid1"}
197 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
198 output = render_matrix_text(matrix)
199 assert "\u2588\u2588\u2588\u2588" in output # ████
200
201 def test_inactive_cell_shows_light_shade(self) -> None:
202 manifest = {
203 "intro/drums/beat.mid": "oid1",
204 "verse/bass/line.mid": "oid2",
205 }
206 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
207 output = render_matrix_text(matrix)
208 assert "\u2591\u2591\u2591\u2591" in output # ░░░░
209
210 def test_section_filter(self) -> None:
211 manifest = {
212 "intro/drums/beat.mid": "oid1",
213 "verse/drums/beat.mid": "oid2",
214 }
215 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
216 output = render_matrix_text(matrix, section_filter="intro")
217 assert "Intro" in output
218 assert "Verse" not in output
219
220 def test_track_filter(self) -> None:
221 manifest = {
222 "verse/drums/beat.mid": "oid1",
223 "verse/bass/line.mid": "oid2",
224 }
225 matrix = build_arrangement_matrix("abcd1234" * 8, manifest)
226 output = render_matrix_text(matrix, track_filter="drums")
227 assert "drums" in output
228 assert "bass" not in output
229
230 def test_density_mode_shows_byte_values(self) -> None:
231 manifest = {"chorus/strings/pad.mid": "oid1"}
232 sizes = {"oid1": 4096}
233 matrix = build_arrangement_matrix("abcd1234" * 8, manifest, object_sizes=sizes)
234 output = render_matrix_text(matrix, density=True)
235 assert "4,096" in output
236
237
238 class TestRenderMatrixJson:
239 """Tests for JSON rendering."""
240
241 def test_json_has_correct_structure(self) -> None:
242 manifest = {
243 "intro/drums/beat.mid": "oid1",
244 "chorus/bass/line.mid": "oid2",
245 }
246 matrix = build_arrangement_matrix("a" * 64, manifest)
247 raw = render_matrix_json(matrix)
248 data = json.loads(raw)
249
250 assert "commit_id" in data
251 assert "sections" in data
252 assert "instruments" in data
253 assert "arrangement" in data
254
255 def test_json_active_values_are_bool(self) -> None:
256 manifest = {"verse/piano/chords.mid": "oid1"}
257 matrix = build_arrangement_matrix("a" * 64, manifest)
258 raw = render_matrix_json(matrix)
259 data = json.loads(raw)
260 assert data["arrangement"]["piano"]["verse"] is True
261
262 def test_json_density_mode_includes_bytes(self) -> None:
263 manifest = {"verse/piano/chords.mid": "oid1"}
264 sizes = {"oid1": 999}
265 matrix = build_arrangement_matrix("a" * 64, manifest, object_sizes=sizes)
266 raw = render_matrix_json(matrix, density=True)
267 data = json.loads(raw)
268 cell = data["arrangement"]["piano"]["verse"]
269 assert isinstance(cell, dict)
270 assert cell["total_bytes"] == 999
271
272
273 class TestRenderMatrixCsv:
274 """Tests for CSV rendering."""
275
276 def test_csv_has_header_row(self) -> None:
277 manifest = {"intro/drums/beat.mid": "oid1"}
278 matrix = build_arrangement_matrix("a" * 64, manifest)
279 output = render_matrix_csv(matrix)
280 lines = output.strip().splitlines()
281 assert lines[0].startswith("instrument")
282
283 def test_csv_active_is_1(self) -> None:
284 manifest = {"intro/drums/beat.mid": "oid1"}
285 matrix = build_arrangement_matrix("a" * 64, manifest)
286 output = render_matrix_csv(matrix)
287 lines = output.strip().splitlines()
288 assert "1" in lines[1]
289
290
291 class TestBuildArrangementDiff:
292 """Tests for the diff builder."""
293
294 def _make_matrix(
295 self, commit_id: str, manifest: dict[str, str]
296 ) -> ArrangementMatrix:
297 return build_arrangement_matrix(commit_id, manifest)
298
299 def test_added_cell_detected(self) -> None:
300 manifest_a = {"intro/drums/beat.mid": "oid1"}
301 manifest_b = {
302 "intro/drums/beat.mid": "oid1",
303 "intro/strings/pad.mid": "oid2",
304 }
305 mx_a = self._make_matrix("a" * 64, manifest_a)
306 mx_b = self._make_matrix("b" * 64, manifest_b)
307
308 diff = build_arrangement_diff(mx_a, mx_b)
309 assert diff.cells[("intro", "strings")].status == "added"
310
311 def test_removed_cell_detected(self) -> None:
312 manifest_a = {
313 "chorus/drums/beat.mid": "oid1",
314 "chorus/piano/chords.mid": "oid2",
315 }
316 manifest_b = {"chorus/drums/beat.mid": "oid1"}
317 mx_a = self._make_matrix("a" * 64, manifest_a)
318 mx_b = self._make_matrix("b" * 64, manifest_b)
319
320 diff = build_arrangement_diff(mx_a, mx_b)
321 assert diff.cells[("chorus", "piano")].status == "removed"
322
323 def test_unchanged_cell_detected(self) -> None:
324 manifest = {"verse/bass/line.mid": "oid1"}
325 mx_a = self._make_matrix("a" * 64, manifest)
326 mx_b = self._make_matrix("b" * 64, manifest)
327
328 diff = build_arrangement_diff(mx_a, mx_b)
329 assert diff.cells[("verse", "bass")].status == "unchanged"
330
331
332 class TestRenderDiff:
333 """Tests for diff renderers."""
334
335 def _make_matrix(
336 self, commit_id: str, manifest: dict[str, str]
337 ) -> ArrangementMatrix:
338 return build_arrangement_matrix(commit_id, manifest)
339
340 def test_diff_text_includes_commit_ids(self) -> None:
341 mx_a = self._make_matrix("a" * 64, {"intro/drums/beat.mid": "oid1"})
342 mx_b = self._make_matrix("b" * 64, {"intro/drums/beat.mid": "oid1"})
343 diff = build_arrangement_diff(mx_a, mx_b)
344 output = render_diff_text(diff)
345 assert "aaaaaaaa" in output
346 assert "bbbbbbbb" in output
347
348 def test_diff_json_lists_changes(self) -> None:
349 manifest_a = {"intro/drums/beat.mid": "oid1"}
350 manifest_b = {
351 "intro/drums/beat.mid": "oid1",
352 "intro/bass/line.mid": "oid2",
353 }
354 mx_a = self._make_matrix("a" * 64, manifest_a)
355 mx_b = self._make_matrix("b" * 64, manifest_b)
356 diff = build_arrangement_diff(mx_a, mx_b)
357 raw = render_diff_json(diff)
358 data = json.loads(raw)
359
360 changes = data["changes"]
361 assert any(
362 c["section"] == "intro" and c["instrument"] == "bass" and c["status"] == "added"
363 for c in changes
364 )
365
366
367 # ---------------------------------------------------------------------------
368 # Integration tests — DB required
369 # ---------------------------------------------------------------------------
370
371
372 @pytest.mark.anyio
373 async def test_arrange_renders_matrix_for_commit(
374 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
375 ) -> None:
376 """muse arrange HEAD renders the arrangement matrix for the current HEAD commit."""
377 _init_muse_repo(tmp_path)
378 _populate_arrangement(tmp_path, _BASIC_LAYOUT)
379
380 commit_id = await _commit_async(
381 message="initial arrangement",
382 root=tmp_path,
383 session=muse_cli_db_session,
384 )
385
386 muse_dir = tmp_path / ".muse"
387 matrix = await _load_matrix(muse_cli_db_session, muse_dir, "HEAD", density=False)
388
389 assert matrix.commit_id == commit_id
390 assert set(matrix.instruments) >= {"drums", "bass", "strings", "piano"}
391 assert "intro" in matrix.sections
392 assert "verse" in matrix.sections
393 assert "chorus" in matrix.sections
394
395 assert matrix.get_cell("intro", "drums").active is True
396 assert matrix.get_cell("intro", "strings").active is False # strings only in verse/chorus
397
398
399 @pytest.mark.anyio
400 async def test_arrange_compare_shows_diff(
401 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
402 ) -> None:
403 """muse arrange --compare shows added and removed cells between two commits."""
404 _init_muse_repo(tmp_path)
405 _populate_arrangement(tmp_path, {
406 "intro/drums/beat.mid": b"MIDI" * 50,
407 "verse/drums/beat.mid": b"MIDI" * 50,
408 })
409
410 commit_a = await _commit_async(
411 message="commit A",
412 root=tmp_path,
413 session=muse_cli_db_session,
414 )
415
416 # Add strings in verse for the second commit
417 (tmp_path / "muse-work" / "verse" / "strings").mkdir(parents=True, exist_ok=True)
418 (tmp_path / "muse-work" / "verse" / "strings" / "pad.mid").write_bytes(b"MIDI" * 80)
419
420 commit_b = await _commit_async(
421 message="commit B",
422 root=tmp_path,
423 session=muse_cli_db_session,
424 )
425
426 muse_dir = tmp_path / ".muse"
427 matrix_a = await _load_matrix(muse_cli_db_session, muse_dir, commit_a, density=False)
428 matrix_b = await _load_matrix(muse_cli_db_session, muse_dir, commit_b, density=False)
429
430 diff = build_arrangement_diff(matrix_a, matrix_b)
431
432 # Strings was added in verse
433 assert diff.cells[("verse", "strings")].status == "added"
434 # Drums unchanged in both sections
435 assert diff.cells[("intro", "drums")].status == "unchanged"
436 assert diff.cells[("verse", "drums")].status == "unchanged"
437
438 # Verify JSON serialisation
439 json_output = render_diff_json(diff)
440 data = json.loads(json_output)
441 changes = data["changes"]
442 added = [c for c in changes if c["status"] == "added"]
443 assert len(added) == 1
444 assert added[0]["instrument"] == "strings"
445 assert added[0]["section"] == "verse"
446
447
448 @pytest.mark.anyio
449 async def test_arrange_density_mode(
450 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
451 ) -> None:
452 """muse arrange --density shows byte totals per cell."""
453 _init_muse_repo(tmp_path)
454 content = b"X" * 4096
455 _populate_arrangement(tmp_path, {"chorus/strings/pad.mid": content})
456
457 commit_id = await _commit_async(
458 message="density test",
459 root=tmp_path,
460 session=muse_cli_db_session,
461 )
462
463 muse_dir = tmp_path / ".muse"
464 matrix = await _load_matrix(muse_cli_db_session, muse_dir, "HEAD", density=True)
465
466 cell = matrix.get_cell("chorus", "strings")
467 assert cell.active is True
468 assert cell.total_bytes == 4096
469
470 output = render_matrix_text(matrix, density=True)
471 assert "4,096" in output
472
473
474 @pytest.mark.anyio
475 async def test_arrange_empty_snapshot_returns_no_data_message(
476 tmp_path: pathlib.Path,
477 muse_cli_db_session: AsyncSession,
478 capsys: pytest.CaptureFixture[str],
479 ) -> None:
480 """When committed files don't follow section/instrument convention, arrange reports no data."""
481 _init_muse_repo(tmp_path)
482 # Files WITHOUT the section/instrument path structure (flat or 2-component paths)
483 _populate_arrangement(tmp_path, {"beat.mid": b"MIDI", "drums/hit.mid": b"MIDI"})
484
485 await _commit_async(
486 message="flat layout",
487 root=tmp_path,
488 session=muse_cli_db_session,
489 )
490
491 muse_dir = tmp_path / ".muse"
492 matrix = await _load_matrix(muse_cli_db_session, muse_dir, "HEAD", density=False)
493
494 assert matrix.sections == []
495 assert matrix.instruments == []
496
497
498 @pytest.mark.anyio
499 async def test_arrange_json_format(
500 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
501 ) -> None:
502 """--format json outputs valid JSON with correct arrangement data."""
503 import io
504 import typer
505 from typer.testing import CliRunner
506 from maestro.muse_cli.app import cli
507
508 _init_muse_repo(tmp_path)
509 _populate_arrangement(tmp_path, {
510 "intro/drums/beat.mid": b"MIDI" * 30,
511 "verse/bass/line.mid": b"MIDI" * 40,
512 })
513
514 await _commit_async(
515 message="json format test",
516 root=tmp_path,
517 session=muse_cli_db_session,
518 )
519
520 muse_dir = tmp_path / ".muse"
521 matrix = await _load_matrix(muse_cli_db_session, muse_dir, "HEAD", density=False)
522
523 raw = render_matrix_json(matrix)
524 data = json.loads(raw)
525
526 assert data["arrangement"]["drums"]["intro"] is True
527 assert data["arrangement"]["bass"].get("intro", False) is False
528 assert data["arrangement"]["bass"]["verse"] is True