225 lines
6.9 KiB
Python
225 lines
6.9 KiB
Python
"""Tests for the FastAPI audio-fingerprinting service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import soundfile as sf
|
|
from fastapi.testclient import TestClient
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from src.api.app import app
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> TestClient:
|
|
return TestClient(app)
|
|
|
|
|
|
def _make_bad_audio_file() -> Path:
|
|
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False, mode="wb") as tmp:
|
|
tmp.write(b"not an audio file at all")
|
|
return Path(tmp.name)
|
|
|
|
|
|
def _make_different_audio() -> Path:
|
|
sample_rate = 22050
|
|
duration = 2.0
|
|
t = np.linspace(0, duration, int(sample_rate * duration))
|
|
audio = 0.5 * np.sin(2 * np.pi * 880.0 * t)
|
|
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
|
sf.write(tmp.name, audio.astype(np.float32), sample_rate)
|
|
return Path(tmp.name)
|
|
|
|
|
|
def test_root_endpoint(client: TestClient) -> None:
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["service"] == "audio-fingerprinting"
|
|
|
|
|
|
def test_fingerprint_endpoint(client: TestClient, audio_file: Path) -> None:
|
|
with open(audio_file, "rb") as f:
|
|
response = client.post(
|
|
"/fingerprint", files={"file": ("test.wav", f, "audio/wav")}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["filename"] == "test.wav"
|
|
assert data["duration"] > 0.0
|
|
assert data["peaks_count"] > 0
|
|
assert data["fingerprints_count"] > 0
|
|
assert isinstance(data["fingerprints"], list)
|
|
assert len(data["fingerprints"]) > 0
|
|
for fp in data["fingerprints"]:
|
|
assert "hash" in fp
|
|
assert "time" in fp
|
|
assert "delta" in fp
|
|
|
|
|
|
def test_fingerprint_endpoint_with_parameters(
|
|
client: TestClient, audio_file: Path
|
|
) -> None:
|
|
with open(audio_file, "rb") as f:
|
|
response = client.post(
|
|
"/fingerprint",
|
|
files={"file": ("test.wav", f, "audio/wav")},
|
|
data={"peak_threshold": 0.6, "fan_value": 20},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["fingerprints_count"] > 0
|
|
|
|
|
|
def test_fingerprint_endpoint_no_file(client: TestClient) -> None:
|
|
response = client.post("/fingerprint")
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_fingerprint_endpoint_invalid_audio(client: TestClient) -> None:
|
|
bad = _make_bad_audio_file()
|
|
try:
|
|
with open(bad, "rb") as f:
|
|
response = client.post(
|
|
"/fingerprint",
|
|
files={"file": ("bad.wav", f, "audio/wav")},
|
|
)
|
|
assert response.status_code == 500
|
|
assert response.json()["detail"].startswith("Failed to process audio file:")
|
|
finally:
|
|
bad.unlink(missing_ok=True)
|
|
|
|
|
|
def test_match_endpoint(client: TestClient, audio_file: Path) -> None:
|
|
with open(audio_file, "rb") as q, open(audio_file, "rb") as r:
|
|
response = client.post(
|
|
"/match",
|
|
files={
|
|
"query_file": ("q.wav", q, "audio/wav"),
|
|
"reference_file": ("r.wav", r, "audio/wav"),
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for key in (
|
|
"query_file",
|
|
"reference_file",
|
|
"query_duration",
|
|
"reference_duration",
|
|
"matched",
|
|
"match_count",
|
|
"confidence",
|
|
"time_offset",
|
|
):
|
|
assert key in data
|
|
assert data["matched"] is True
|
|
assert data["confidence"] > 0.9
|
|
assert abs(data["time_offset"]) <= 5
|
|
|
|
|
|
def test_match_endpoint_with_parameters(
|
|
client: TestClient, audio_file: Path
|
|
) -> None:
|
|
with open(audio_file, "rb") as q, open(audio_file, "rb") as r:
|
|
response = client.post(
|
|
"/match",
|
|
files={
|
|
"query_file": ("q.wav", q, "audio/wav"),
|
|
"reference_file": ("r.wav", r, "audio/wav"),
|
|
},
|
|
data={"peak_threshold": 0.6, "fan_value": 20, "min_matches": 5},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["matched"] is True
|
|
|
|
|
|
def test_match_endpoint_different_files(
|
|
client: TestClient, audio_file: Path
|
|
) -> None:
|
|
different = _make_different_audio()
|
|
try:
|
|
with open(audio_file, "rb") as q, open(different, "rb") as r:
|
|
response = client.post(
|
|
"/match",
|
|
files={
|
|
"query_file": ("q.wav", q, "audio/wav"),
|
|
"reference_file": ("r.wav", r, "audio/wav"),
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "confidence" in data
|
|
assert "matched" in data
|
|
finally:
|
|
different.unlink(missing_ok=True)
|
|
|
|
|
|
def test_match_endpoint_missing_files(client: TestClient) -> None:
|
|
response = client.post("/match")
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_match_endpoint_invalid_audio(
|
|
client: TestClient, audio_file: Path
|
|
) -> None:
|
|
bad = _make_bad_audio_file()
|
|
try:
|
|
with open(audio_file, "rb") as q, open(bad, "rb") as r:
|
|
response = client.post(
|
|
"/match",
|
|
files={
|
|
"query_file": ("q.wav", q, "audio/wav"),
|
|
"reference_file": ("bad.wav", r, "audio/wav"),
|
|
},
|
|
)
|
|
assert response.status_code == 500
|
|
assert response.json()["detail"].startswith("Failed to match audio files:")
|
|
finally:
|
|
bad.unlink(missing_ok=True)
|
|
|
|
|
|
def test_cors_headers(client: TestClient) -> None:
|
|
response = client.options(
|
|
"/fingerprint",
|
|
headers={
|
|
"Origin": "http://localhost:3000",
|
|
"Access-Control-Request-Method": "POST",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "access-control-allow-origin" in response.headers
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fingerprint_endpoint_async(audio_file: Path) -> None:
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
with open(audio_file, "rb") as f:
|
|
response = await ac.post(
|
|
"/fingerprint", files={"file": ("test.wav", f, "audio/wav")}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["fingerprints_count"] > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_match_endpoint_async(audio_file: Path) -> None:
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
with open(audio_file, "rb") as q, open(audio_file, "rb") as r:
|
|
response = await ac.post(
|
|
"/match",
|
|
files={
|
|
"query_file": ("q.wav", q, "audio/wav"),
|
|
"reference_file": ("r.wav", r, "audio/wav"),
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["matched"] is True
|