Initial Commit

packages/media/audio/audio-fingerprint/tests/test_api.py

@@ -0,0 +1,224 @@
+"""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