rin.kagamine
6c1e1b4509
- conftest.py: isolated async test fixtures with temp DB and directories - test_waveform.py: waveform generation, downsampling, caching, SVG output - test_api.py: full API endpoint tests (health, models, clips CRUD, waveform endpoint, admin cache clearing) - test_analysis.py: BPM detection, key detection, clip analysis - test_models.py: request validation, model registry, prompt building - Fixed off-by-one in waveform.py downsampling (linspace upper bound) - All 40 tests passing
222 lines
7.5 KiB
Python
222 lines
7.5 KiB
Python
"""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 "<svg" in content
|
|
assert "</svg>" 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("<svg>1</svg>")
|
|
(waveforms_dir / "test2.svg").write_text("<svg>2</svg>")
|
|
|
|
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 |