266 lines
10 KiB
Python
266 lines
10 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
import struct
|
|
import sqlite3
|
|
import sys
|
|
import threading
|
|
from types import SimpleNamespace
|
|
import uuid
|
|
import wave
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import service
|
|
|
|
|
|
def wav():
|
|
output = BytesIO()
|
|
with wave.open(output, "wb") as audio:
|
|
audio.setnchannels(1)
|
|
audio.setsampwidth(2)
|
|
audio.setframerate(16000)
|
|
audio.writeframes(struct.pack("<h", 100) * 1600)
|
|
return output.getvalue()
|
|
|
|
|
|
def upload(jobs, key=None, data=None, language="de"):
|
|
return jobs.upload(BytesIO(wav() if data is None else data), "call.wav",
|
|
key or "swyxweb:" + str(uuid.uuid4()), language, "meeting")
|
|
|
|
|
|
def test_job_and_audio_survive_restart_and_complete_without_summary(tmp_path):
|
|
calls = []
|
|
|
|
def transcribe(path, language):
|
|
calls.append((path.read_bytes(), language))
|
|
return "Bitte morgen zurückrufen."
|
|
|
|
jobs = service.Jobs(tmp_path, transcribe)
|
|
job = upload(jobs)
|
|
assert job["status"] == "queued"
|
|
# Simulate interruption after acknowledgement and during inference.
|
|
with jobs.db() as db:
|
|
db.execute("UPDATE jobs SET status='transcribing' WHERE id=?", (job["id"],))
|
|
restarted = service.Jobs(tmp_path, transcribe)
|
|
assert restarted.get(job["id"])["status"] == "queued"
|
|
assert restarted.process_next()
|
|
result = restarted.get(job["id"])
|
|
assert result["status"] == "done"
|
|
assert result["transcript"] == "Bitte morgen zurückrufen."
|
|
assert result["summary"] == ""
|
|
assert calls == [(wav(), "de")]
|
|
assert not restarted.process_next()
|
|
assert service.Jobs(tmp_path, transcribe).get(job["id"]) == result
|
|
|
|
|
|
def test_concurrent_duplicate_uploads_create_one_durable_job(tmp_path):
|
|
jobs = service.Jobs(tmp_path, lambda *_: "Text")
|
|
key = "swyxweb:" + str(uuid.uuid4())
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(lambda _: upload(jobs, key), range(8)))
|
|
assert len({job["id"] for job in results}) == 1
|
|
assert len(list(tmp_path.glob("*.wav"))) == 1
|
|
with pytest.raises(HTTPException) as conflict:
|
|
upload(jobs, key, language="en")
|
|
assert conflict.value.status_code == 409
|
|
with pytest.raises(HTTPException) as conflict:
|
|
upload(jobs, key, data=wav() + b"changed")
|
|
assert conflict.value.status_code == 409
|
|
assert len(list(tmp_path.glob("*.wav"))) == 1
|
|
|
|
|
|
def test_failed_job_can_be_retried_without_reuploading(tmp_path):
|
|
def failure(*_):
|
|
raise ValueError("Unreadable audio")
|
|
|
|
jobs = service.Jobs(tmp_path, failure)
|
|
job = upload(jobs)
|
|
assert jobs.process_next()
|
|
assert jobs.get(job["id"])["status"] == "failed"
|
|
assert jobs.get(job["id"])["error"]
|
|
jobs.transcribe = lambda *_: "Erneut transkribiert."
|
|
assert jobs.retry(job["id"])["status"] == "queued"
|
|
with pytest.raises(HTTPException) as conflict:
|
|
jobs.retry(job["id"])
|
|
assert conflict.value.status_code == 409
|
|
jobs.process_next()
|
|
assert jobs.get(job["id"])["transcript"] == "Erneut transkribiert."
|
|
assert jobs.get(job["id"])["error"] == ""
|
|
|
|
|
|
@pytest.mark.parametrize("data", [b"", b"not a WAV recording"])
|
|
def test_invalid_upload_does_not_leave_files_or_jobs(tmp_path, data):
|
|
jobs = service.Jobs(tmp_path, lambda *_: "")
|
|
with pytest.raises(HTTPException) as error:
|
|
upload(jobs, data=data)
|
|
assert error.value.status_code == 400
|
|
assert not list(tmp_path.glob("*.wav"))
|
|
assert not jobs.process_next()
|
|
|
|
|
|
def test_size_limit_cleans_up_partial_file(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(service, "MAX_BYTES", 100)
|
|
jobs = service.Jobs(tmp_path, lambda *_: "")
|
|
with pytest.raises(HTTPException) as error:
|
|
upload(jobs)
|
|
assert error.value.status_code == 413
|
|
assert not list(tmp_path.glob("*.wav"))
|
|
|
|
|
|
def test_http_contract_remains_responsive_during_transcription(tmp_path):
|
|
started, release = threading.Event(), threading.Event()
|
|
|
|
def transcribe(*_):
|
|
started.set()
|
|
assert release.wait(5)
|
|
return "Lokales Transkript"
|
|
|
|
with TestClient(service.create_app(tmp_path, transcribe), base_url="http://127.0.0.1") as client:
|
|
try:
|
|
response = client.post("/api/jobs", files={"file": ("call.wav", wav(), "audio/wav")},
|
|
data={"language": "de", "style": "meeting"}, headers={"Idempotency-Key": "swyxweb:" + str(uuid.uuid4())})
|
|
assert response.status_code == 202
|
|
job = response.json()
|
|
assert set(job) == set(service.PUBLIC_FIELDS)
|
|
assert started.wait(5)
|
|
assert client.get("/health").status_code == 200
|
|
assert client.get("/api/jobs/" + job["id"]).json()["status"] == "transcribing"
|
|
assert client.post("/api/jobs/" + job["id"] + "/retry").status_code == 409
|
|
assert client.get("/api/jobs/" + "0" * 32).status_code == 404
|
|
assert client.get("/api/jobs/invalid").status_code == 400
|
|
assert client.get("/health", headers={"Host": "untrusted.example"}).status_code == 400
|
|
finally:
|
|
release.set()
|
|
|
|
|
|
def test_whisper_runs_cpu_offline_and_consumes_all_segments(monkeypatch):
|
|
consumed = []
|
|
|
|
class Model:
|
|
def __init__(self, path, **options):
|
|
assert path == "/models/test"
|
|
assert options == {"device": "cpu", "compute_type": "int8", "local_files_only": True,
|
|
"cpu_threads": 2, "num_workers": 1}
|
|
|
|
def transcribe(self, path, **options):
|
|
assert path == "call.wav"
|
|
assert options["language"] is None
|
|
assert options["task"] == "transcribe"
|
|
|
|
def segments():
|
|
for text in [" Guten Tag.", " Bitte zurückrufen."]:
|
|
consumed.append(text)
|
|
yield SimpleNamespace(text=text)
|
|
|
|
return segments(), SimpleNamespace(language="de")
|
|
|
|
monkeypatch.setenv("WHISPER_MODEL_PATH", "/models/test")
|
|
monkeypatch.setenv("WHISPER_THREADS", "2")
|
|
monkeypatch.setitem(sys.modules, "faster_whisper", SimpleNamespace(WhisperModel=Model))
|
|
assert service.load_whisper()(Path("call.wav"), "auto") == "Guten Tag. Bitte zurückrufen."
|
|
assert len(consumed) == 2
|
|
|
|
|
|
def test_real_audio_decoder_is_compatible_with_installed_pyav():
|
|
audio = pytest.importorskip("faster_whisper.audio")
|
|
decoded = audio.decode_audio(BytesIO(wav()), sampling_rate=16000)
|
|
assert decoded.shape == (1600,)
|
|
assert decoded[0] == pytest.approx(100 / 32768)
|
|
|
|
|
|
def test_summary_failure_preserves_transcript_and_retry_skips_whisper(tmp_path):
|
|
calls = []
|
|
|
|
def transcribe(*_):
|
|
calls.append("whisper")
|
|
return "Rückruf am Freitag."
|
|
|
|
def fail_summary(text, style):
|
|
assert text == "Rückruf am Freitag."
|
|
assert style == "meeting"
|
|
raise service.SummaryError("LLM vorübergehend nicht erreichbar.")
|
|
|
|
jobs = service.Jobs(tmp_path, transcribe, fail_summary)
|
|
job = upload(jobs)
|
|
jobs.process_next()
|
|
result = jobs.get(job["id"])
|
|
assert result["status"] == "failed"
|
|
assert result["stage"] == "Zusammenfassung fehlgeschlagen"
|
|
assert result["transcript"] == "Rückruf am Freitag."
|
|
assert "LLM" in result["error"]
|
|
jobs.summarize = lambda text, style: "Am Freitag zurückrufen."
|
|
jobs.retry(job["id"])
|
|
jobs.process_next()
|
|
result = jobs.get(job["id"])
|
|
assert result["status"] == "done"
|
|
assert result["summary"] == "Am Freitag zurückrufen."
|
|
assert result["error"] == ""
|
|
assert calls == ["whisper"]
|
|
|
|
|
|
def test_transcript_is_available_while_summary_is_still_running(tmp_path):
|
|
started, release = threading.Event(), threading.Event()
|
|
|
|
def summarize(*_):
|
|
started.set()
|
|
assert release.wait(5)
|
|
return "Zusammenfassung"
|
|
|
|
with TestClient(service.create_app(tmp_path, lambda *_: "Gesprächstext", summarize), base_url="http://127.0.0.1") as client:
|
|
try:
|
|
response = client.post("/api/jobs", files={"file": ("call.wav", wav(), "audio/wav")},
|
|
headers={"Idempotency-Key": "swyxweb:" + str(uuid.uuid4())})
|
|
assert response.status_code == 202
|
|
assert started.wait(5)
|
|
job = client.get("/api/jobs/" + response.json()["id"]).json()
|
|
assert job["status"] == "summarizing"
|
|
assert job["transcript"] == "Gesprächstext"
|
|
assert job["summary"] == ""
|
|
assert client.get("/health").status_code == 200
|
|
finally:
|
|
release.set()
|
|
|
|
|
|
def test_restart_resumes_summary_without_retranscribing(tmp_path):
|
|
jobs = service.Jobs(tmp_path, lambda *_: "Gesprächstext")
|
|
job = upload(jobs)
|
|
jobs.process_next()
|
|
with jobs.db() as db:
|
|
db.execute("UPDATE jobs SET status='summarizing' WHERE id=?", (job["id"],))
|
|
|
|
def unexpected_transcription(*_):
|
|
pytest.fail("An existing transcript must be reused")
|
|
|
|
restarted = service.Jobs(tmp_path, unexpected_transcription, lambda text, _: "Ergebnis: " + text)
|
|
assert restarted.get(job["id"])["status"] == "queued"
|
|
restarted.process_next()
|
|
assert restarted.get(job["id"])["summary"] == "Ergebnis: Gesprächstext"
|
|
|
|
|
|
def test_silent_recording_does_not_call_llm(tmp_path):
|
|
def unexpected_summary(*_):
|
|
pytest.fail("Silence must not be sent to the LLM")
|
|
|
|
jobs = service.Jobs(tmp_path, lambda *_: "", unexpected_summary)
|
|
job = upload(jobs)
|
|
jobs.process_next()
|
|
assert jobs.get(job["id"])["status"] == "done"
|
|
assert jobs.get(job["id"])["stage"] == "Keine Sprache erkannt"
|
|
|
|
|
|
def test_existing_database_is_migrated_without_losing_results(tmp_path):
|
|
jobs = service.Jobs(tmp_path, lambda *_: "Vorhandenes Transkript")
|
|
job = upload(jobs)
|
|
jobs.process_next()
|
|
with sqlite3.connect(tmp_path / "jobs.sqlite3") as db:
|
|
db.execute("ALTER TABLE jobs DROP COLUMN transcribed")
|
|
migrated = service.Jobs(tmp_path, lambda *_: "")
|
|
assert migrated.get(job["id"])["transcript"] == "Vorhandenes Transkript"
|
|
assert migrated.get(job["id"])["status"] == "done"
|
|
with migrated.db() as db:
|
|
assert db.execute("SELECT transcribed FROM jobs WHERE id=?", (job["id"],)).fetchone()[0] == 1
|