cgcardona / muse public
test_show.py python
573 lines 20.0 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse show``.
2
3 All async tests call ``_show_async`` / ``_diff_vs_parent_async`` directly
4 with an in-memory SQLite session and a ``tmp_path`` repo root — no real
5 Postgres or running process required.
6
7 Coverage:
8 - Regression: show displays commit metadata (test_muse_show_displays_commit_metadata)
9 - JSON output (test_muse_show_json_output)
10 - Diff vs parent (test_muse_show_diff_vs_parent)
11 - Diff on root commit (no parent) (test_muse_show_diff_root_commit)
12 - MIDI file listing (test_muse_show_midi_list)
13 - Audio preview stub path (test_muse_show_audio_preview)
14 - HEAD resolution (test_muse_show_head_resolution)
15 - Branch name resolution (test_muse_show_branch_name_resolution)
16 - Short prefix resolution (test_muse_show_prefix_resolution)
17 - Ambiguous prefix returns USER_ERROR
18 - Unknown ref returns USER_ERROR
19 - Commit with no parent shows no parent line
20 - _looks_like_hex_prefix unit tests
21 - _midi_files_in_manifest unit tests
22 """
23 from __future__ import annotations
24
25 import json
26 import os
27 import pathlib
28 import uuid
29
30 import pytest
31 import typer
32 from sqlalchemy.ext.asyncio import AsyncSession
33
34 from maestro.muse_cli.commands.commit import _commit_async
35 from maestro.muse_cli.commands.show import (
36 ShowCommitResult,
37 ShowDiffResult,
38 _diff_vs_parent_async,
39 _looks_like_hex_prefix,
40 _midi_files_in_manifest,
41 _show_async,
42 )
43 from maestro.muse_cli.errors import ExitCode
44
45
46 # ---------------------------------------------------------------------------
47 # Helpers
48 # ---------------------------------------------------------------------------
49
50
51 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
52 """Initialise a minimal .muse/ directory structure for testing."""
53 rid = repo_id or str(uuid.uuid4())
54 muse = root / ".muse"
55 (muse / "refs" / "heads").mkdir(parents=True)
56 (muse / "repo.json").write_text(
57 json.dumps({"repo_id": rid, "schema_version": "1"})
58 )
59 (muse / "HEAD").write_text("refs/heads/main")
60 (muse / "refs" / "heads" / "main").write_text("")
61 return rid
62
63
64 def _write_workdir(root: pathlib.Path, files: dict[str, bytes]) -> None:
65 """Overwrite the muse-work directory with exactly *files* (clears old content)."""
66 workdir = root / "muse-work"
67 # Remove old files so each commit has a clean, deterministic snapshot
68 if workdir.exists():
69 for child in list(workdir.iterdir()):
70 child.unlink()
71 else:
72 workdir.mkdir(parents=True)
73 for name, content in files.items():
74 (workdir / name).write_bytes(content)
75
76
77 async def _make_commit(
78 root: pathlib.Path,
79 session: AsyncSession,
80 message: str,
81 files: dict[str, bytes],
82 ) -> str:
83 """Create one commit with exactly *files* and return its commit_id."""
84 _write_workdir(root, files)
85 return await _commit_async(message=message, root=root, session=session)
86
87
88 # ---------------------------------------------------------------------------
89 # Unit tests — pure functions
90 # ---------------------------------------------------------------------------
91
92
93 class TestLooksLikeHexPrefix:
94 def test_full_64_char_sha(self) -> None:
95 sha = "a" * 64
96 assert _looks_like_hex_prefix(sha) is True
97
98 def test_short_8_char_prefix(self) -> None:
99 assert _looks_like_hex_prefix("abc12345") is True
100
101 def test_four_char_minimum(self) -> None:
102 assert _looks_like_hex_prefix("abcd") is True
103
104 def test_three_chars_too_short(self) -> None:
105 assert _looks_like_hex_prefix("abc") is False
106
107 def test_uppercase_is_valid_hex(self) -> None:
108 assert _looks_like_hex_prefix("ABCDEF12") is True
109
110 def test_branch_name_not_hex(self) -> None:
111 assert _looks_like_hex_prefix("main") is False
112
113 def test_head_not_hex(self) -> None:
114 assert _looks_like_hex_prefix("HEAD") is False
115
116 def test_hyphenated_branch_not_hex(self) -> None:
117 assert _looks_like_hex_prefix("feat/my-branch") is False
118
119
120 class TestMidiFilesInManifest:
121 def test_empty_manifest(self) -> None:
122 assert _midi_files_in_manifest({}) == []
123
124 def test_no_midi_files(self) -> None:
125 manifest = {"beat.txt": "aaa", "notes.xml": "bbb"}
126 assert _midi_files_in_manifest(manifest) == []
127
128 def test_dot_mid_extension(self) -> None:
129 manifest = {"beat.mid": "aaa"}
130 assert _midi_files_in_manifest(manifest) == ["beat.mid"]
131
132 def test_dot_midi_extension(self) -> None:
133 manifest = {"track.midi": "bbb"}
134 assert _midi_files_in_manifest(manifest) == ["track.midi"]
135
136 def test_dot_smf_extension(self) -> None:
137 manifest = {"song.smf": "ccc"}
138 assert _midi_files_in_manifest(manifest) == ["song.smf"]
139
140 def test_mixed_files_returns_only_midi(self) -> None:
141 manifest = {
142 "beat.mid": "aaa",
143 "notes.txt": "bbb",
144 "keys.mid": "ccc",
145 "cover.png": "ddd",
146 }
147 result = _midi_files_in_manifest(manifest)
148 assert result == ["beat.mid", "keys.mid"]
149
150 def test_sorted_output(self) -> None:
151 manifest = {"z.mid": "1", "a.mid": "2", "m.mid": "3"}
152 result = _midi_files_in_manifest(manifest)
153 assert result == ["a.mid", "m.mid", "z.mid"]
154
155 def test_uppercase_extension(self) -> None:
156 manifest = {"TRACK.MID": "aaa"}
157 assert _midi_files_in_manifest(manifest) == ["TRACK.MID"]
158
159 def test_nested_path_with_midi_extension(self) -> None:
160 manifest = {"sections/verse/piano.mid": "hash1"}
161 assert _midi_files_in_manifest(manifest) == ["sections/verse/piano.mid"]
162
163
164 # ---------------------------------------------------------------------------
165 # Async integration tests — _show_async
166 # ---------------------------------------------------------------------------
167
168
169 @pytest.mark.anyio
170 async def test_muse_show_displays_commit_metadata(
171 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
172 ) -> None:
173 """Regression: show returns correct metadata for a committed snapshot."""
174 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
175 try:
176 _init_muse_repo(tmp_path)
177 commit_id = await _make_commit(
178 tmp_path,
179 muse_cli_db_session,
180 "Add piano melody to verse",
181 {"piano.mid": b"\x00\x01"},
182 )
183 muse_dir = tmp_path / ".muse"
184 result = await _show_async(
185 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
186 )
187 assert result["commit_id"] == commit_id
188 assert result["message"] == "Add piano melody to verse"
189 assert result["branch"] == "main"
190 assert "piano.mid" in result["snapshot_manifest"]
191 assert result["parent_commit_id"] is None # root commit
192 finally:
193 del os.environ["MUSE_REPO_ROOT"]
194
195
196 @pytest.mark.anyio
197 async def test_muse_show_json_output(
198 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
199 ) -> None:
200 """show returns a JSON-serialisable dict with all required fields."""
201 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
202 try:
203 _init_muse_repo(tmp_path)
204 commit_id = await _make_commit(
205 tmp_path,
206 muse_cli_db_session,
207 "boom bap demo",
208 {"beat.mid": b"\x00\x02"},
209 )
210 muse_dir = tmp_path / ".muse"
211 result = await _show_async(
212 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
213 )
214 # Must be JSON-serialisable without error
215 serialised = json.dumps(dict(result))
216 payload = json.loads(serialised)
217 assert payload["commit_id"] == commit_id
218 assert payload["message"] == "boom bap demo"
219 assert "snapshot_manifest" in payload
220 assert "beat.mid" in payload["snapshot_manifest"]
221 assert "committed_at" in payload
222 assert "snapshot_id" in payload
223 finally:
224 del os.environ["MUSE_REPO_ROOT"]
225
226
227 @pytest.mark.anyio
228 async def test_muse_show_diff_vs_parent(
229 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
230 ) -> None:
231 """show --diff: shows added/modified/removed paths vs parent commit."""
232 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
233 try:
234 _init_muse_repo(tmp_path)
235 muse_dir = tmp_path / ".muse"
236
237 # First commit: beat.mid + keys.mid
238 await _make_commit(
239 tmp_path,
240 muse_cli_db_session,
241 "initial take",
242 {"beat.mid": b"\x01", "keys.mid": b"\x02"},
243 )
244 # Second commit: beat.mid changed, bass.mid added, keys.mid removed
245 commit_id = await _make_commit(
246 tmp_path,
247 muse_cli_db_session,
248 "revise arrangement",
249 {"beat.mid": b"\xff", "bass.mid": b"\x03"},
250 )
251
252 diff_result = await _diff_vs_parent_async(
253 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
254 )
255
256 assert diff_result["commit_id"] == commit_id
257 assert diff_result["parent_commit_id"] is not None
258 assert "bass.mid" in diff_result["added"]
259 assert "beat.mid" in diff_result["modified"]
260 assert "keys.mid" in diff_result["removed"]
261 assert diff_result["total_changed"] == 3
262 finally:
263 del os.environ["MUSE_REPO_ROOT"]
264
265
266 @pytest.mark.anyio
267 async def test_muse_show_diff_root_commit(
268 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
269 ) -> None:
270 """show --diff on a root commit treats all snapshot paths as 'added'."""
271 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
272 try:
273 _init_muse_repo(tmp_path)
274 muse_dir = tmp_path / ".muse"
275 commit_id = await _make_commit(
276 tmp_path,
277 muse_cli_db_session,
278 "initial commit",
279 {"intro.mid": b"\x01", "verse.mid": b"\x02"},
280 )
281 diff_result = await _diff_vs_parent_async(
282 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
283 )
284 assert diff_result["parent_commit_id"] is None
285 assert "intro.mid" in diff_result["added"]
286 assert "verse.mid" in diff_result["added"]
287 assert diff_result["modified"] == []
288 assert diff_result["removed"] == []
289 assert diff_result["total_changed"] == 2
290 finally:
291 del os.environ["MUSE_REPO_ROOT"]
292
293
294 @pytest.mark.anyio
295 async def test_muse_show_midi_list(
296 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
297 ) -> None:
298 """show --midi: snapshot manifest filtered to MIDI files only."""
299 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
300 try:
301 _init_muse_repo(tmp_path)
302 muse_dir = tmp_path / ".muse"
303 commit_id = await _make_commit(
304 tmp_path,
305 muse_cli_db_session,
306 "mixed file commit",
307 {
308 "beat.mid": b"\x01",
309 "keys.midi": b"\x02",
310 "readme.txt": b"notes",
311 "cover.png": b"\x89PNG",
312 },
313 )
314 result = await _show_async(
315 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
316 )
317 midi_files = _midi_files_in_manifest(result["snapshot_manifest"])
318 assert "beat.mid" in midi_files
319 assert "keys.midi" in midi_files
320 assert "readme.txt" not in midi_files
321 assert "cover.png" not in midi_files
322 finally:
323 del os.environ["MUSE_REPO_ROOT"]
324
325
326 @pytest.mark.anyio
327 async def test_muse_show_audio_preview(
328 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
329 ) -> None:
330 """show --audio-preview: when no export dir, returns a helpful message stub."""
331 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
332 try:
333 _init_muse_repo(tmp_path)
334 muse_dir = tmp_path / ".muse"
335 commit_id = await _make_commit(
336 tmp_path,
337 muse_cli_db_session,
338 "needs audio",
339 {"track.mid": b"\x01"},
340 )
341 result = await _show_async(
342 session=muse_cli_db_session, muse_dir=muse_dir, ref=commit_id
343 )
344 # No export directory exists → audio preview is stubbed
345 export_dir = tmp_path / ".muse" / "exports" / commit_id[:8]
346 assert not export_dir.exists()
347 # Verify the commit itself is valid (preview logic is tested via render function)
348 assert result["commit_id"] == commit_id
349 finally:
350 del os.environ["MUSE_REPO_ROOT"]
351
352
353 @pytest.mark.anyio
354 async def test_muse_show_head_resolution(
355 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
356 ) -> None:
357 """show resolves 'HEAD' to the current branch tip."""
358 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
359 try:
360 _init_muse_repo(tmp_path)
361 muse_dir = tmp_path / ".muse"
362 commit_id = await _make_commit(
363 tmp_path, muse_cli_db_session, "HEAD test", {"a.mid": b"\x01"}
364 )
365 result = await _show_async(
366 session=muse_cli_db_session, muse_dir=muse_dir, ref="HEAD"
367 )
368 assert result["commit_id"] == commit_id
369 finally:
370 del os.environ["MUSE_REPO_ROOT"]
371
372
373 @pytest.mark.anyio
374 async def test_muse_show_branch_name_resolution(
375 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
376 ) -> None:
377 """show resolves a branch name to its tip commit."""
378 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
379 try:
380 _init_muse_repo(tmp_path)
381 muse_dir = tmp_path / ".muse"
382 commit_id = await _make_commit(
383 tmp_path, muse_cli_db_session, "branch name test", {"b.mid": b"\x02"}
384 )
385 # muse commit writes the commit_id to refs/heads/main
386 result = await _show_async(
387 session=muse_cli_db_session, muse_dir=muse_dir, ref="main"
388 )
389 assert result["commit_id"] == commit_id
390 finally:
391 del os.environ["MUSE_REPO_ROOT"]
392
393
394 @pytest.mark.anyio
395 async def test_muse_show_prefix_resolution(
396 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
397 ) -> None:
398 """show resolves a short hex prefix to the matching commit."""
399 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
400 try:
401 _init_muse_repo(tmp_path)
402 muse_dir = tmp_path / ".muse"
403 commit_id = await _make_commit(
404 tmp_path, muse_cli_db_session, "prefix test", {"c.mid": b"\x03"}
405 )
406 prefix = commit_id[:8]
407 result = await _show_async(
408 session=muse_cli_db_session, muse_dir=muse_dir, ref=prefix
409 )
410 assert result["commit_id"] == commit_id
411 finally:
412 del os.environ["MUSE_REPO_ROOT"]
413
414
415 @pytest.mark.anyio
416 async def test_muse_show_unknown_ref_exits_user_error(
417 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
418 ) -> None:
419 """show exits with USER_ERROR when the branch/ref does not exist."""
420 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
421 try:
422 _init_muse_repo(tmp_path)
423 muse_dir = tmp_path / ".muse"
424 with pytest.raises(typer.Exit) as exc_info:
425 await _show_async(
426 session=muse_cli_db_session, muse_dir=muse_dir, ref="nonexistent-branch"
427 )
428 assert exc_info.value.exit_code == ExitCode.USER_ERROR
429 finally:
430 del os.environ["MUSE_REPO_ROOT"]
431
432
433 @pytest.mark.anyio
434 async def test_muse_show_ambiguous_prefix_exits_user_error(
435 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
436 ) -> None:
437 """show exits with USER_ERROR when a hex prefix matches multiple commits."""
438 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
439 try:
440 _init_muse_repo(tmp_path)
441 muse_dir = tmp_path / ".muse"
442
443 # Create two commits whose IDs share the same first character (unlikely
444 # but we force it by inserting commits directly via the ORM).
445 from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot
446 from datetime import datetime, timezone
447
448 shared_prefix = "0000"
449 snap1_id = "snap001" + "x" * 57
450 snap2_id = "snap002" + "x" * 57
451
452 snap1 = MuseCliSnapshot(snapshot_id=snap1_id, manifest={})
453 snap2 = MuseCliSnapshot(snapshot_id=snap2_id, manifest={})
454 muse_cli_db_session.add(snap1)
455 muse_cli_db_session.add(snap2)
456
457 now = datetime.now(timezone.utc)
458 commit1 = MuseCliCommit(
459 commit_id=shared_prefix + "aaaa" + "b" * 56,
460 repo_id="repo1",
461 branch="main",
462 snapshot_id=snap1_id,
463 message="commit one",
464 author="test",
465 committed_at=now,
466 )
467 commit2 = MuseCliCommit(
468 commit_id=shared_prefix + "bbbb" + "c" * 56,
469 repo_id="repo1",
470 branch="main",
471 snapshot_id=snap2_id,
472 message="commit two",
473 author="test",
474 committed_at=now,
475 )
476 muse_cli_db_session.add(commit1)
477 muse_cli_db_session.add(commit2)
478 await muse_cli_db_session.flush()
479
480 with pytest.raises(typer.Exit) as exc_info:
481 await _show_async(
482 session=muse_cli_db_session,
483 muse_dir=muse_dir,
484 ref=shared_prefix,
485 )
486 assert exc_info.value.exit_code == ExitCode.USER_ERROR
487 finally:
488 del os.environ["MUSE_REPO_ROOT"]
489
490
491 @pytest.mark.anyio
492 async def test_muse_show_commit_with_parent(
493 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
494 ) -> None:
495 """show includes parent_commit_id when a parent exists."""
496 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
497 try:
498 _init_muse_repo(tmp_path)
499 muse_dir = tmp_path / ".muse"
500
501 parent_id = await _make_commit(
502 tmp_path, muse_cli_db_session, "first", {"a.mid": b"\x01"}
503 )
504 child_id = await _make_commit(
505 tmp_path, muse_cli_db_session, "second", {"b.mid": b"\x02"}
506 )
507
508 result = await _show_async(
509 session=muse_cli_db_session, muse_dir=muse_dir, ref=child_id
510 )
511 assert result["parent_commit_id"] == parent_id
512 finally:
513 del os.environ["MUSE_REPO_ROOT"]
514
515
516 @pytest.mark.anyio
517 async def test_muse_show_diff_identical_files_across_commits(
518 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
519 ) -> None:
520 """show --diff reports zero changes when only an untracked file changes.
521
522 We commit twice with different messages but the same *tracked* file
523 content, then verify that the diff between the second commit and its
524 parent is empty. We use slightly different file contents to avoid the
525 'nothing to commit' early-exit from _commit_async.
526 """
527 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
528 try:
529 _init_muse_repo(tmp_path)
530 muse_dir = tmp_path / ".muse"
531
532 # First commit: a.mid with content 0x01
533 await _make_commit(
534 tmp_path, muse_cli_db_session, "first", {"a.mid": b"\x01"}
535 )
536 # Second commit: same file, different content (forces a new commit)
537 # Both commits have a.mid; we then compare the diff output.
538 # To test zero-change scenario, commit the *same* content again under
539 # a different message via a direct DB insert that shares the snapshot.
540 from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot
541 from datetime import datetime, timezone
542
543 # Grab the current HEAD commit to get its snapshot_id
544 head_ref_text = (muse_dir / "HEAD").read_text().strip()
545 first_commit_id = (muse_dir / pathlib.Path(head_ref_text)).read_text().strip()
546 first_commit = await muse_cli_db_session.get(MuseCliCommit, first_commit_id)
547 assert first_commit is not None
548
549 # Create a second commit that points to the same snapshot (identical content)
550 now = datetime.now(timezone.utc)
551 second_commit_id = "ee" * 32 # deterministic, distinct ID
552 second_commit = MuseCliCommit(
553 commit_id=second_commit_id,
554 repo_id="test-repo",
555 branch="main",
556 parent_commit_id=first_commit_id,
557 snapshot_id=first_commit.snapshot_id, # same snapshot
558 message="second commit same snapshot",
559 author="test",
560 committed_at=now,
561 )
562 muse_cli_db_session.add(second_commit)
563 await muse_cli_db_session.flush()
564
565 diff_result = await _diff_vs_parent_async(
566 session=muse_cli_db_session, muse_dir=muse_dir, ref=second_commit_id
567 )
568 assert diff_result["total_changed"] == 0
569 assert diff_result["added"] == []
570 assert diff_result["modified"] == []
571 assert diff_result["removed"] == []
572 finally:
573 del os.environ["MUSE_REPO_ROOT"]