514d75fe92
A non-AIError from transcribe/summarize (av.InvalidDataError on corrupt audio, seen live with 'My recording 63') escaped the consumer, left the row 'running', and got requeued at every engine restart forever. Both runners now catch it, fail the row with the exception type in the message, and keep a usable transcript (summarize completes with a 'Summary failed' note). Regression tests pin both paths (proven red without the fix).
507 lines
21 KiB
Python
507 lines
21 KiB
Python
"""AI pipeline tests (M7): status flow, versioning, failure modes, endpoints.
|
|
|
|
Provider fakes stand in for real STT/LLM services (no network, no model
|
|
downloads); adapter wire-protocol tests live in test_ai_adapters.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
|
|
from shonar.db import session as db_session
|
|
from shonar.db.models import (
|
|
JobStatus,
|
|
JobType,
|
|
ProcessingJob,
|
|
Transcript,
|
|
)
|
|
from shonar.services import processing
|
|
from shonar.services.ai import (
|
|
ProviderConfigError,
|
|
ProviderTransientError,
|
|
Segment,
|
|
SummaryResult,
|
|
TranscriptResult,
|
|
)
|
|
|
|
|
|
def wav_bytes(payload_len: int = 64) -> bytes:
|
|
data = bytes(range(payload_len % 256)) * (payload_len // 256 + 1)
|
|
data = data[:payload_len]
|
|
header = (
|
|
b"RIFF" + struct.pack("<I", 36 + len(data)) + b"WAVE"
|
|
+ b"fmt " + struct.pack("<IHHIIHH", 16, 1, 1, 8000, 8000, 1, 8)
|
|
+ b"data" + struct.pack("<I", len(data))
|
|
)
|
|
return header + data
|
|
|
|
|
|
AUTH = {"email": "m7@example.com", "password": "m7-test-passw0rd-123"}
|
|
|
|
|
|
async def user_tokens(client, email=AUTH["email"], password=AUTH["password"]):
|
|
r = await client.post("/api/v1/auth/register", json={"email": email, "password": password})
|
|
assert r.status_code == 201, r.text
|
|
return r.json()["access_token"]
|
|
|
|
|
|
async def upload_recording(client, token, client_id=None, title="M7 standup"):
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
data = wav_bytes()
|
|
r = await client.post(
|
|
"/api/v1/uploads",
|
|
json={"declared_mime_type": "audio/wav", "declared_size_bytes": len(data),
|
|
"client_recording_id": client_id, "title": title},
|
|
headers=h,
|
|
)
|
|
assert r.status_code == 201, r.text
|
|
sid = r.json()["id"]
|
|
r = await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data,
|
|
headers={**h, "content-type": "application/octet-stream"})
|
|
assert r.status_code == 201, r.text
|
|
r = await client.post(f"/api/v1/uploads/{sid}/finalize",
|
|
json={"duration_seconds": 5.0}, headers=h)
|
|
assert r.status_code == 201, r.text
|
|
return r.json()
|
|
|
|
|
|
class FakeTranscriber:
|
|
name = "fake-stt"
|
|
|
|
def __init__(self, text="hello world from the meeting", fail=None):
|
|
self.text = text
|
|
self.fail = fail
|
|
self.calls = 0
|
|
|
|
async def transcribe(self, audio, mime, *, language_hint=None):
|
|
self.calls += 1
|
|
assert len(audio) > 0 and mime == "audio/wav"
|
|
if self.fail is not None:
|
|
raise self.fail
|
|
return TranscriptResult(
|
|
text=self.text, language="en",
|
|
segments=[Segment(0.0, 1.0, self.text)], model="fake-stt-1",
|
|
)
|
|
|
|
|
|
class FakeLlm:
|
|
name = "fake-llm"
|
|
|
|
def __init__(self, fail=None):
|
|
self.fail = fail
|
|
self.seen = []
|
|
self.seen_tones = []
|
|
|
|
async def summarize(self, transcript, *, title=None, tone=None,
|
|
on_progress=None):
|
|
self.seen.append(transcript)
|
|
self.seen_tones.append(tone)
|
|
if self.fail is not None:
|
|
raise self.fail
|
|
return SummaryResult(
|
|
short="Standup happened.", detailed="The team met and spoke.",
|
|
key_points=("a",), decisions=(), action_items=("ship it",),
|
|
questions=(), model="fake-llm-1",
|
|
)
|
|
|
|
|
|
_UNSET = object()
|
|
|
|
|
|
def use_fakes(monkeypatch, stt=_UNSET, llm=_UNSET, fallback=None):
|
|
tprov = FakeTranscriber() if stt is _UNSET else stt
|
|
lprov = FakeLlm() if llm is _UNSET else llm
|
|
monkeypatch.setattr(processing, "get_transcription_provider", lambda settings: tprov)
|
|
monkeypatch.setattr(processing, "get_llm_provider", lambda settings: lprov)
|
|
monkeypatch.setattr(processing, "get_llm_fallback_provider",
|
|
lambda settings: fallback)
|
|
|
|
|
|
async def jobs_for(recording_id):
|
|
async with db_session._session_factory() as s:
|
|
rows = (await s.scalars(
|
|
select(ProcessingJob).where(ProcessingJob.recording_id == uuid.UUID(recording_id))
|
|
)).all()
|
|
return {j.job_type: j for j in rows}
|
|
|
|
|
|
# --- no AI configured --------------------------------------------------------
|
|
|
|
|
|
async def test_none_configured_marks_ai_disabled(client):
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token)
|
|
assert rec["processing_status"] == "ai_disabled"
|
|
assert await jobs_for(rec["id"]) == {}
|
|
# Status endpoints: nothing there yet, but the recording exists.
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}/transcript", headers=h)
|
|
assert r.status_code == 404
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h)
|
|
assert r.status_code == 404
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}/jobs", headers=h)
|
|
assert r.status_code == 200 and r.json() == []
|
|
|
|
|
|
# --- happy path ----------------------------------------------------------------
|
|
|
|
|
|
async def test_full_pipeline_transcribe_then_summarize(client, monkeypatch):
|
|
stt, llm = FakeTranscriber(), FakeLlm()
|
|
use_fakes(monkeypatch, stt, llm)
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-full-1")
|
|
assert rec["processing_status"] == "queued"
|
|
jobs = await jobs_for(rec["id"])
|
|
assert set(jobs) == {JobType.transcribe}
|
|
assert jobs[JobType.transcribe].status == JobStatus.queued
|
|
|
|
await processing.run_transcribe({}, rec["id"])
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.succeeded
|
|
assert set(jobs) == {JobType.transcribe, JobType.summarize}
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
assert r.json()["processing_status"] == "processing"
|
|
|
|
t = await client.get(f"/api/v1/recordings/{rec['id']}/transcript", headers=h)
|
|
assert t.status_code == 200, t.text
|
|
body = t.json()
|
|
assert body["text"] == "hello world from the meeting"
|
|
assert body["version"] == 1 and body["provider"] == "fake-stt"
|
|
assert body["segments"][0]["text"].startswith("hello")
|
|
assert llm.seen == []
|
|
|
|
await processing.run_summarize({}, rec["id"])
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.summarize].status == JobStatus.succeeded
|
|
s = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h)
|
|
assert s.status_code == 200, s.text
|
|
content = s.json()["content"]
|
|
assert content["short"] == "Standup happened."
|
|
assert content["action_items"] == ["ship it"]
|
|
assert set(content) == {"short", "detailed", "key_points", "decisions",
|
|
"action_items", "questions"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
assert r.json()["processing_status"] == "completed"
|
|
assert llm.seen == ["hello world from the meeting"]
|
|
|
|
jobs_rows = await client.get(f"/api/v1/recordings/{rec['id']}/jobs", headers=h)
|
|
assert jobs_rows.status_code == 200
|
|
assert {j["job_type"] for j in jobs_rows.json()} == {"transcribe", "summarize"}
|
|
|
|
|
|
async def test_enqueue_is_idempotent(client, monkeypatch):
|
|
use_fakes(monkeypatch, FakeTranscriber(), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-idem-1")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
await processing.run_summarize({}, rec["id"])
|
|
async with db_session._session_factory() as s:
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, uuid.UUID(rec["id"]))
|
|
queued = await processing.enqueue_for_recording(s, row)
|
|
assert queued == []
|
|
n = await s.scalar(
|
|
select(func.count(ProcessingJob.id)).where(
|
|
ProcessingJob.recording_id == uuid.UUID(rec["id"]))
|
|
)
|
|
assert n == 2
|
|
await s.commit()
|
|
|
|
|
|
# --- failure modes -------------------------------------------------------------
|
|
|
|
|
|
async def test_transient_failure_retries_then_fails(client, monkeypatch):
|
|
stt = FakeTranscriber(fail=ProviderTransientError("stt down"))
|
|
use_fakes(monkeypatch, stt, FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-fail-1")
|
|
# First tries raise for arq retry; the attempt rolls back with the
|
|
# transaction, so the row still reads queued (arq tracks the tries).
|
|
with pytest.raises(ProviderTransientError):
|
|
await processing.run_transcribe({"job_try": 1}, rec["id"])
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.queued
|
|
# Last try marks the job (and recording) failed with a safe message.
|
|
await processing.run_transcribe({"job_try": 3}, rec["id"])
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.failed
|
|
assert jobs[JobType.transcribe].error == "stt down"
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
assert r.json()["processing_status"] == "failed"
|
|
assert r.json()["processing_error"] == "stt down"
|
|
|
|
|
|
async def test_config_error_fails_fast(client, monkeypatch):
|
|
stt = FakeTranscriber(fail=ProviderConfigError("bad credentials"))
|
|
use_fakes(monkeypatch, stt, FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-cfg-1")
|
|
await processing.run_transcribe({"job_try": 1}, rec["id"]) # no raise
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.failed
|
|
assert stt.calls == 1
|
|
|
|
|
|
async def test_raw_provider_crash_fails_job_no_escape(client, monkeypatch):
|
|
"""A non-AIError from the provider (e.g. av.InvalidDataError on corrupt
|
|
audio) must FAIL the row, not escape: an escaping exception leaves the
|
|
job 'running' and the startup sweep requeues it forever."""
|
|
stt = FakeTranscriber(fail=ValueError("Invalid data found when processing input"))
|
|
use_fakes(monkeypatch, stt, FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-crash-1")
|
|
await processing.run_transcribe({"job_try": 1}, rec["id"]) # must NOT raise
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.failed
|
|
assert "ValueError" in (jobs[JobType.transcribe].error or "")
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
assert r.json()["processing_status"] == "failed"
|
|
|
|
|
|
async def test_raw_summarize_crash_keeps_transcript(client, monkeypatch):
|
|
use_fakes(monkeypatch, FakeTranscriber("fine text"), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-crash-2")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
|
|
class ExplodingLlm(FakeLlm):
|
|
async def summarize(self, transcript, **kw):
|
|
raise RuntimeError("llm segfaulted")
|
|
|
|
use_fakes(monkeypatch, FakeTranscriber("fine text"), ExplodingLlm())
|
|
await processing.run_summarize({"job_try": 1}, rec["id"]) # must NOT raise
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.summarize].status == JobStatus.failed
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
body = r.json()
|
|
# Transcript survived: recording completes with a summary-failure note.
|
|
assert body["processing_status"] == "completed"
|
|
assert "Summary failed" in body["processing_error"]
|
|
|
|
|
|
# --- versioning -----------------------------------------------------------------
|
|
|
|
|
|
async def test_rerun_supersedes_auto_but_not_user_edits(client, monkeypatch):
|
|
use_fakes(monkeypatch, FakeTranscriber("v1 text"), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-ver-1")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
rid = uuid.UUID(rec["id"])
|
|
|
|
async def versions():
|
|
async with db_session._session_factory() as s:
|
|
rows = (await s.scalars(
|
|
select(Transcript).where(Transcript.recording_id == rid)
|
|
.order_by(Transcript.version))).all()
|
|
return [(r.version, r.text, r.superseded_at is not None, r.edited_by_user)
|
|
for r in rows]
|
|
|
|
assert await versions() == [(1, "v1 text", False, False)]
|
|
|
|
# Queue another transcription run manually (re-entry path).
|
|
async with db_session._session_factory() as s:
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, rid)
|
|
await processing.enqueue_for_recording(s, row)
|
|
await s.commit()
|
|
use_fakes(monkeypatch, FakeTranscriber("v2 text"), FakeLlm())
|
|
await processing.run_transcribe({}, rec["id"])
|
|
assert await versions() == [(1, "v1 text", True, False), (2, "v2 text", False, False)]
|
|
|
|
# A user edit wins: the next auto run inserts nothing.
|
|
async with db_session._session_factory() as s:
|
|
v2 = await s.scalar(
|
|
select(Transcript).where(Transcript.recording_id == rid,
|
|
Transcript.version == 2))
|
|
v2.edited_by_user = True
|
|
v2.text = "user corrected text"
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, rid)
|
|
await processing.enqueue_for_recording(s, row)
|
|
await s.commit()
|
|
use_fakes(monkeypatch, FakeTranscriber("v3 text"), FakeLlm())
|
|
await processing.run_transcribe({}, rec["id"])
|
|
got = await versions()
|
|
assert len(got) == 2 and got[1][1] == "user corrected text"
|
|
|
|
|
|
# --- llm-only and skips ----------------------------------------------------------
|
|
|
|
|
|
async def test_llm_only_without_transcript_queues_nothing(client, monkeypatch):
|
|
use_fakes(monkeypatch, None, FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-llm-1")
|
|
assert rec["processing_status"] == "uploaded"
|
|
assert await jobs_for(rec["id"]) == {}
|
|
|
|
|
|
async def test_llm_only_with_manual_transcript_summarizes(client, monkeypatch):
|
|
use_fakes(monkeypatch, None, FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-llm-2")
|
|
rid = uuid.UUID(rec["id"])
|
|
async with db_session._session_factory() as s:
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, rid)
|
|
s.add(Transcript(recording_id=rid, version=1, provider="manual",
|
|
text="handwritten notes", edited_by_user=True))
|
|
await s.flush()
|
|
queued = await processing.enqueue_for_recording(s, row)
|
|
assert queued == [JobType.summarize]
|
|
await s.commit()
|
|
await processing.run_summarize({}, rec["id"])
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
s = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h)
|
|
assert s.status_code == 200
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
|
|
assert r.json()["processing_status"] == "completed"
|
|
|
|
|
|
# --- sweep ------------------------------------------------------------------------
|
|
|
|
|
|
async def test_sweep_requeues_stale_jobs(client, monkeypatch):
|
|
use_fakes(monkeypatch, FakeTranscriber(), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-sweep-1")
|
|
rid = uuid.UUID(rec["id"])
|
|
async with db_session._session_factory() as s:
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, rid)
|
|
# Simulate a crashed worker + a lost transport, respectively.
|
|
await processing.reset_or_create(s, row, JobType.transcribe)
|
|
jobs = await jobs_for(rec["id"])
|
|
jobs[JobType.transcribe].status = JobStatus.running
|
|
await s.commit()
|
|
# reset_or_create leaves running rows alone, so force the second shape:
|
|
async with db_session._session_factory() as s:
|
|
extra = ProcessingJob(recording_id=rid, job_type=JobType.summarize,
|
|
status=JobStatus.queued)
|
|
s.add(extra)
|
|
await s.commit()
|
|
count = await processing.sweep_stale()
|
|
assert count == 2
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.queued
|
|
assert jobs[JobType.summarize].status == JobStatus.queued
|
|
|
|
|
|
async def test_sweep_leaves_live_running_jobs_alone(client, monkeypatch):
|
|
"""Regression: sweep re-enqueued every in-flight transcription every
|
|
5 minutes; with local whisper each copy ran concurrently and one worker
|
|
OOM-swap-swelled the box. A running job started recently is live work."""
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
use_fakes(monkeypatch, FakeTranscriber(), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-sweep-live")
|
|
rid = uuid.UUID(rec["id"])
|
|
async with db_session._session_factory() as s:
|
|
from shonar.db.models import Recording
|
|
|
|
row = await s.get(Recording, rid)
|
|
await processing.reset_or_create(s, row, JobType.transcribe)
|
|
live = (await s.scalars(
|
|
select(ProcessingJob).where(
|
|
ProcessingJob.recording_id == rid,
|
|
ProcessingJob.job_type == JobType.transcribe)
|
|
)).one()
|
|
live.status = JobStatus.running
|
|
live.started_at = datetime.now(UTC) # worker is chewing it right now
|
|
stale = ProcessingJob(recording_id=rid, job_type=JobType.summarize,
|
|
status=JobStatus.running, attempt=1)
|
|
stale.started_at = datetime.now(UTC) - timedelta(
|
|
hours=processing.STALE_RUNNING_AFTER.total_seconds() / 3600 + 1)
|
|
s.add(stale)
|
|
await s.commit()
|
|
count = await processing.sweep_stale()
|
|
assert count == 1 # only the orphaned old run
|
|
jobs = await jobs_for(rec["id"])
|
|
assert jobs[JobType.transcribe].status == JobStatus.running
|
|
assert jobs[JobType.summarize].status == JobStatus.queued
|
|
|
|
|
|
# --- ownership ---------------------------------------------------------------------
|
|
|
|
|
|
async def test_ai_endpoints_enforce_ownership(client, monkeypatch):
|
|
use_fakes(monkeypatch, FakeTranscriber(), FakeLlm())
|
|
token = await user_tokens(client)
|
|
rec = await upload_recording(client, token, client_id="m7-own-1")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
other = await user_tokens(client, email="m7-other@example.com",
|
|
password="m7-test-passw0rd-456")
|
|
h = {"Authorization": f"Bearer {other}"}
|
|
for path in ("transcript", "summary", "jobs"):
|
|
r = await client.get(f"/api/v1/recordings/{rec['id']}/{path}", headers=h)
|
|
assert r.status_code == 404, path
|
|
|
|
|
|
# --- summarize fallback to the rescue summarizer --------------------------------
|
|
|
|
async def test_summarize_fallback_rescues_after_primary_exhausted(client, monkeypatch):
|
|
primary = FakeLlm(fail=ProviderTransientError("LLM reply was not JSON."))
|
|
fb = FakeLlm()
|
|
fb.name = "fake-ollama"
|
|
use_fakes(monkeypatch, FakeTranscriber(), primary, fallback=fb)
|
|
token = await user_tokens(client, email="m7-fb1@example.com")
|
|
rec = await upload_recording(client, token, client_id="m7-fb-1")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
# Last try: primary out of budget -> rescue runs and the note lands in job.error.
|
|
await processing.run_summarize({"job_try": 3}, rec["id"])
|
|
job = (await jobs_for(rec["id"]))[JobType.summarize]
|
|
assert job.status == JobStatus.succeeded
|
|
assert "fake-llm" in (job.error or "") and "fake-ollama" in (job.error or "")
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
s = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h)
|
|
assert s.status_code == 200
|
|
assert s.json()["provider"] == "fake-ollama"
|
|
assert fb.seen == ["hello world from the meeting"]
|
|
|
|
|
|
async def test_summarize_fallback_also_fails_marks_job_failed(client, monkeypatch):
|
|
err = ProviderTransientError("LLM reply was not JSON.")
|
|
fb = FakeLlm(fail=ProviderTransientError("ollama down"))
|
|
fb.name = "fake-ollama"
|
|
use_fakes(monkeypatch, FakeTranscriber(), FakeLlm(fail=err), fallback=fb)
|
|
token = await user_tokens(client, email="m7-fb2@example.com")
|
|
rec = await upload_recording(client, token, client_id="m7-fb-2")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
await processing.run_summarize({"job_try": 3}, rec["id"])
|
|
job = (await jobs_for(rec["id"]))[JobType.summarize]
|
|
assert job.status == JobStatus.failed
|
|
assert "not JSON" in job.error # the primary's error is what we surface
|
|
assert fb.seen == ["hello world from the meeting"] # fallback was tried
|
|
|
|
|
|
async def test_summarize_keeps_retry_budget_before_fallback(client, monkeypatch):
|
|
# Try 1 of 3 transient: normal arq retry wins, fallback stays untouched.
|
|
fb = FakeLlm()
|
|
fb.name = "fake-ollama"
|
|
use_fakes(monkeypatch, FakeTranscriber(),
|
|
FakeLlm(fail=ProviderTransientError("boom")), fallback=fb)
|
|
token = await user_tokens(client, email="m7-fb3@example.com")
|
|
rec = await upload_recording(client, token, client_id="m7-fb-3")
|
|
await processing.run_transcribe({}, rec["id"])
|
|
with pytest.raises(ProviderTransientError):
|
|
await processing.run_summarize({"job_try": 1}, rec["id"])
|
|
job = (await jobs_for(rec["id"]))[JobType.summarize]
|
|
assert job.status == JobStatus.queued
|
|
assert fb.seen == []
|