147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
import io
|
|
import json
|
|
import wave
|
|
from dataclasses import replace
|
|
|
|
import numpy as np
|
|
from fastapi.testclient import TestClient
|
|
|
|
from sentence_api.config import Settings
|
|
from sentence_api.main import create_app
|
|
from sentence_api.models import SentenceBoundaryDocument
|
|
from sentence_api.repository import VideoRepository
|
|
from sentence_api.store import BoundaryStore
|
|
from sentence_api.transcription import Transcript
|
|
|
|
|
|
VIDEO_HASH = "c" * 64
|
|
|
|
|
|
class FakeTranscriber:
|
|
available = True
|
|
|
|
def transcribe(self, audio_path, language=None):
|
|
return Transcript(text="The meeting starts at nine.", segments=[])
|
|
|
|
|
|
def make_wav() -> bytes:
|
|
sample_rate = 16_000
|
|
time = np.arange(sample_rate * 2, dtype=np.float64) / sample_rate
|
|
samples = (np.sin(2 * np.pi * 220 * time) * 8000).astype("<i2")
|
|
output = io.BytesIO()
|
|
with wave.open(output, "wb") as wav_file:
|
|
wav_file.setnchannels(1)
|
|
wav_file.setsampwidth(2)
|
|
wav_file.setframerate(sample_rate)
|
|
wav_file.writeframes(samples.tobytes())
|
|
return output.getvalue()
|
|
|
|
|
|
def make_client(tmp_path, client_api_key=""):
|
|
legacy_path = tmp_path / "legacy.json"
|
|
legacy_path.write_text(json.dumps({"videos": {}}), encoding="utf-8")
|
|
settings = replace(
|
|
Settings.from_env(),
|
|
data_dir=tmp_path / "data",
|
|
legacy_boundaries_file=legacy_path,
|
|
admin_api_key="test-admin-key",
|
|
client_api_key=client_api_key,
|
|
)
|
|
settings.ensure_directories()
|
|
repository = VideoRepository(settings.database_path)
|
|
stored_filename = f"{VIDEO_HASH}.mp4"
|
|
(settings.videos_dir / stored_filename).write_bytes(b"video-placeholder")
|
|
repository.upsert_upload(
|
|
video_hash=VIDEO_HASH,
|
|
title="Lesson",
|
|
filename="lesson.mp4",
|
|
stored_filename=stored_filename,
|
|
content_type="video/mp4",
|
|
size_bytes=17,
|
|
language="en",
|
|
)
|
|
repository.save_processing_result(
|
|
SentenceBoundaryDocument(
|
|
video_hash=VIDEO_HASH,
|
|
duration_ms=2000,
|
|
algorithm_version="test-v1",
|
|
sentences=[
|
|
{
|
|
"index": 0,
|
|
"start_ms": 0,
|
|
"end_ms": 2000,
|
|
"text": "The meeting starts at nine.",
|
|
"language": "en",
|
|
"reference_speech_duration_ms": 2000,
|
|
}
|
|
],
|
|
),
|
|
"The meeting starts at nine.",
|
|
)
|
|
app = create_app(
|
|
BoundaryStore(legacy_path),
|
|
settings=settings,
|
|
repository=repository,
|
|
transcriber=FakeTranscriber(),
|
|
)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_assessment_returns_duration_and_content_breakdown(tmp_path):
|
|
client = make_client(tmp_path)
|
|
|
|
response = client.post(
|
|
f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments",
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
data={"language": "en"},
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
assert payload["overall_score"] == 100
|
|
assert payload["passed"] is True
|
|
assert 0.99 <= payload["duration_ratio"] <= 1.02
|
|
assert payload["pronunciation_score"] is None
|
|
assert payload["details"]["phoneme_scoring"] == "not_enabled"
|
|
|
|
|
|
def test_admin_routes_require_key(tmp_path):
|
|
client = make_client(tmp_path)
|
|
|
|
response = client.delete(f"/api/v1/admin/videos/{VIDEO_HASH}")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_catalog_and_range_streaming(tmp_path):
|
|
client = make_client(tmp_path)
|
|
|
|
catalog = client.get("/api/v1/videos")
|
|
stream = client.get(
|
|
f"/api/v1/videos/{VIDEO_HASH}/content",
|
|
headers={"Range": "bytes=0-4"},
|
|
)
|
|
|
|
assert catalog.status_code == 200
|
|
assert catalog.json()["videos"][0]["stream_url"].endswith(f"/{VIDEO_HASH}/content")
|
|
assert stream.status_code == 206
|
|
assert stream.content == b"video"
|
|
|
|
|
|
def test_assessment_client_key_is_enforced_when_configured(tmp_path):
|
|
client = make_client(tmp_path, client_api_key="tablet-key")
|
|
endpoint = f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments"
|
|
|
|
unauthorized = client.post(
|
|
endpoint,
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
authorized = client.post(
|
|
endpoint,
|
|
headers={"X-Client-Key": "tablet-key"},
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
|
|
assert unauthorized.status_code == 401
|
|
assert authorized.status_code == 200
|