Spaces:
Sleeping
Sleeping
File size: 3,938 Bytes
75ba57a dd271a6 75ba57a dd271a6 75ba57a dd271a6 75ba57a dd271a6 75ba57a dd271a6 69923c9 dd271a6 75ba57a dd271a6 75ba57a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """
LectureLens — Tests: API Integration
Run with: pytest tests/test_api.py -v
Requires all dependencies installed (no Docker).
"""
from __future__ import annotations
import io
import time
import wave
import numpy as np
import pytest
from fastapi.testclient import TestClient
from app.main import app
# ── Fixtures ───────────────────────────────────────────────────────────────────
@pytest.fixture(scope="module")
def client():
with TestClient(app) as c:
yield c
@pytest.fixture(scope="module")
def wav_bytes() -> bytes:
"""Minimal 2-second sine WAV in memory."""
sr = 16_000
duration = 2
t = np.linspace(0, duration, sr * duration, endpoint=False)
samples = (np.sin(2 * np.pi * 440 * t) * 0.3 * 32767).astype(np.int16)
buf = io.BytesIO()
with wave.open(buf, "w") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(sr)
wf.writeframes(samples.tobytes())
return buf.getvalue()
# ── Health ─────────────────────────────────────────────────────────────────────
def test_health(client):
r = client.get("/health")
assert r.status_code == 200
assert r.json()["status"] == "ok"
# ── Polling Analysis ───────────────────────────────────────────────────────────
def test_audio_analysis_polling(client, wav_bytes):
# 1. Submit the job
r = client.post(
"/analyze",
data={"media_type": "audio", "participant_label": "test_speaker"},
files={"file": ("test.wav", wav_bytes, "audio/wav")},
)
assert r.status_code == 200, r.text
submit_body = r.json()
assert submit_body["status"] == "processing"
assert "job_id" in submit_body
job_id = submit_body["job_id"]
# 2. Poll for the result
for _ in range(10):
r_poll = client.get(f"/analyze/{job_id}")
assert r_poll.status_code == 200
poll_body = r_poll.json()
# TestClient actually runs BackgroundTasks synchronously at the end of the request.
# So by the time we poll, it's usually already 'completed'.
if poll_body["status"] == "completed":
res = poll_body["result"]
assert res["media_type"] == "audio"
assert res["participant_label"] == "test_speaker"
assert "audio_metrics" in res
assert res["audio_metrics"] is not None
assert res["video_metrics"] is None
assert "alerts" in res
assert res["overall_audio_score"] is not None
break
elif poll_body["status"] == "failed":
pytest.fail(f"Job failed: {poll_body['error_detail']}")
time.sleep(0.5)
else:
pytest.fail("Job did not complete within timeout")
# ── Validation ─────────────────────────────────────────────────────────────────
def test_wrong_extension_returns_400(client, wav_bytes):
r = client.post(
"/analyze",
data={"media_type": "video"}, # declare video but send .wav
files={"file": ("test.wav", wav_bytes, "audio/wav")},
)
assert r.status_code == 400
assert r.json()["error_code"] == "INVALID_MEDIA_TYPE"
def test_job_not_found(client):
r = client.get("/analyze/non-existent-job-id")
assert r.status_code == 404
assert r.json()["error_code"] == "NOT_FOUND"
def test_docs_accessible(client):
r = client.get("/docs")
assert r.status_code == 200
|