test_render_preview.py
python
| 1 | """Tests for ``muse render-preview`` command and ``muse_render_preview`` service. |
| 2 | |
| 3 | Test matrix: |
| 4 | - ``test_render_preview_outputs_path_for_head`` |
| 5 | - ``test_render_preview_service_returns_result_with_correct_fields`` |
| 6 | - ``test_render_preview_service_filter_by_track`` |
| 7 | - ``test_render_preview_service_filter_by_section`` |
| 8 | - ``test_render_preview_service_raises_when_no_midi_after_filter`` |
| 9 | - ``test_render_preview_service_raises_when_storpheus_unreachable`` |
| 10 | - ``test_render_preview_service_uses_custom_output_path`` |
| 11 | - ``test_render_preview_service_mp3_format`` |
| 12 | - ``test_render_preview_service_flac_format`` |
| 13 | - ``test_render_preview_service_skips_non_midi_files`` |
| 14 | - ``test_render_preview_service_skips_missing_files`` |
| 15 | - ``test_render_preview_cli_head_commit`` |
| 16 | - ``test_render_preview_cli_json_output`` |
| 17 | - ``test_render_preview_cli_no_repo`` |
| 18 | - ``test_render_preview_cli_no_commits`` |
| 19 | - ``test_render_preview_cli_ambiguous_prefix`` |
| 20 | - ``test_render_preview_cli_empty_snapshot`` |
| 21 | - ``test_render_preview_cli_storpheus_unreachable`` |
| 22 | - ``test_render_preview_cli_custom_format_and_output`` |
| 23 | - ``test_render_preview_async_core_resolves_head`` |
| 24 | - ``test_default_output_path_uses_tmp`` |
| 25 | """ |
| 26 | from __future__ import annotations |
| 27 | |
| 28 | import json |
| 29 | import pathlib |
| 30 | import uuid |
| 31 | from typing import Any |
| 32 | from unittest.mock import MagicMock, patch |
| 33 | |
| 34 | import pytest |
| 35 | import pytest_asyncio |
| 36 | from sqlalchemy.ext.asyncio import AsyncSession |
| 37 | from typer.testing import CliRunner |
| 38 | |
| 39 | from maestro.muse_cli.app import cli |
| 40 | from maestro.muse_cli.commands.render_preview import ( |
| 41 | _default_output_path, |
| 42 | _render_preview_async, |
| 43 | ) |
| 44 | from maestro.muse_cli.snapshot import hash_file |
| 45 | from maestro.services.muse_render_preview import ( |
| 46 | PreviewFormat, |
| 47 | RenderPreviewResult, |
| 48 | StorpheusRenderUnavailableError, |
| 49 | _collect_midi_files, |
| 50 | render_preview, |
| 51 | ) |
| 52 | |
| 53 | runner = CliRunner() |
| 54 | |
| 55 | |
| 56 | # --------------------------------------------------------------------------- |
| 57 | # Fixtures / helpers |
| 58 | # --------------------------------------------------------------------------- |
| 59 | |
| 60 | |
| 61 | def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str: |
| 62 | """Create a minimal .muse/ layout for CLI tests.""" |
| 63 | rid = repo_id or str(uuid.uuid4()) |
| 64 | muse = root / ".muse" |
| 65 | (muse / "refs" / "heads").mkdir(parents=True) |
| 66 | (muse / "repo.json").write_text( |
| 67 | json.dumps({"repo_id": rid, "schema_version": "1"}) |
| 68 | ) |
| 69 | (muse / "HEAD").write_text("refs/heads/main") |
| 70 | (muse / "refs" / "heads" / "main").write_text("") |
| 71 | return rid |
| 72 | |
| 73 | |
| 74 | def _set_head(root: pathlib.Path, commit_id: str) -> None: |
| 75 | """Point the HEAD of the main branch at commit_id.""" |
| 76 | ref_path = root / ".muse" / "refs" / "heads" / "main" |
| 77 | ref_path.write_text(commit_id) |
| 78 | |
| 79 | |
| 80 | def _make_minimal_midi() -> bytes: |
| 81 | """Return a minimal well-formed MIDI file (single note, type 0).""" |
| 82 | header = b"MThd\x00\x00\x00\x06\x00\x00\x00\x01\x01\xe0" |
| 83 | track_data = ( |
| 84 | b"\x00\x90\x3c\x40" |
| 85 | b"\x81\x60\x80\x3c\x00" |
| 86 | b"\x00\xff\x2f\x00" |
| 87 | ) |
| 88 | track_len = len(track_data).to_bytes(4, "big") |
| 89 | return header + b"MTrk" + track_len + track_data |
| 90 | |
| 91 | |
| 92 | def _make_manifest_with_midi( |
| 93 | tmp_path: pathlib.Path, |
| 94 | filenames: list[str] | None = None, |
| 95 | ) -> dict[str, str]: |
| 96 | """Write MIDI files to muse-work/ and return a manifest dict. |
| 97 | |
| 98 | Creates parent subdirectories as needed so callers can pass paths like |
| 99 | ``"drums/beat.mid"`` or ``"chorus/piano.mid"``. |
| 100 | """ |
| 101 | workdir = tmp_path / "muse-work" |
| 102 | workdir.mkdir(exist_ok=True) |
| 103 | filenames = filenames or ["beat.mid"] |
| 104 | midi_bytes = _make_minimal_midi() |
| 105 | manifest: dict[str, str] = {} |
| 106 | for name in filenames: |
| 107 | p = workdir / name |
| 108 | p.parent.mkdir(parents=True, exist_ok=True) |
| 109 | p.write_bytes(midi_bytes) |
| 110 | manifest[name] = hash_file(p) |
| 111 | return manifest |
| 112 | |
| 113 | |
| 114 | def _storpheus_healthy_mock() -> MagicMock: |
| 115 | """Return a mock httpx.Client whose GET /health returns 200.""" |
| 116 | mock_resp = MagicMock() |
| 117 | mock_resp.status_code = 200 |
| 118 | mock_client = MagicMock() |
| 119 | mock_client.__enter__ = MagicMock(return_value=mock_client) |
| 120 | mock_client.__exit__ = MagicMock(return_value=False) |
| 121 | mock_client.get = MagicMock(return_value=mock_resp) |
| 122 | return mock_client |
| 123 | |
| 124 | |
| 125 | # --------------------------------------------------------------------------- |
| 126 | # Unit tests — _default_output_path |
| 127 | # --------------------------------------------------------------------------- |
| 128 | |
| 129 | |
| 130 | def test_default_output_path_uses_tmp() -> None: |
| 131 | """_default_output_path returns a /tmp/muse-preview-<short>.<fmt> path.""" |
| 132 | commit_id = "abcdef1234567890" + "0" * 48 |
| 133 | path = _default_output_path(commit_id, PreviewFormat.WAV) |
| 134 | assert str(path).startswith("/tmp/muse-preview-abcdef12") |
| 135 | assert path.suffix == ".wav" |
| 136 | |
| 137 | |
| 138 | def test_default_output_path_mp3() -> None: |
| 139 | """_default_output_path uses the correct extension for mp3.""" |
| 140 | commit_id = "ff00ff1234567890" + "0" * 48 |
| 141 | path = _default_output_path(commit_id, PreviewFormat.MP3) |
| 142 | assert path.suffix == ".mp3" |
| 143 | |
| 144 | |
| 145 | def test_default_output_path_flac() -> None: |
| 146 | """_default_output_path uses the correct extension for flac.""" |
| 147 | commit_id = "aa00aa1234567890" + "0" * 48 |
| 148 | path = _default_output_path(commit_id, PreviewFormat.FLAC) |
| 149 | assert path.suffix == ".flac" |
| 150 | |
| 151 | |
| 152 | # --------------------------------------------------------------------------- |
| 153 | # Unit tests — _collect_midi_files |
| 154 | # --------------------------------------------------------------------------- |
| 155 | |
| 156 | |
| 157 | def test_collect_midi_files_returns_all_midi(tmp_path: pathlib.Path) -> None: |
| 158 | """_collect_midi_files returns all MIDI paths when no filter is set.""" |
| 159 | manifest = _make_manifest_with_midi(tmp_path, ["drums.mid", "bass.mid"]) |
| 160 | paths, skipped = _collect_midi_files(manifest, tmp_path, track=None, section=None) |
| 161 | assert len(paths) == 2 |
| 162 | assert skipped == 0 |
| 163 | |
| 164 | |
| 165 | def test_collect_midi_files_skips_non_midi(tmp_path: pathlib.Path) -> None: |
| 166 | """_collect_midi_files skips non-MIDI entries and counts them as skipped.""" |
| 167 | workdir = tmp_path / "muse-work" |
| 168 | workdir.mkdir() |
| 169 | (workdir / "beat.mid").write_bytes(_make_minimal_midi()) |
| 170 | (workdir / "notes.json").write_text("{}") |
| 171 | manifest = { |
| 172 | "beat.mid": hash_file(workdir / "beat.mid"), |
| 173 | "notes.json": hash_file(workdir / "notes.json"), |
| 174 | } |
| 175 | paths, skipped = _collect_midi_files(manifest, tmp_path, track=None, section=None) |
| 176 | assert len(paths) == 1 |
| 177 | assert skipped == 1 |
| 178 | |
| 179 | |
| 180 | def test_collect_midi_files_filter_by_track(tmp_path: pathlib.Path) -> None: |
| 181 | """_collect_midi_files applies the track substring filter.""" |
| 182 | manifest = _make_manifest_with_midi( |
| 183 | tmp_path, ["drums/beat.mid", "bass/groove.mid"] |
| 184 | ) |
| 185 | paths, skipped = _collect_midi_files(manifest, tmp_path, track="drums", section=None) |
| 186 | assert len(paths) == 1 |
| 187 | assert "drums" in str(paths[0]) |
| 188 | assert skipped == 1 |
| 189 | |
| 190 | |
| 191 | def test_collect_midi_files_filter_by_section(tmp_path: pathlib.Path) -> None: |
| 192 | """_collect_midi_files applies the section substring filter.""" |
| 193 | manifest = _make_manifest_with_midi( |
| 194 | tmp_path, ["chorus/piano.mid", "verse/piano.mid"] |
| 195 | ) |
| 196 | paths, skipped = _collect_midi_files( |
| 197 | manifest, tmp_path, track=None, section="chorus" |
| 198 | ) |
| 199 | assert len(paths) == 1 |
| 200 | assert "chorus" in str(paths[0]) |
| 201 | assert skipped == 1 |
| 202 | |
| 203 | |
| 204 | def test_collect_midi_files_skips_missing_file(tmp_path: pathlib.Path) -> None: |
| 205 | """_collect_midi_files counts missing files as skipped without raising.""" |
| 206 | workdir = tmp_path / "muse-work" |
| 207 | workdir.mkdir() |
| 208 | manifest = {"ghost.mid": "abc123"} |
| 209 | paths, skipped = _collect_midi_files(manifest, tmp_path, track=None, section=None) |
| 210 | assert paths == [] |
| 211 | assert skipped == 1 |
| 212 | |
| 213 | |
| 214 | # --------------------------------------------------------------------------- |
| 215 | # Unit tests — render_preview service |
| 216 | # --------------------------------------------------------------------------- |
| 217 | |
| 218 | |
| 219 | def test_render_preview_service_returns_result_with_correct_fields( |
| 220 | tmp_path: pathlib.Path, |
| 221 | ) -> None: |
| 222 | """render_preview returns a RenderPreviewResult with expected fields on success.""" |
| 223 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 224 | out = tmp_path / "preview.wav" |
| 225 | commit_id = "a" * 64 |
| 226 | |
| 227 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 228 | mock_cls.return_value = _storpheus_healthy_mock() |
| 229 | result = render_preview( |
| 230 | manifest=manifest, |
| 231 | root=tmp_path, |
| 232 | commit_id=commit_id, |
| 233 | output_path=out, |
| 234 | fmt=PreviewFormat.WAV, |
| 235 | ) |
| 236 | |
| 237 | assert isinstance(result, RenderPreviewResult) |
| 238 | assert result.output_path == out |
| 239 | assert result.format == PreviewFormat.WAV |
| 240 | assert result.commit_id == commit_id |
| 241 | assert result.midi_files_used == 1 |
| 242 | assert result.skipped_count == 0 |
| 243 | assert result.stubbed is True |
| 244 | assert out.exists() |
| 245 | |
| 246 | |
| 247 | def test_render_preview_outputs_path_for_head(tmp_path: pathlib.Path) -> None: |
| 248 | """Regression: render_preview writes the output file and returns its path.""" |
| 249 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 250 | out = tmp_path / "muse-preview-head.wav" |
| 251 | commit_id = "b" * 64 |
| 252 | |
| 253 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 254 | mock_cls.return_value = _storpheus_healthy_mock() |
| 255 | result = render_preview( |
| 256 | manifest=manifest, |
| 257 | root=tmp_path, |
| 258 | commit_id=commit_id, |
| 259 | output_path=out, |
| 260 | ) |
| 261 | |
| 262 | assert result.output_path.exists() |
| 263 | assert str(result.output_path) == str(out) |
| 264 | |
| 265 | |
| 266 | def test_render_preview_service_filter_by_track(tmp_path: pathlib.Path) -> None: |
| 267 | """render_preview respects the track filter and skips non-matching MIDI.""" |
| 268 | manifest = _make_manifest_with_midi( |
| 269 | tmp_path, ["drums/beat.mid", "bass/groove.mid"] |
| 270 | ) |
| 271 | out = tmp_path / "preview.wav" |
| 272 | commit_id = "c" * 64 |
| 273 | |
| 274 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 275 | mock_cls.return_value = _storpheus_healthy_mock() |
| 276 | result = render_preview( |
| 277 | manifest=manifest, |
| 278 | root=tmp_path, |
| 279 | commit_id=commit_id, |
| 280 | output_path=out, |
| 281 | track="drums", |
| 282 | ) |
| 283 | |
| 284 | assert result.midi_files_used == 1 |
| 285 | assert result.skipped_count == 1 |
| 286 | |
| 287 | |
| 288 | def test_render_preview_service_filter_by_section(tmp_path: pathlib.Path) -> None: |
| 289 | """render_preview respects the section filter and skips non-matching MIDI.""" |
| 290 | manifest = _make_manifest_with_midi( |
| 291 | tmp_path, ["chorus/lead.mid", "verse/lead.mid"] |
| 292 | ) |
| 293 | out = tmp_path / "preview.wav" |
| 294 | commit_id = "d" * 64 |
| 295 | |
| 296 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 297 | mock_cls.return_value = _storpheus_healthy_mock() |
| 298 | result = render_preview( |
| 299 | manifest=manifest, |
| 300 | root=tmp_path, |
| 301 | commit_id=commit_id, |
| 302 | output_path=out, |
| 303 | section="chorus", |
| 304 | ) |
| 305 | |
| 306 | assert result.midi_files_used == 1 |
| 307 | assert result.skipped_count == 1 |
| 308 | |
| 309 | |
| 310 | def test_render_preview_service_raises_when_no_midi_after_filter( |
| 311 | tmp_path: pathlib.Path, |
| 312 | ) -> None: |
| 313 | """render_preview raises ValueError when the filter leaves no MIDI files.""" |
| 314 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 315 | commit_id = "e" * 64 |
| 316 | |
| 317 | with pytest.raises(ValueError, match="No MIDI files found"): |
| 318 | render_preview( |
| 319 | manifest=manifest, |
| 320 | root=tmp_path, |
| 321 | commit_id=commit_id, |
| 322 | output_path=tmp_path / "out.wav", |
| 323 | track="nonexistent_track_xyz", |
| 324 | ) |
| 325 | |
| 326 | |
| 327 | def test_render_preview_service_raises_when_storpheus_unreachable( |
| 328 | tmp_path: pathlib.Path, |
| 329 | ) -> None: |
| 330 | """render_preview raises StorpheusRenderUnavailableError when Storpheus is down.""" |
| 331 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 332 | commit_id = "f" * 64 |
| 333 | |
| 334 | with patch( |
| 335 | "maestro.services.muse_render_preview.httpx.Client", |
| 336 | side_effect=Exception("connection refused"), |
| 337 | ): |
| 338 | with pytest.raises(StorpheusRenderUnavailableError): |
| 339 | render_preview( |
| 340 | manifest=manifest, |
| 341 | root=tmp_path, |
| 342 | commit_id=commit_id, |
| 343 | output_path=tmp_path / "out.wav", |
| 344 | ) |
| 345 | |
| 346 | |
| 347 | def test_render_preview_service_raises_when_storpheus_non_200( |
| 348 | tmp_path: pathlib.Path, |
| 349 | ) -> None: |
| 350 | """render_preview raises StorpheusRenderUnavailableError on non-200 health check.""" |
| 351 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 352 | commit_id = "g" * 64 |
| 353 | |
| 354 | mock_resp = MagicMock() |
| 355 | mock_resp.status_code = 503 |
| 356 | mock_client = MagicMock() |
| 357 | mock_client.__enter__ = MagicMock(return_value=mock_client) |
| 358 | mock_client.__exit__ = MagicMock(return_value=False) |
| 359 | mock_client.get = MagicMock(return_value=mock_resp) |
| 360 | |
| 361 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 362 | mock_cls.return_value = mock_client |
| 363 | with pytest.raises(StorpheusRenderUnavailableError): |
| 364 | render_preview( |
| 365 | manifest=manifest, |
| 366 | root=tmp_path, |
| 367 | commit_id=commit_id, |
| 368 | output_path=tmp_path / "out.wav", |
| 369 | ) |
| 370 | |
| 371 | |
| 372 | def test_render_preview_service_mp3_format(tmp_path: pathlib.Path) -> None: |
| 373 | """render_preview sets the correct format on the result for mp3.""" |
| 374 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 375 | out = tmp_path / "preview.mp3" |
| 376 | commit_id = "h" * 64 |
| 377 | |
| 378 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 379 | mock_cls.return_value = _storpheus_healthy_mock() |
| 380 | result = render_preview( |
| 381 | manifest=manifest, |
| 382 | root=tmp_path, |
| 383 | commit_id=commit_id, |
| 384 | output_path=out, |
| 385 | fmt=PreviewFormat.MP3, |
| 386 | ) |
| 387 | |
| 388 | assert result.format == PreviewFormat.MP3 |
| 389 | |
| 390 | |
| 391 | def test_render_preview_service_flac_format(tmp_path: pathlib.Path) -> None: |
| 392 | """render_preview sets the correct format on the result for flac.""" |
| 393 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 394 | out = tmp_path / "preview.flac" |
| 395 | commit_id = "i" * 64 |
| 396 | |
| 397 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 398 | mock_cls.return_value = _storpheus_healthy_mock() |
| 399 | result = render_preview( |
| 400 | manifest=manifest, |
| 401 | root=tmp_path, |
| 402 | commit_id=commit_id, |
| 403 | output_path=out, |
| 404 | fmt=PreviewFormat.FLAC, |
| 405 | ) |
| 406 | |
| 407 | assert result.format == PreviewFormat.FLAC |
| 408 | |
| 409 | |
| 410 | def test_render_preview_service_skips_non_midi_files(tmp_path: pathlib.Path) -> None: |
| 411 | """render_preview counts non-MIDI manifest entries as skipped.""" |
| 412 | workdir = tmp_path / "muse-work" |
| 413 | workdir.mkdir() |
| 414 | (workdir / "beat.mid").write_bytes(_make_minimal_midi()) |
| 415 | (workdir / "meta.json").write_text("{}") |
| 416 | manifest = { |
| 417 | "beat.mid": hash_file(workdir / "beat.mid"), |
| 418 | "meta.json": hash_file(workdir / "meta.json"), |
| 419 | } |
| 420 | out = tmp_path / "preview.wav" |
| 421 | commit_id = "j" * 64 |
| 422 | |
| 423 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 424 | mock_cls.return_value = _storpheus_healthy_mock() |
| 425 | result = render_preview( |
| 426 | manifest=manifest, |
| 427 | root=tmp_path, |
| 428 | commit_id=commit_id, |
| 429 | output_path=out, |
| 430 | ) |
| 431 | |
| 432 | assert result.midi_files_used == 1 |
| 433 | assert result.skipped_count == 1 |
| 434 | |
| 435 | |
| 436 | def test_render_preview_service_uses_custom_output_path( |
| 437 | tmp_path: pathlib.Path, |
| 438 | ) -> None: |
| 439 | """render_preview writes to the caller-supplied output_path.""" |
| 440 | manifest = _make_manifest_with_midi(tmp_path, ["beat.mid"]) |
| 441 | custom_out = tmp_path / "custom" / "my-preview.wav" |
| 442 | commit_id = "k" * 64 |
| 443 | |
| 444 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 445 | mock_cls.return_value = _storpheus_healthy_mock() |
| 446 | result = render_preview( |
| 447 | manifest=manifest, |
| 448 | root=tmp_path, |
| 449 | commit_id=commit_id, |
| 450 | output_path=custom_out, |
| 451 | ) |
| 452 | |
| 453 | assert result.output_path == custom_out |
| 454 | assert custom_out.exists() |
| 455 | |
| 456 | |
| 457 | # --------------------------------------------------------------------------- |
| 458 | # Integration tests — _render_preview_async (injectable core) |
| 459 | # --------------------------------------------------------------------------- |
| 460 | |
| 461 | |
| 462 | @pytest.mark.anyio |
| 463 | async def test_render_preview_async_core_resolves_head( |
| 464 | tmp_path: pathlib.Path, |
| 465 | muse_cli_db_session: AsyncSession, |
| 466 | ) -> None: |
| 467 | """_render_preview_async resolves HEAD and returns a RenderPreviewResult.""" |
| 468 | from datetime import datetime, timezone |
| 469 | |
| 470 | from maestro.muse_cli.db import insert_commit, upsert_object, upsert_snapshot |
| 471 | from maestro.muse_cli.models import MuseCliCommit |
| 472 | from maestro.muse_cli.snapshot import compute_commit_id, compute_snapshot_id |
| 473 | |
| 474 | repo_id = _init_muse_repo(tmp_path) |
| 475 | workdir = tmp_path / "muse-work" |
| 476 | workdir.mkdir() |
| 477 | (workdir / "beat.mid").write_bytes(_make_minimal_midi()) |
| 478 | |
| 479 | oid = hash_file(workdir / "beat.mid") |
| 480 | manifest = {"beat.mid": oid} |
| 481 | snapshot_id = compute_snapshot_id(manifest) |
| 482 | ts = datetime(2026, 1, 1, tzinfo=timezone.utc) |
| 483 | commit_id = compute_commit_id( |
| 484 | parent_ids=[], |
| 485 | snapshot_id=snapshot_id, |
| 486 | message="initial", |
| 487 | committed_at_iso=ts.isoformat(), |
| 488 | ) |
| 489 | |
| 490 | await upsert_object(muse_cli_db_session, oid, 100) |
| 491 | await upsert_snapshot(muse_cli_db_session, manifest=manifest, snapshot_id=snapshot_id) |
| 492 | await muse_cli_db_session.flush() |
| 493 | |
| 494 | commit = MuseCliCommit( |
| 495 | commit_id=commit_id, |
| 496 | repo_id=repo_id, |
| 497 | branch="main", |
| 498 | parent_commit_id=None, |
| 499 | snapshot_id=snapshot_id, |
| 500 | message="initial", |
| 501 | author="", |
| 502 | committed_at=ts, |
| 503 | ) |
| 504 | await insert_commit(muse_cli_db_session, commit) |
| 505 | await muse_cli_db_session.flush() |
| 506 | |
| 507 | _set_head(tmp_path, commit_id) |
| 508 | |
| 509 | out = tmp_path / "preview.wav" |
| 510 | with patch("maestro.services.muse_render_preview.httpx.Client") as mock_cls: |
| 511 | mock_cls.return_value = _storpheus_healthy_mock() |
| 512 | with patch("maestro.muse_cli.commands.render_preview.settings") as mock_settings: |
| 513 | mock_settings.storpheus_base_url = "http://storpheus:10002" |
| 514 | result = await _render_preview_async( |
| 515 | commit_ref=None, |
| 516 | fmt=PreviewFormat.WAV, |
| 517 | output=out, |
| 518 | track=None, |
| 519 | section=None, |
| 520 | root=tmp_path, |
| 521 | session=muse_cli_db_session, |
| 522 | ) |
| 523 | |
| 524 | assert result.output_path == out |
| 525 | assert result.midi_files_used == 1 |
| 526 | assert result.commit_id == commit_id |
| 527 | |
| 528 | |
| 529 | # --------------------------------------------------------------------------- |
| 530 | # CLI integration tests — typer CliRunner |
| 531 | # --------------------------------------------------------------------------- |
| 532 | |
| 533 | |
| 534 | def test_render_preview_cli_no_repo(tmp_path: pathlib.Path) -> None: |
| 535 | """muse render-preview exits with REPO_NOT_FOUND when not in a Muse repo.""" |
| 536 | import os |
| 537 | |
| 538 | with runner.isolated_filesystem(temp_dir=tmp_path): |
| 539 | result = runner.invoke(cli, ["render-preview"]) |
| 540 | assert result.exit_code != 0 |
| 541 | assert "not a muse repository" in result.output.lower() or result.exit_code == 2 |
| 542 | |
| 543 | |
| 544 | def test_render_preview_cli_no_commits(tmp_path: pathlib.Path) -> None: |
| 545 | """muse render-preview exits with USER_ERROR when HEAD has no commits.""" |
| 546 | _init_muse_repo(tmp_path) |
| 547 | |
| 548 | import os |
| 549 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 550 | result = runner.invoke(cli, ["render-preview"], env=env) |
| 551 | assert result.exit_code != 0 |
| 552 | |
| 553 | |
| 554 | def test_render_preview_cli_head_commit(tmp_path: pathlib.Path) -> None: |
| 555 | """muse render-preview renders HEAD and prints the output path.""" |
| 556 | from unittest.mock import patch as _patch |
| 557 | |
| 558 | _init_muse_repo(tmp_path) |
| 559 | workdir = tmp_path / "muse-work" |
| 560 | workdir.mkdir() |
| 561 | (workdir / "beat.mid").write_bytes(_make_minimal_midi()) |
| 562 | |
| 563 | commit_id = "abcdef" + "0" * 58 |
| 564 | _set_head(tmp_path, commit_id) |
| 565 | |
| 566 | mock_manifest = {"beat.mid": hash_file(workdir / "beat.mid")} |
| 567 | import os |
| 568 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 569 | |
| 570 | out = tmp_path / "preview.wav" |
| 571 | |
| 572 | with _patch( |
| 573 | "maestro.muse_cli.commands.render_preview.open_session" |
| 574 | ) as mock_session_cm: |
| 575 | mock_session = MagicMock() |
| 576 | |
| 577 | async def _fake_session_aenter(_: Any) -> Any: |
| 578 | return mock_session |
| 579 | |
| 580 | async def _fake_session_aexit(_: Any, *args: Any) -> None: |
| 581 | pass |
| 582 | |
| 583 | mock_session_cm.return_value.__aenter__ = _fake_session_aenter |
| 584 | mock_session_cm.return_value.__aexit__ = _fake_session_aexit |
| 585 | |
| 586 | with _patch( |
| 587 | "maestro.muse_cli.commands.render_preview._render_preview_async", |
| 588 | return_value=RenderPreviewResult( |
| 589 | output_path=out, |
| 590 | format=PreviewFormat.WAV, |
| 591 | commit_id=commit_id, |
| 592 | midi_files_used=1, |
| 593 | skipped_count=0, |
| 594 | stubbed=True, |
| 595 | ), |
| 596 | ): |
| 597 | result = runner.invoke( |
| 598 | cli, |
| 599 | ["render-preview", "--output", str(out)], |
| 600 | env=env, |
| 601 | ) |
| 602 | |
| 603 | assert result.exit_code == 0 |
| 604 | assert str(out) in result.output |
| 605 | |
| 606 | |
| 607 | def test_render_preview_cli_json_output(tmp_path: pathlib.Path) -> None: |
| 608 | """muse render-preview --json emits valid JSON with expected keys.""" |
| 609 | from unittest.mock import patch as _patch |
| 610 | |
| 611 | _init_muse_repo(tmp_path) |
| 612 | commit_id = "json00" + "0" * 58 |
| 613 | _set_head(tmp_path, commit_id) |
| 614 | out = tmp_path / "preview.wav" |
| 615 | |
| 616 | import os |
| 617 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 618 | |
| 619 | with _patch( |
| 620 | "maestro.muse_cli.commands.render_preview._render_preview_async", |
| 621 | return_value=RenderPreviewResult( |
| 622 | output_path=out, |
| 623 | format=PreviewFormat.WAV, |
| 624 | commit_id=commit_id, |
| 625 | midi_files_used=1, |
| 626 | skipped_count=0, |
| 627 | stubbed=True, |
| 628 | ), |
| 629 | ): |
| 630 | result = runner.invoke( |
| 631 | cli, |
| 632 | ["render-preview", "--json", "--output", str(out)], |
| 633 | env=env, |
| 634 | ) |
| 635 | |
| 636 | assert result.exit_code == 0 |
| 637 | payload = json.loads(result.output) |
| 638 | assert "output_path" in payload |
| 639 | assert "commit_id" in payload |
| 640 | assert "format" in payload |
| 641 | assert "stubbed" in payload |
| 642 | assert payload["stubbed"] is True |
| 643 | |
| 644 | |
| 645 | @pytest.mark.anyio |
| 646 | async def test_render_preview_cli_ambiguous_prefix( |
| 647 | tmp_path: pathlib.Path, |
| 648 | muse_cli_db_session: AsyncSession, |
| 649 | ) -> None: |
| 650 | """_render_preview_async exits with USER_ERROR when the prefix matches multiple commits.""" |
| 651 | from datetime import datetime, timezone |
| 652 | from unittest.mock import AsyncMock, patch as _patch |
| 653 | |
| 654 | from maestro.muse_cli.db import insert_commit, upsert_object, upsert_snapshot |
| 655 | from maestro.muse_cli.models import MuseCliCommit |
| 656 | from maestro.muse_cli.snapshot import compute_commit_id, compute_snapshot_id |
| 657 | |
| 658 | repo_id = _init_muse_repo(tmp_path) |
| 659 | workdir = tmp_path / "muse-work" |
| 660 | workdir.mkdir() |
| 661 | (workdir / "beat.mid").write_bytes(_make_minimal_midi()) |
| 662 | |
| 663 | oid = hash_file(workdir / "beat.mid") |
| 664 | manifest = {"beat.mid": oid} |
| 665 | snapshot_id = compute_snapshot_id(manifest) |
| 666 | ts = datetime(2026, 1, 1, tzinfo=timezone.utc) |
| 667 | |
| 668 | await upsert_object(muse_cli_db_session, oid, 100) |
| 669 | await upsert_snapshot(muse_cli_db_session, manifest=manifest, snapshot_id=snapshot_id) |
| 670 | await muse_cli_db_session.flush() |
| 671 | |
| 672 | # Insert two commits that share the same prefix |
| 673 | commit_a = compute_commit_id([], snapshot_id, "commit A", ts.isoformat()) |
| 674 | commit_b = compute_commit_id([], snapshot_id, "commit B", ts.isoformat()) |
| 675 | for cid, msg in [(commit_a, "commit A"), (commit_b, "commit B")]: |
| 676 | await insert_commit( |
| 677 | muse_cli_db_session, |
| 678 | MuseCliCommit( |
| 679 | commit_id=cid, |
| 680 | repo_id=repo_id, |
| 681 | branch="main", |
| 682 | parent_commit_id=None, |
| 683 | snapshot_id=snapshot_id, |
| 684 | message=msg, |
| 685 | author="", |
| 686 | committed_at=ts, |
| 687 | ), |
| 688 | ) |
| 689 | await muse_cli_db_session.flush() |
| 690 | _set_head(tmp_path, commit_a) |
| 691 | |
| 692 | # Patch find_commits_by_prefix to simulate an ambiguous prefix |
| 693 | with _patch( |
| 694 | "maestro.muse_cli.commands.render_preview.find_commits_by_prefix", |
| 695 | new=AsyncMock(return_value=[ |
| 696 | type("C", (), {"commit_id": commit_a, "message": "commit A"})(), |
| 697 | type("C", (), {"commit_id": commit_b, "message": "commit B"})(), |
| 698 | ]), |
| 699 | ): |
| 700 | with pytest.raises(typer.Exit) as exc_info: |
| 701 | await _render_preview_async( |
| 702 | commit_ref="abc", |
| 703 | fmt=PreviewFormat.WAV, |
| 704 | output=tmp_path / "preview.wav", |
| 705 | track=None, |
| 706 | section=None, |
| 707 | root=tmp_path, |
| 708 | session=muse_cli_db_session, |
| 709 | ) |
| 710 | |
| 711 | assert exc_info.value.exit_code != 0 |
| 712 | |
| 713 | |
| 714 | def test_render_preview_cli_storpheus_unreachable(tmp_path: pathlib.Path) -> None: |
| 715 | """muse render-preview exits with INTERNAL_ERROR when Storpheus is down.""" |
| 716 | from unittest.mock import patch as _patch |
| 717 | |
| 718 | _init_muse_repo(tmp_path) |
| 719 | commit_id = "stdown" + "0" * 58 |
| 720 | _set_head(tmp_path, commit_id) |
| 721 | |
| 722 | import os |
| 723 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 724 | |
| 725 | with _patch( |
| 726 | "maestro.muse_cli.commands.render_preview._render_preview_async", |
| 727 | side_effect=StorpheusRenderUnavailableError("Connection refused"), |
| 728 | ): |
| 729 | result = runner.invoke(cli, ["render-preview"], env=env) |
| 730 | |
| 731 | assert result.exit_code != 0 |
| 732 | assert "storpheus" in result.output.lower() |
| 733 | |
| 734 | |
| 735 | def test_render_preview_cli_empty_snapshot(tmp_path: pathlib.Path) -> None: |
| 736 | """muse render-preview exits with USER_ERROR for an empty snapshot.""" |
| 737 | from unittest.mock import patch as _patch |
| 738 | |
| 739 | _init_muse_repo(tmp_path) |
| 740 | commit_id = "empty0" + "0" * 58 |
| 741 | _set_head(tmp_path, commit_id) |
| 742 | |
| 743 | import os |
| 744 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 745 | |
| 746 | with _patch( |
| 747 | "maestro.muse_cli.commands.render_preview._render_preview_async", |
| 748 | side_effect=typer.Exit(code=1), |
| 749 | ): |
| 750 | result = runner.invoke(cli, ["render-preview"], env=env) |
| 751 | |
| 752 | assert result.exit_code != 0 |
| 753 | |
| 754 | |
| 755 | def test_render_preview_cli_custom_format_and_output(tmp_path: pathlib.Path) -> None: |
| 756 | """muse render-preview --format mp3 --output writes to the custom path.""" |
| 757 | from unittest.mock import patch as _patch |
| 758 | |
| 759 | _init_muse_repo(tmp_path) |
| 760 | commit_id = "mp3000" + "0" * 58 |
| 761 | _set_head(tmp_path, commit_id) |
| 762 | custom_out = tmp_path / "my-song.mp3" |
| 763 | |
| 764 | import os |
| 765 | env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)} |
| 766 | |
| 767 | with _patch( |
| 768 | "maestro.muse_cli.commands.render_preview._render_preview_async", |
| 769 | return_value=RenderPreviewResult( |
| 770 | output_path=custom_out, |
| 771 | format=PreviewFormat.MP3, |
| 772 | commit_id=commit_id, |
| 773 | midi_files_used=1, |
| 774 | skipped_count=0, |
| 775 | stubbed=True, |
| 776 | ), |
| 777 | ): |
| 778 | result = runner.invoke( |
| 779 | cli, |
| 780 | ["render-preview", "--format", "mp3", "--output", str(custom_out)], |
| 781 | env=env, |
| 782 | ) |
| 783 | |
| 784 | assert result.exit_code == 0 |
| 785 | assert str(custom_out) in result.output |
| 786 | |
| 787 | |
| 788 | # --------------------------------------------------------------------------- |
| 789 | # Additional imports needed for tests |
| 790 | # --------------------------------------------------------------------------- |
| 791 | |
| 792 | import typer # noqa: E402 (imported here for use in test bodies above) |