307 lines
10 KiB
Python
307 lines
10 KiB
Python
import io
|
|
import json
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
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 __init__(self, segments=None):
|
|
self.segments = segments or []
|
|
|
|
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="", transcriber=None):
|
|
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=transcriber or FakeTranscriber(),
|
|
)
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "tester", "password": "password-123"},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
client.headers.update({"Authorization": f"Bearer {response.json()['token']}"})
|
|
return client
|
|
|
|
|
|
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_get_dub_share_includes_video_hash(tmp_path):
|
|
client = make_client(tmp_path)
|
|
audio = ("dub.wav", make_wav(), "audio/wav")
|
|
|
|
created = client.post(
|
|
"/api/v1/dub-shares",
|
|
data={
|
|
"video_hash": VIDEO_HASH,
|
|
"title": "Shared lesson",
|
|
"segments": json.dumps([{"sentence_index": 0}]),
|
|
},
|
|
files={"files": audio},
|
|
)
|
|
response = client.get(f"/api/v1/dub-shares/{created.json()['share_id']}")
|
|
|
|
assert created.status_code == 201, created.text
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["video_hash"] == VIDEO_HASH
|
|
|
|
|
|
def test_create_dub_share_keeps_original_audio_without_alignment(tmp_path):
|
|
"""Share audio should be stored as uploaded, without tempo/align processing."""
|
|
client = make_client(tmp_path)
|
|
payload = make_wav()
|
|
response = client.post(
|
|
"/api/v1/dub-shares",
|
|
data={
|
|
"video_hash": VIDEO_HASH,
|
|
"segments": json.dumps([{"sentence_index": 0}]),
|
|
},
|
|
files={"files": ("dub.wav", payload, "audio/wav")},
|
|
)
|
|
|
|
assert response.status_code == 201, response.text
|
|
share_id = response.json()["share_id"]
|
|
detail = client.get(f"/api/v1/dub-shares/{share_id}")
|
|
assert detail.status_code == 200
|
|
audio_url = detail.json()["segments"][0]["audio_url"]
|
|
audio = client.get(audio_url)
|
|
assert audio.status_code == 200
|
|
assert audio.content == payload
|
|
|
|
|
|
def test_assessment_requires_a_valid_user_session(tmp_path):
|
|
client = make_client(tmp_path, client_api_key="tablet-key")
|
|
endpoint = f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments"
|
|
client.headers.clear()
|
|
|
|
unauthorized = client.post(
|
|
endpoint,
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
login = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "tester", "password": "password-123"},
|
|
)
|
|
assert login.status_code == 200
|
|
authorized = client.post(
|
|
endpoint,
|
|
headers={"Authorization": f"Bearer {login.json()['token']}"},
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
|
|
assert unauthorized.status_code == 401
|
|
assert authorized.status_code == 200
|
|
|
|
|
|
def test_user_enrollment_results_and_dub_shares_are_scoped(tmp_path):
|
|
client = make_client(tmp_path)
|
|
|
|
assert client.get("/api/v1/courses").status_code == 200
|
|
assert client.post(f"/api/v1/courses/{VIDEO_HASH}/enroll").status_code == 204
|
|
courses = client.get("/api/v1/courses").json()
|
|
assert courses[0]["enrolled"] is True
|
|
|
|
assessed = client.post(
|
|
f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments",
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
data={"language": "en"},
|
|
)
|
|
assert assessed.status_code == 200
|
|
results = client.get("/api/v1/me/results").json()
|
|
assert len(results) == 1
|
|
assert results[0]["course_title"] == "Lesson"
|
|
|
|
shared = client.post(
|
|
"/api/v1/dub-shares",
|
|
data={
|
|
"video_hash": VIDEO_HASH,
|
|
"segments": json.dumps([{"sentence_index": 0}]),
|
|
"scores": json.dumps([{"sentence_index": 0, "overall_score": 88}]),
|
|
},
|
|
files={"files": ("dub.wav", make_wav(), "audio/wav")},
|
|
)
|
|
assert shared.status_code == 201, shared.text
|
|
shares = client.get("/api/v1/me/dub-shares").json()
|
|
assert len(shares) == 1
|
|
assert shares[0]["segment_count"] == 1
|
|
assert shares[0]["average_score"] == 88
|
|
|
|
|
|
def test_user_can_delete_owned_result_and_dub_share(tmp_path):
|
|
client = make_client(tmp_path)
|
|
assessed = client.post(
|
|
f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments",
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
attempt_id = client.get("/api/v1/me/results").json()[0]["attempt_id"]
|
|
shared = client.post(
|
|
"/api/v1/dub-shares",
|
|
data={
|
|
"video_hash": VIDEO_HASH,
|
|
"segments": json.dumps([{"sentence_index": 0}]),
|
|
},
|
|
files={"files": ("dub.wav", make_wav(), "audio/wav")},
|
|
)
|
|
share_id = shared.json()["share_id"]
|
|
|
|
deleted_result = client.delete(f"/api/v1/me/results/{attempt_id}")
|
|
deleted_share = client.delete(f"/api/v1/me/dub-shares/{share_id}")
|
|
|
|
assert assessed.status_code == 200
|
|
assert shared.status_code == 201
|
|
assert deleted_result.status_code == 204
|
|
assert deleted_share.status_code == 204
|
|
assert client.get("/api/v1/me/results").json() == []
|
|
assert client.get("/api/v1/me/dub-shares").json() == []
|
|
assert client.get(f"/api/v1/dub-shares/{share_id}").status_code == 404
|
|
|
|
|
|
def test_user_cannot_delete_another_users_history(tmp_path):
|
|
client = make_client(tmp_path)
|
|
client.post(
|
|
f"/api/v1/videos/{VIDEO_HASH}/sentences/0/assessments",
|
|
files={"audio": ("reading.wav", make_wav(), "audio/wav")},
|
|
)
|
|
attempt_id = client.get("/api/v1/me/results").json()[0]["attempt_id"]
|
|
shared = client.post(
|
|
"/api/v1/dub-shares",
|
|
data={
|
|
"video_hash": VIDEO_HASH,
|
|
"segments": json.dumps([{"sentence_index": 0}]),
|
|
},
|
|
files={"files": ("dub.wav", make_wav(), "audio/wav")},
|
|
)
|
|
share_id = shared.json()["share_id"]
|
|
|
|
other = TestClient(create_app(
|
|
BoundaryStore(tmp_path / "legacy.json"),
|
|
settings=replace(
|
|
Settings.from_env(),
|
|
data_dir=tmp_path / "data",
|
|
legacy_boundaries_file=tmp_path / "legacy.json",
|
|
),
|
|
))
|
|
registered = other.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "other", "password": "password-123"},
|
|
)
|
|
other.headers.update({"Authorization": f"Bearer {registered.json()['token']}"})
|
|
|
|
assert registered.status_code == 201
|
|
assert other.delete(f"/api/v1/me/results/{attempt_id}").status_code == 404
|
|
assert other.delete(f"/api/v1/me/dub-shares/{share_id}").status_code == 404
|