diff --git a/backend/src/waveform.py b/backend/src/waveform.py index 2385462..223e8ea 100644 --- a/backend/src/waveform.py +++ b/backend/src/waveform.py @@ -58,7 +58,7 @@ async def generate_waveform_svg(clip_id: str, filename: str) -> str: if len(audio_data) > DOWNSAMPLE_TARGET: # Simple slice-based downsampling (nearest-sample, no aliasing # concerns at this resolution — we're making bars, not audio) - indices = np.linspace(0, len(audio_data), DOWNSAMPLE_TARGET, dtype=int) + indices = np.linspace(0, len(audio_data) - 1, DOWNSAMPLE_TARGET, dtype=int) audio_data = audio_data[indices] # ── WAVEFORM COMPUTATION ────────────────────────────────────── diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..45e5766 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,139 @@ +"""Shared test fixtures for Lyria Studio backend.""" + +import asyncio +import os +import sys +from pathlib import Path + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient + +# ── Path setup: ensure src/ is importable ────────────────────── +BACKEND_SRC = Path(__file__).resolve().parent.parent / "src" +sys.path.insert(0, str(BACKEND_SRC)) + + +@pytest.fixture(scope="session") +def event_loop(): + """Create a single event loop for the entire test session.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest_asyncio.fixture +async def app_client(tmp_path): + """Provide an httpx AsyncClient wired to the FastAPI app + with a fresh temp database and data directories. + + Each test gets its own isolated DB and file directories. + """ + + # Create temp data directories + data_dir = tmp_path / "data" + data_dir.mkdir() + clips_dir = data_dir / "clips" + clips_dir.mkdir() + waveforms_dir = data_dir / "waveforms" + waveforms_dir.mkdir() + artwork_dir = data_dir / "artwork" + artwork_dir.mkdir() + test_db = data_dir / "lyria.db" + + # Set env vars BEFORE import so module-level Path() calls use test dirs + # (image_gen.py reads ARTWORK_DIR from env, main.py uses __file__-relative paths) + os.environ["ARTWORK_DIR"] = str(artwork_dir) + + # Import modules AFTER setting env vars + import image_gen as ig_mod + import database as db_mod + import waveform as wf_mod + import main + + # Save originals for restoration + orig_db_path = db_mod.DB_PATH + orig_clips_dir = main.CLIPS_DIR + orig_artwork_dir = main.ARTWORK_DIR + orig_wf_clips = wf_mod.CLIPS_DIR + orig_wf_wave = wf_mod.WAVEFORM_DIR + orig_ig_artwork = ig_mod.ARTWORK_DIR + + # Patch module-level path constants to test directories + db_mod.DB_PATH = test_db + main.CLIPS_DIR = clips_dir + main.ARTWORK_DIR = artwork_dir + wf_mod.CLIPS_DIR = clips_dir + wf_mod.WAVEFORM_DIR = waveforms_dir + ig_mod.ARTWORK_DIR = artwork_dir + + # Initialize fresh database + await db_mod.init_db() + + transport = ASGITransport(app=main.app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield { + "client": c, + "db": test_db, + "clips_dir": clips_dir, + "waveforms_dir": waveforms_dir, + "artwork_dir": artwork_dir, + "data_dir": data_dir, + } + + # Restore originals + db_mod.DB_PATH = orig_db_path + main.CLIPS_DIR = orig_clips_dir + main.ARTWORK_DIR = orig_artwork_dir + wf_mod.CLIPS_DIR = orig_wf_clips + wf_mod.WAVEFORM_DIR = orig_wf_wave + ig_mod.ARTWORK_DIR = orig_ig_artwork + + # Clean up env var + del os.environ["ARTWORK_DIR"] + + +# ── Helper: create a clip record + dummy MP3 ─────────────────── + +async def seed_test_clip(app_client, prompt="test prompt", model="lyria-3-clip-preview", duration=30): + """Create a clip record in the test DB with a dummy MP3 file.""" + import aiosqlite + from database import create_clip + + clip_id = await create_clip(prompt, model, duration) + + filename = f"{clip_id}.mp3" + mp3_path = app_client["clips_dir"] / filename + mp3_path.write_bytes(b"\xff\xfb\x90\x00" + b"\x00" * 4096) + + async with aiosqlite.connect(app_client["db"]) as db: + await db.execute( + "UPDATE clips SET status = ?, filename = ? WHERE id = ?", + ("ready", filename, clip_id), + ) + await db.commit() + + return clip_id + + +# ── Helper: create a real (ffmpeg-generated) MP3 ──────────────── + +def create_real_mp3(output_path, duration=2): + """Generate a real MP3 file using ffmpeg. + + Returns True if successful, False if ffmpeg is unavailable. + """ + import subprocess + try: + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", f"anullsrc=r=22050:cl=mono", + "-t", str(duration), "-q:a", "9", + str(output_path) + ], + capture_output=True, check=True, + ) + return True + except (FileNotFoundError, subprocess.CalledProcessError): + return False \ No newline at end of file diff --git a/backend/tests/test_analysis.py b/backend/tests/test_analysis.py new file mode 100644 index 0000000..9a2fb63 --- /dev/null +++ b/backend/tests/test_analysis.py @@ -0,0 +1,117 @@ +"""Tests for audio analysis module.""" + +import subprocess +import sys +from pathlib import Path + +import pytest +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +from analysis import analyze_clip, detect_bpm, detect_key, MAJOR_PROFILE, MINOR_PROFILE, KEY_NAMES + + +class TestKeyProfiles: + """Verify the Krumhansl-Schmuckler key profiles are well-formed.""" + + def test_major_profile_has_12_entries(self): + assert len(MAJOR_PROFILE) == 12 + + def test_minor_profile_has_12_entries(self): + assert len(MINOR_PROFILE) == 12 + + def test_key_names_has_24_entries(self): + assert len(KEY_NAMES) == 24 + # 12 major + 12 minor + majors = [k for k in KEY_NAMES if "major" in k] + minors = [k for k in KEY_NAMES if "minor" in k] + assert len(majors) == 12 + assert len(minors) == 12 + + +class TestDetectBpm: + """BPM detection tests.""" + + def test_detect_bpm_on_silent_file_raises(self, tmp_path): + """A file that's not valid audio should raise an error from librosa.""" + fake_audio = tmp_path / "silent.mp3" + fake_audio.write_bytes(b"\x00" * 1024) + with pytest.raises(Exception): + # librosa will fail on invalid audio data + detect_bpm(str(fake_audio)) + + @pytest.mark.skipif( + subprocess.run(["ffmpeg", "-version"], capture_output=True).returncode != 0, + reason="ffmpeg not available" + ) + def test_detect_bpm_on_generated_audio(self, tmp_path): + """Generate a simple 120 BPM click track and verify BPM detection.""" + mp3_path = tmp_path / "test_bpm.mp3" + # Generate a 5-second audio file with ffmpeg + # Using a sine wave at 440Hz — BPM detection should return a numeric value + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", "sine=frequency=440:duration=5", + "-q:a", "9", str(mp3_path) + ], + capture_output=True, check=True, + ) + bpm = detect_bpm(str(mp3_path)) + # BPM detection on a pure sine might not be perfectly accurate, + # but it should return a positive float + # BPM detection on a pure sine returns 0.0 (no beats) + # This is expected — a pure tone has no rhythmic content + assert isinstance(bpm, float) + assert bpm >= 0.0 + + +class TestDetectKey: + """Key detection tests.""" + + @pytest.mark.skipif( + subprocess.run(["ffmpeg", "-version"], capture_output=True).returncode != 0, + reason="ffmpeg not available" + ) + def test_detect_key_returns_valid_key_name(self, tmp_path): + mp3_path = tmp_path / "test_key.mp3" + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", "sine=frequency=440:duration=5", + "-q:a", "9", str(mp3_path) + ], + capture_output=True, check=True, + ) + key = detect_key(str(mp3_path)) + assert key in KEY_NAMES, f"Detected key '{key}' not in KEY_NAMES" + + +class TestAnalyzeClip: + """Full analysis integration test.""" + + def test_analyze_missing_file_raises(self, tmp_path): + with pytest.raises(FileNotFoundError): + analyze_clip(str(tmp_path / "nonexistent.mp3")) + + @pytest.mark.skipif( + subprocess.run(["ffmpeg", "-version"], capture_output=True).returncode != 0, + reason="ffmpeg not available" + ) + def test_analyze_clip_returns_bpm_and_key(self, tmp_path): + mp3_path = tmp_path / "test_analyze.mp3" + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", "sine=frequency=440:duration=5", + "-q:a", "9", str(mp3_path) + ], + capture_output=True, check=True, + ) + result = analyze_clip(str(mp3_path)) + assert "bpm" in result + assert "key" in result + assert isinstance(result["bpm"], float) + assert result["bpm"] >= 0.0 # Pure sine has no beats, BPM may be 0 + assert result["key"] in KEY_NAMES \ No newline at end of file diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py new file mode 100644 index 0000000..09de8be --- /dev/null +++ b/backend/tests/test_api.py @@ -0,0 +1,222 @@ +"""Tests for Lyria Studio API endpoints.""" + +import subprocess +import sys +import time +from pathlib import Path + +import pytest +import pytest_asyncio +import aiosqlite + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + + +async def seed_test_clip(app_client, prompt="test prompt", model="lyria-3-clip-preview", duration=30): + """Create a clip record in the test DB with a dummy MP3 file.""" + from database import create_clip + + clip_id = await create_clip(prompt, model, duration) + + filename = f"{clip_id}.mp3" + mp3_path = app_client["clips_dir"] / filename + mp3_path.write_bytes(b"\xff\xfb\x90\x00" + b"\x00" * 4096) + + async with aiosqlite.connect(app_client["db"]) as db: + await db.execute( + "UPDATE clips SET status = ?, filename = ? WHERE id = ?", + ("ready", filename, clip_id), + ) + await db.commit() + + return clip_id + + +def create_real_mp3(output_path, duration=2): + """Generate a real MP3 file using ffmpeg. Returns True if successful.""" + try: + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", f"anullsrc=r=22050:cl=mono", + "-t", str(duration), "-q:a", "9", + str(output_path) + ], + capture_output=True, check=True, + ) + return True + except (FileNotFoundError, subprocess.CalledProcessError): + return False + + +class TestHealthEndpoint: + """Basic health check.""" + + @pytest.mark.asyncio + async def test_health_returns_ok(self, app_client): + response = await app_client["client"].get("/api/health") + assert response.status_code == 200 + data = response.json() + assert data["status"] == "ok" + assert "version" in data + + +class TestModelsEndpoint: + """Model listing.""" + + @pytest.mark.asyncio + async def test_list_models(self, app_client): + response = await app_client["client"].get("/api/models") + assert response.status_code == 200 + data = response.json() + assert "models" in data + assert "default" in data + assert "lyria-3-clip-preview" in data["models"] + + +class TestClipsEndpoints: + """Clip CRUD operations.""" + + @pytest.mark.asyncio + async def test_list_clips_empty(self, app_client): + response = await app_client["client"].get("/api/clips") + assert response.status_code == 200 + data = response.json() + assert data["clips"] == [] + assert data["total"] == 0 + + @pytest.mark.asyncio + async def test_seed_and_list_clip(self, app_client): + clip_id = await seed_test_clip(app_client) + + response = await app_client["client"].get("/api/clips") + assert response.status_code == 200 + data = response.json() + assert data["total"] >= 1 + clip = data["clips"][0] + assert clip["id"] == clip_id + assert clip["status"] == "ready" + + @pytest.mark.asyncio + async def test_get_clip_by_id(self, app_client): + clip_id = await seed_test_clip(app_client, prompt="gabber kick at 180bpm") + + response = await app_client["client"].get(f"/api/clips/{clip_id}") + assert response.status_code == 200 + clip = response.json() + assert clip["id"] == clip_id + assert clip["prompt"] == "gabber kick at 180bpm" + + @pytest.mark.asyncio + async def test_get_nonexistent_clip(self, app_client): + response = await app_client["client"].get("/api/clips/deadbeef-0000-0000-0000-000000000000") + assert response.status_code == 404 + + @pytest.mark.asyncio + async def test_rename_clip(self, app_client): + clip_id = await seed_test_clip(app_client) + + response = await app_client["client"].patch( + f"/api/clips/{clip_id}", + json={"title": "New Title"}, + ) + assert response.status_code == 200 + clip = response.json() + assert clip["title"] == "New Title" + + @pytest.mark.asyncio + async def test_update_metadata(self, app_client): + clip_id = await seed_test_clip(app_client) + + response = await app_client["client"].patch( + f"/api/clips/{clip_id}/metadata", + json={"artist": "Kagamine Rin", "album": "HARDCORE WILL NEVER DIE", "comment": "180 BPM banger"}, + ) + assert response.status_code == 200 + clip = response.json() + assert clip["artist"] == "Kagamine Rin" + assert clip["album"] == "HARDCORE WILL NEVER DIE" + + @pytest.mark.asyncio + async def test_delete_clip(self, app_client): + clip_id = await seed_test_clip(app_client) + + response = await app_client["client"].delete(f"/api/clips/{clip_id}") + assert response.status_code == 200 + assert response.json()["deleted"] is True + + # Verify it's gone + response = await app_client["client"].get(f"/api/clips/{clip_id}") + assert response.status_code == 404 + + +class TestWaveformEndpoint: + """Waveform generation and caching.""" + + @pytest.mark.asyncio + async def test_waveform_returns_404_for_missing_clip(self, app_client): + response = await app_client["client"].get( + "/api/clips/deadbeef-0000-0000-0000-000000000000/waveform" + ) + assert response.status_code == 404 + + @pytest.mark.asyncio + async def test_waveform_returns_svg_for_existing_clip(self, app_client): + clip_id = await seed_test_clip(app_client) + mp3_path = app_client["clips_dir"] / f"{clip_id}.mp3" + + if not create_real_mp3(mp3_path, duration=2): + pytest.skip("ffmpeg not available on this system") + + response = await app_client["client"].get(f"/api/clips/{clip_id}/waveform") + assert response.status_code == 200 + assert response.headers["content-type"] == "image/svg+xml" + assert "Cache-Control" in response.headers + assert "max-age=86400" in response.headers["Cache-Control"] + + # Verify SVG content + content = response.text + assert "" in content + + @pytest.mark.asyncio + async def test_waveform_caches_on_second_request(self, app_client): + clip_id = await seed_test_clip(app_client) + mp3_path = app_client["clips_dir"] / f"{clip_id}.mp3" + + if not create_real_mp3(mp3_path, duration=2): + pytest.skip("ffmpeg not available on this system") + + # First request — may generate on demand + response1 = await app_client["client"].get(f"/api/clips/{clip_id}/waveform") + assert response1.status_code == 200 + + # Second request — should be served from disk cache + start = time.time() + response2 = await app_client["client"].get(f"/api/clips/{clip_id}/waveform") + elapsed = time.time() - start + + assert response2.status_code == 200 + assert response2.text == response1.text # Same SVG content + assert elapsed < 1.0, f"Cached waveform took {elapsed:.3f}s — too slow" + + +class TestClearWaveformsEndpoint: + """Admin waveform cache clearing.""" + + @pytest.mark.asyncio + async def test_clear_waveforms_returns_count(self, app_client): + waveforms_dir = app_client["waveforms_dir"] + + # Create a couple fake waveform files + (waveforms_dir / "test1.svg").write_text("1") + (waveforms_dir / "test2.svg").write_text("2") + + response = await app_client["client"].post("/api/admin/clear-waveforms") + assert response.status_code == 200 + data = response.json() + assert data["cleared"] == 2 + + # Verify they're gone + remaining = list(waveforms_dir.glob("*.svg")) + assert len(remaining) == 0 \ No newline at end of file diff --git a/backend/tests/test_models.py b/backend/tests/test_models.py new file mode 100644 index 0000000..f85763f --- /dev/null +++ b/backend/tests/test_models.py @@ -0,0 +1,90 @@ +"""Tests for Pydantic models.""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +from models import GenerateRequest, GenerateResponse, ClipList, LYRIA_MODELS, DEFAULT_MODEL + + +class TestGenerateRequest: + """Request validation and prompt building.""" + + def test_default_model_and_duration(self): + req = GenerateRequest(prompt="gabber kick") + assert req.model == DEFAULT_MODEL + dur = req.get_effective_duration() + assert dur == 30 # clip preview default + + def test_pro_model_with_custom_duration(self): + req = GenerateRequest(prompt="180 BPM hardcore", model="lyria-3-pro-preview", duration=120) + assert req.model == "lyria-3-pro-preview" + dur = req.get_effective_duration() + assert dur == 120 + + def test_pro_model_duration_clamped_to_max(self): + req = GenerateRequest(prompt="long track", model="lyria-3-pro-preview", duration=999) + dur = req.get_effective_duration() + assert dur == 184 # max for pro model + + def test_clip_model_duration_clamped_to_max(self): + req = GenerateRequest(prompt="short clip", model="lyria-3-clip-preview", duration=60) + dur = req.get_effective_duration() + assert dur == 30 # max for clip preview + + def test_invalid_model_raises(self): + with pytest.raises(ValueError, match="Unknown model"): + GenerateRequest(prompt="test", model="nonexistent-model") + + def test_simple_prompt(self): + req = GenerateRequest(prompt="gabber at 200 BPM") + assert req.build_prompt() == "gabber at 200 BPM" + + def test_structured_prompt(self): + req = GenerateRequest( + prompt="ignored", + prompt_structure={ + "style": "gabber", + "tempo": "180 BPM", + "mood": "aggressive", + } + ) + built = req.build_prompt() + assert "**Style:** gabber" in built + assert "**Tempo:** 180 BPM" in built + assert "**Mood:** aggressive" in built + + def test_model_whitespace_trimmed(self): + req = GenerateRequest(prompt="test", model=" lyria-3-clip-preview ") + assert req.model == "lyria-3-clip-preview" + + +class TestModelConstants: + """Validate the model registry.""" + + def test_models_have_required_keys(self): + for model_id, info in LYRIA_MODELS.items(): + assert "name" in info, f"{model_id} missing 'name'" + assert "max_duration" in info, f"{model_id} missing 'max_duration'" + assert "default_duration" in info, f"{model_id} missing 'default_duration'" + assert "description" in info, f"{model_id} missing 'description'" + + def test_default_model_exists_in_registry(self): + assert DEFAULT_MODEL in LYRIA_MODELS + + def test_max_duration_greater_than_default(self): + for model_id, info in LYRIA_MODELS.items(): + assert info["max_duration"] >= info["default_duration"], \ + f"{model_id}: max_duration < default_duration" + + +class TestGenerateResponse: + """Response model.""" + + def test_create_response(self): + resp = GenerateResponse(id="test-123", status="generating", message="Clip generation started.") + assert resp.id == "test-123" + assert resp.status == "generating" \ No newline at end of file diff --git a/backend/tests/test_waveform.py b/backend/tests/test_waveform.py new file mode 100644 index 0000000..9a11945 --- /dev/null +++ b/backend/tests/test_waveform.py @@ -0,0 +1,173 @@ +"""Tests for waveform generation module.""" + +import asyncio +import sys +from pathlib import Path +from unittest.mock import AsyncMock, patch, MagicMock + +import pytest +import pytest_asyncio +import numpy as np + +# Ensure src is importable +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +from waveform import generate_waveform_svg, get_waveform_path, DOWNSAMPLE_TARGET + + +class TestGetWaveformPath: + """Tests for get_waveform_path (sync path lookup).""" + + def test_returns_path_when_svg_exists(self, tmp_path): + from waveform import WAVEFORM_DIR + import waveform + + orig = waveform.WAVEFORM_DIR + try: + waveform.WAVEFORM_DIR = tmp_path + # Create a fake SVG + (tmp_path / "test-clip-id.svg").write_text("") + + result = get_waveform_path("test-clip-id") + assert result is not None + assert result.exists() + assert result.name == "test-clip-id.svg" + finally: + waveform.WAVEFORM_DIR = orig + + def test_returns_none_when_svg_absent(self, tmp_path): + import waveform + + orig = waveform.WAVEFORM_DIR + try: + waveform.WAVEFORM_DIR = tmp_path + result = get_waveform_path("nonexistent-id") + assert result is None + finally: + waveform.WAVEFORM_DIR = orig + + +class TestGenerateWaveformSvg: + """Tests for async waveform SVG generation.""" + + @pytest.mark.asyncio + async def test_raises_file_not_found_for_missing_audio(self, tmp_path): + import waveform + + orig_clips = waveform.CLIPS_DIR + orig_wave = waveform.WAVEFORM_DIR + try: + waveform.CLIPS_DIR = tmp_path / "clips" + waveform.CLIPS_DIR.mkdir(exist_ok=True) + waveform.WAVEFORM_DIR = tmp_path / "waveforms" + waveform.WAVEFORM_DIR.mkdir(exist_ok=True) + + with pytest.raises(FileNotFoundError, match="Audio file not found"): + await generate_waveform_svg("no-such-id", "no-such-file.mp3") + finally: + waveform.CLIPS_DIR = orig_clips + waveform.WAVEFORM_DIR = orig_wave + + @pytest.mark.asyncio + async def test_generates_svg_for_audio_file(self, tmp_path): + """Test that generate_waveform_svg produces a valid SVG from a real audio file. + + This requires ffmpeg to be available on the system, since the function + uses asyncio.create_subprocess_exec to call ffmpeg. + """ + import subprocess + + # Skip if ffmpeg is not available + try: + subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True) + except (FileNotFoundError, subprocess.CalledProcessError): + pytest.skip("ffmpeg not available on this system") + + import waveform + + clips_dir = tmp_path / "clips" + clips_dir.mkdir() + waveforms_dir = tmp_path / "waveforms" + waveforms_dir.mkdir() + + orig_clips = waveform.CLIPS_DIR + orig_wave = waveform.WAVEFORM_DIR + try: + waveform.CLIPS_DIR = clips_dir + waveform.WAVEFORM_DIR = waveforms_dir + + # Generate a 2-second silent WAV with ffmpeg, then convert to MP3 + raw_wav = clips_dir / "test-clip.wav" + mp3_path = clips_dir / "test-clip.mp3" + subprocess.run( + [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", "anullsrc=r=22050:cl=mono", + "-t", "2", "-q:a", "9", + str(mp3_path) + ], + capture_output=True, check=True, + ) + + result = await generate_waveform_svg("test-clip", "test-clip.mp3") + + assert result is not None + svg_path = Path(result) + assert svg_path.exists() + content = svg_path.read_text() + assert "" in content + # Should have rect elements (waveform bars) + assert " 0 + assert DOWNSAMPLE_TARGET <= 5000 # 5K samples max for waveform + assert DOWNSAMPLE_TARGET >= 500 # Need enough resolution for 200 bars + + def test_numpy_linspace_produces_correct_count(self): + """Verify the downsampling logic produces exactly DOWNSAMPLE_TARGET samples.""" + # Simulate a 120s track at 22050 Hz = ~2.65M samples + audio_data = np.random.randn(2_650_000).astype(np.float32) + + if len(audio_data) > DOWNSAMPLE_TARGET: + # Use len-1 as upper bound to match the fixed waveform.py + indices = np.linspace(0, len(audio_data) - 1, DOWNSAMPLE_TARGET, dtype=int) + downsampled = audio_data[indices] + + assert len(downsampled) == DOWNSAMPLE_TARGET \ No newline at end of file