bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update documentation for consistency and refactor the code. An unresolved error in testing still exists.

Browse Source
This commit is contained in:
Jim Lancaster
2026-07-29 14:12:18 -05:00
parent eaf9805121
commit 0973311d9f
23 changed files with 451 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_pipeline_flow.py
+33 -30
View File
@@ -3,12 +3,12 @@
from pathlib import Path
import pytest
from sqlmodel import select
from transcription.config import Settings
from transcription.models import Job, JobStatus, Transcript
from transcription.models import Job
from transcription.models import JobStatus
from transcription.providers.base import TranscriptionResult
from transcription.services.upload import create_upload_job
from transcription.services.store import create_upload_job
from transcription.worker import process_next_queued_job
@@ -16,61 +16,64 @@ from transcription.worker import process_next_queued_job
class TestPipelineSuccessFlow:
"""Verify end-to-end success lifecycle behavior."""
def test_upload_then_worker_persists_transcribed_terminal_state(self, session, tmp_path: Path, monkeypatch):
"""Upload followed by worker processing persists transcript and transcribed status."""
@pytest.mark.asyncio
async def test_upload_then_worker_persists_transcribed_terminal_state(
self, async_session, tmp_path: Path, monkeypatch
):
"""Upload followed by worker processing persists job transcription and transcribed status."""
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
upload_result = create_upload_job(
upload_result = await create_upload_job(
filename="pipeline.jpg",
file_bytes=b"pipeline-bytes",
session=session,
session=async_session,
settings=settings,
)
def _fake_transcribe(_path: str) -> TranscriptionResult:
return TranscriptionResult(text="Pipeline transcript", provider="openrouter", model="test-model")
async def _fake_transcribe(*, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
_ = (prompt_text, image_bytes, mime_type)
return TranscriptionResult(text="Pipeline transcript", provider="openrouter", model="test-model", prompt_name="transcribe_document.md")
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
monkeypatch.setattr("transcription.services.transcription.OpenRouterTranscriptionProvider.transcribe", _fake_transcribe)
processed = process_next_queued_job(session=session)
job = session.get(Job, upload_result.job_id)
transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
processed = await process_next_queued_job(session=async_session)
job = await async_session.get(Job, upload_result.job_id)
assert processed is True
assert job is not None
assert job.status == JobStatus.TRANSCRIBED
assert transcript is not None
assert transcript.text == "Pipeline transcript"
assert transcript.error_detail is None
assert job.text == "Pipeline transcript"
assert job.error_detail is None
@pytest.mark.integration
class TestPipelineFailureFlow:
"""Verify end-to-end failure lifecycle behavior."""
def test_upload_then_worker_persists_failed_terminal_state(self, session, tmp_path: Path, monkeypatch):
"""Upload followed by worker processing persists error detail and failed status."""
@pytest.mark.asyncio
async def test_upload_then_worker_persists_failed_terminal_state(self, async_session, tmp_path: Path, monkeypatch):
"""Upload followed by worker processing persists error detail and failed status on the job."""
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
upload_result = create_upload_job(
upload_result = await create_upload_job(
filename="pipeline.jpg",
file_bytes=b"pipeline-bytes",
session=session,
session=async_session,
settings=settings,
)
def _fake_transcribe(_path: str) -> TranscriptionResult:
async def _fake_transcribe(*, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
_ = (prompt_text, image_bytes, mime_type)
raise RuntimeError("pipeline provider failure")
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
monkeypatch.setattr("transcription.services.transcription.OpenRouterTranscriptionProvider.transcribe", _fake_transcribe)
processed = process_next_queued_job(session=session)
job = session.get(Job, upload_result.job_id)
transcript = session.exec(select(Transcript).where(Transcript.job_id == upload_result.job_id)).first()
processed = await process_next_queued_job(session=async_session)
job = await async_session.get(Job, upload_result.job_id)
assert processed is True
assert job is not None
assert job.status == JobStatus.FAILED
assert transcript is not None
assert transcript.text is None
assert "pipeline provider failure" in transcript.error_detail
assert "[internal_unexpected_error]" in transcript.error_detail
assert "error_id=" in transcript.error_detail
assert job.text is None
assert job.error_detail is not None
assert "pipeline provider failure" in job.error_detail
assert "[internal_unexpected_error]" in job.error_detail
assert "error_id=" in job.error_detail
tests/test_db.py
+4 -3
View File
@@ -18,10 +18,10 @@ class TestSchemaBootstrap:
"""Verify create_all produces the expected table set."""
def test_create_all_creates_expected_tables(self):
"""After create_all(), document, job, and transcript tables exist."""
"""After create_all(), document, source, job, and revision tables exist."""
engine = _in_memory_engine()
# Ensure models are imported so metadata is populated
from transcription.models import Document, Job, Transcript # noqa: F401
from transcription.models import Document, Job, Revision, Source # noqa: F401
import transcription.db as db_module
@@ -31,7 +31,8 @@ class TestSchemaBootstrap:
table_names = set(inspector.get_table_names())
assert "document" in table_names
assert "job" in table_names
assert "transcript" in table_names
assert "source" in table_names
assert "revision" in table_names
class TestSessionFactory:
tests/test_models.py
+86 -104
View File
@@ -1,31 +1,28 @@
"""Tests for transcription.models — Document, Job, Transcript persistence and relationships."""
"""Tests for transcription.models — Document, Source, Job, Revision persistence and relationships."""
from uuid import UUID
import pytest
from sqlalchemy.exc import IntegrityError
from transcription.models import Document, Job, JobStatus, Transcript
from transcription.models import Document, Job, JobStatus, Revision, Source
def _make_document(**overrides) -> Document:
"""Create a Document with sensible defaults."""
defaults = {"filename": "letter.jpg", "file_path": "/uploads/letter.jpg"}
defaults = {"name": "letter bundle"}
defaults.update(overrides)
return Document(**defaults)
def _persist_document(session) -> Document:
"""Create, persist, and return a Document."""
doc = _make_document()
session.add(doc)
document = _make_document()
session.add(document)
session.commit()
session.refresh(doc)
return doc
session.refresh(document)
return document
def _persist_job(session, document: Document) -> Job:
"""Create, persist, and return a Job linked to a Document."""
job = Job(document_id=document.id)
session.add(job)
session.commit()
@@ -33,147 +30,132 @@ def _persist_job(session, document: Document) -> Job:
return job
class TestDocumentModel:
"""Verify Document creation and default field population."""
def _persist_source(session, document: Document, job: Job, **overrides) -> Source:
defaults = {
"document_id": document.id,
"job_id": job.id,
"upload_name": "letter.jpg",
"filename": "stored-letter.jpg",
"file_path": "/uploads/stored-letter.jpg",
}
defaults.update(overrides)
source = Source(**defaults)
session.add(source)
session.commit()
session.refresh(source)
return source
class TestDocumentModel:
def test_can_be_persisted(self, session):
"""A Document round-trips through the database with correct fields."""
doc = _persist_document(session)
fetched = session.get(Document, doc.id)
document = _persist_document(session)
fetched = session.get(Document, document.id)
assert fetched is not None
assert fetched.filename == "letter.jpg"
assert fetched.file_path == "/uploads/letter.jpg"
assert fetched.name == "letter bundle"
def test_defaults_are_populated(self, session):
"""id is a UUID and uploaded_at is populated on creation."""
doc = _persist_document(session)
assert isinstance(doc.id, UUID)
assert doc.uploaded_at is not None
document = _persist_document(session)
assert isinstance(document.id, UUID)
class TestJobModel:
"""Verify Job creation, defaults, and status transitions."""
def test_can_be_created_for_document(self, session):
"""A Job linked to a Document via FK persists correctly."""
doc = _persist_document(session)
job = _persist_job(session, doc)
document = _persist_document(session)
job = _persist_job(session, document)
fetched = session.get(Job, job.id)
assert fetched is not None
assert fetched.document_id == doc.id
assert fetched.document_id == document.id
def test_defaults_are_populated(self, session):
"""Default status is queued; created_at and updated_at are populated."""
doc = _persist_document(session)
job = _persist_job(session, doc)
document = _persist_document(session)
job = _persist_job(session, document)
assert job.status == JobStatus.QUEUED
assert job.retry_count == 0
assert job.created_at is not None
assert job.updated_at is not None
assert job.date_created is not None
assert job.date_updated is not None
def test_transitions_to_transcribed(self, session):
"""Status updates from queued to processing to transcribed."""
doc = _persist_document(session)
job = _persist_job(session, doc)
assert job.status == JobStatus.QUEUED
document = _persist_document(session)
job = _persist_job(session, document)
job.status = JobStatus.PROCESSING
session.add(job)
session.commit()
session.refresh(job)
assert job.status == JobStatus.PROCESSING
job.status = JobStatus.TRANSCRIBED
session.add(job)
session.commit()
session.refresh(job)
assert job.status == JobStatus.TRANSCRIBED
def test_transitions_to_failed(self, session):
"""Status updates from processing to failed."""
doc = _persist_document(session)
job = _persist_job(session, doc)
job.status = JobStatus.PROCESSING
session.add(job)
session.commit()
session.refresh(job)
class TestSourceModel:
def test_can_be_created_for_document_and_job(self, session):
document = _persist_document(session)
job = _persist_job(session, document)
source = _persist_source(session, document, job)
job.status = JobStatus.FAILED
session.add(job)
session.commit()
session.refresh(job)
assert job.status == JobStatus.FAILED
class TestTranscriptModel:
"""Verify Transcript persistence for success and failure cases."""
def test_success_record_persists(self, session):
"""A Transcript with text set and error_detail None persists correctly."""
doc = _persist_document(session)
job = _persist_job(session, doc)
transcript = Transcript(job_id=job.id, text="Dear Sir, ...")
session.add(transcript)
session.commit()
session.refresh(transcript)
fetched = session.get(Transcript, transcript.id)
fetched = session.get(Source, source.id)
assert fetched is not None
assert fetched.text == "Dear Sir, ..."
assert fetched.error_detail is None
assert fetched.document_id == document.id
assert fetched.job_id == job.id
assert fetched.date_uploaded is not None
def test_failure_record_persists(self, session):
"""A Transcript with text None and error_detail set persists correctly."""
doc = _persist_document(session)
job = _persist_job(session, doc)
transcript = Transcript(job_id=job.id, error_detail="Provider timeout")
session.add(transcript)
class TestRevisionModel:
def test_revision_persists_for_source(self, session):
document = _persist_document(session)
job = _persist_job(session, document)
source = _persist_source(session, document, job)
revision = Revision(source_id=source.id, text="Edited revision text")
session.add(revision)
session.commit()
session.refresh(transcript)
session.refresh(revision)
fetched = session.get(Transcript, transcript.id)
fetched = session.get(Revision, revision.id)
assert fetched is not None
assert fetched.text is None
assert fetched.error_detail == "Provider timeout"
assert fetched.text == "Edited revision text"
assert fetched.date_created is not None
def test_job_id_is_unique(self, session):
"""Inserting two transcripts with the same job_id raises an integrity error."""
doc = _persist_document(session)
job = _persist_job(session, doc)
def test_source_id_is_unique(self, session):
document = _persist_document(session)
job = _persist_job(session, document)
source = _persist_source(session, document, job)
t1 = Transcript(job_id=job.id, text="First")
session.add(t1)
first = Revision(source_id=source.id, text="First")
session.add(first)
session.commit()
t2 = Transcript(job_id=job.id, text="Duplicate")
session.add(t2)
duplicate = Revision(source_id=source.id, text="Duplicate")
session.add(duplicate)
with pytest.raises(IntegrityError):
session.commit()
class TestRelationships:
"""Verify SQLModel relationship navigation between models."""
def test_document_exposes_jobs_and_sources(self, session):
document = _persist_document(session)
job = _persist_job(session, document)
_persist_source(session, document, job)
def test_document_exposes_jobs(self, session):
"""document.jobs returns the linked Job list."""
doc = _persist_document(session)
_persist_job(session, doc)
_persist_job(session, doc)
session.refresh(document)
assert len(document.jobs) == 1
assert len(document.sources) == 1
session.refresh(doc)
assert len(doc.jobs) == 2
assert all(isinstance(j, Job) for j in doc.jobs)
def test_source_exposes_optional_single_revision(self, session):
document = _persist_document(session)
job = _persist_job(session, document)
source = _persist_source(session, document, job)
def test_job_exposes_transcript(self, session):
"""job.transcript returns the linked Transcript."""
doc = _persist_document(session)
job = _persist_job(session, doc)
transcript = Transcript(job_id=job.id, text="Transcribed text")
session.add(transcript)
assert source.revision is None
revision = Revision(source_id=source.id, text="Edited")
session.add(revision)
session.commit()
session.refresh(job)
assert job.transcript is not None
assert isinstance(job.transcript, Transcript)
assert job.transcript.text == "Transcribed text"
session.refresh(source)
assert source.revision is not None
assert source.revision.text == "Edited"