Spaces:
Sleeping
Sleeping
| """API integration tests.""" | |
| import pytest | |
| import io | |
| import numpy as np | |
| import soundfile as sf | |
| from fastapi.testclient import TestClient | |
| from backend.main import app | |
| def client(): | |
| """Create test client.""" | |
| return TestClient(app) | |
| def make_wav_bytes(audio: np.ndarray, sr: int) -> bytes: | |
| """Helper: encode numpy array to WAV bytes.""" | |
| buf = io.BytesIO() | |
| sf.write(buf, audio, sr, format='WAV') | |
| buf.seek(0) | |
| return buf.read() | |
| def uploaded_session(client, sample_stems): | |
| """Upload stems and return session_id.""" | |
| stems, sr = sample_stems | |
| files = [] | |
| for name, audio in stems.items(): | |
| wav_bytes = make_wav_bytes(audio, sr) | |
| files.append(("files", (f"{name}.wav", wav_bytes, "audio/wav"))) | |
| response = client.post("/api/upload", files=files) | |
| assert response.status_code == 200 | |
| return response.json()["session_id"] | |
| def test_health_check(client): | |
| """Health endpoint should return healthy status.""" | |
| response = client.get("/api/health") | |
| assert response.status_code == 200 | |
| assert response.json()["status"] == "healthy" | |
| def test_upload_returns_session(client, sample_stems): | |
| """Upload should return a valid session with stem names.""" | |
| stems, sr = sample_stems | |
| files = [] | |
| for name, audio in stems.items(): | |
| wav_bytes = make_wav_bytes(audio, sr) | |
| files.append(("files", (f"{name}.wav", wav_bytes, "audio/wav"))) | |
| response = client.post("/api/upload", files=files) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "session_id" in data | |
| assert set(data["stems"]) == {"bass", "drums", "guitar"} | |
| def test_detection_returns_bpm_and_key(client, uploaded_session): | |
| """Detection should return BPM and key.""" | |
| response = client.post(f"/api/detect/{uploaded_session}") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "bpm" in data | |
| assert "key" in data | |
| assert "mode" in data | |
| assert data["bpm"] > 0 | |
| def test_process_pitch_shift(client, uploaded_session): | |
| """Processing with pitch shift should succeed.""" | |
| # First detect | |
| client.post(f"/api/detect/{uploaded_session}") | |
| # Then process | |
| response = client.post(f"/api/process/{uploaded_session}", json={ | |
| "semitones": 2, | |
| "target_bpm": None | |
| }) | |
| assert response.status_code == 200 | |
| assert response.json()["success"] is True | |
| def test_get_stem_after_processing(client, uploaded_session): | |
| """Should be able to fetch a processed stem as audio.""" | |
| client.post(f"/api/detect/{uploaded_session}") | |
| client.post(f"/api/process/{uploaded_session}", json={"semitones": 2}) | |
| response = client.get(f"/api/stem/{uploaded_session}/bass?processed=true") | |
| assert response.status_code == 200 | |
| assert response.headers["content-type"] in ["audio/wav", "audio/x-wav"] | |
| def test_invalid_session_404(client): | |
| """Requesting a nonexistent session should return 404.""" | |
| response = client.post("/api/detect/nonexistent-id") | |
| assert response.status_code == 404 | |
| def test_upload_no_files_422(client): | |
| """Uploading with no files should return 422.""" | |
| response = client.post("/api/upload", files=[]) | |
| assert response.status_code == 422 | |
| def test_upload_non_wav_400(client): | |
| """Uploading non-WAV files should return 400.""" | |
| files = [("files", ("test.txt", b"not audio", "text/plain"))] | |
| response = client.post("/api/upload", files=files) | |
| assert response.status_code == 400 | |
| def test_list_stems(client, uploaded_session): | |
| """Should be able to list all stems.""" | |
| response = client.get(f"/api/stems/{uploaded_session}") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "stems" in data | |
| assert len(data["stems"]) == 3 | |
| def test_get_original_stem(client, uploaded_session): | |
| """Should be able to fetch original stem without processing.""" | |
| response = client.get(f"/api/stem/{uploaded_session}/bass?processed=false") | |
| assert response.status_code == 200 | |
| def test_get_nonexistent_stem_404(client, uploaded_session): | |
| """Requesting nonexistent stem should return 404.""" | |
| response = client.get(f"/api/stem/{uploaded_session}/nonexistent") | |
| assert response.status_code == 404 | |
| def test_process_without_detection_400(client, uploaded_session): | |
| """Processing without detection should return 400.""" | |
| response = client.post(f"/api/process/{uploaded_session}", json={ | |
| "semitones": 2 | |
| }) | |
| assert response.status_code == 400 | |
| def test_upload_with_mix_file(client, sample_stems): | |
| """Upload with a mix file should recognize it.""" | |
| stems, sr = sample_stems | |
| files = [] | |
| for name, audio in stems.items(): | |
| wav_bytes = make_wav_bytes(audio, sr) | |
| files.append(("files", (f"{name}.wav", wav_bytes, "audio/wav"))) | |
| # Add a mix file | |
| mix_audio = np.sum([a for a in stems.values()], axis=0) / 3 | |
| mix_bytes = make_wav_bytes(mix_audio.astype(np.float32), sr) | |
| files.append(("files", ("full_mix.wav", mix_bytes, "audio/wav"))) | |
| response = client.post("/api/upload", files=files) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["has_full_mix"] is True | |