generated from john/python-template
345 lines
13 KiB
Python
345 lines
13 KiB
Python
"""Integration tests for end-to-end upload and worker pipeline behavior."""
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import JobStatus
|
|
from transcription.providers.base import TranscriptionResult
|
|
from transcription.services import ServiceBundle
|
|
from transcription.services.store import create_job_for_document
|
|
from transcription.services.store import create_upload_job
|
|
from transcription.services.workflows import advance_job
|
|
|
|
|
|
def _build_services(default_session_factory) -> ServiceBundle:
|
|
services = ServiceBundle()
|
|
object.__setattr__(
|
|
services,
|
|
"documents",
|
|
services.documents.__class__(session_factory=default_session_factory),
|
|
)
|
|
object.__setattr__(
|
|
services,
|
|
"jobs",
|
|
services.jobs.__class__(session_factory=default_session_factory),
|
|
)
|
|
object.__setattr__(
|
|
services,
|
|
"transcriptions",
|
|
services.transcriptions.__class__(session_factory=default_session_factory),
|
|
)
|
|
return services
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestPipelineSuccessFlow:
|
|
"""Verify end-to-end success lifecycle behavior."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_then_worker_persists_transcribed_terminal_state(
|
|
self, async_session, default_session_factory, 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 = await create_upload_job(
|
|
filename="pipeline.jpg",
|
|
file_bytes=b"pipeline-bytes",
|
|
session=async_session,
|
|
settings=settings,
|
|
)
|
|
|
|
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",
|
|
)
|
|
|
|
async def _fake_transcribe_document_image(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
settings=None,
|
|
provider=None,
|
|
) -> TranscriptionResult:
|
|
_ = (image_path, prompt_name, settings, provider)
|
|
return TranscriptionResult(
|
|
text="Pipeline transcript",
|
|
provider="openrouter",
|
|
model="test-model",
|
|
prompt_name="transcribe_document.md",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"transcription.services.workflows.transcribe_document_image",
|
|
_fake_transcribe_document_image,
|
|
)
|
|
|
|
services = _build_services(default_session_factory)
|
|
queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
|
|
processed = queued_job is not None
|
|
if queued_job is not None:
|
|
await advance_job(job=queued_job, services=services, session=async_session)
|
|
job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
|
|
|
|
assert processed is True
|
|
assert job is not None
|
|
assert job.status == JobStatus.TRANSCRIBED
|
|
assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources)
|
|
assert all(job_source.error_detail is None for job_source in job.job_sources)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_transcribes_all_sources_for_multi_page_job(
|
|
self,
|
|
async_session,
|
|
default_session_factory,
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""Worker stores transcription output for every source linked to the queued job."""
|
|
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
|
document = Document(id=uuid4(), name="multi-page-document")
|
|
async_session.add(document)
|
|
await async_session.commit()
|
|
|
|
create_result = await create_job_for_document(
|
|
document_id=document.id,
|
|
uploads=[
|
|
("page-01.jpg", b"one"),
|
|
("page-02.jpg", b"two"),
|
|
("page-03.jpg", b"three"),
|
|
],
|
|
session=async_session,
|
|
settings=settings,
|
|
)
|
|
|
|
async def _fake_transcribe_document_image(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
settings=None,
|
|
provider=None,
|
|
) -> TranscriptionResult:
|
|
page_name = Path(image_path).name
|
|
_ = (prompt_name, settings, provider)
|
|
return TranscriptionResult(
|
|
text=f"Transcript for {page_name}",
|
|
provider="openrouter",
|
|
model="test-model",
|
|
prompt_name="transcribe_document.md",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"transcription.services.workflows.transcribe_document_image",
|
|
_fake_transcribe_document_image,
|
|
)
|
|
|
|
services = _build_services(default_session_factory)
|
|
queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
assert queued_job is not None
|
|
|
|
await advance_job(job=queued_job, services=services, session=async_session)
|
|
job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
|
|
assert job.status == JobStatus.TRANSCRIBED
|
|
assert len(job.job_sources) == 3
|
|
assert all(job_source.status == JobSourceStatus.TRANSCRIBED for job_source in job.job_sources)
|
|
assert all(job_source.raw_transcription for job_source in job.job_sources)
|
|
assert all(job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_marks_partial_success_when_some_sources_fail(
|
|
self,
|
|
async_session,
|
|
default_session_factory,
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""Mixed page outcomes produce PARTIAL_SUCCESS and preserve per-source status."""
|
|
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
|
document = Document(id=uuid4(), name="partial-page-document")
|
|
async_session.add(document)
|
|
await async_session.commit()
|
|
|
|
create_result = await create_job_for_document(
|
|
document_id=document.id,
|
|
uploads=[
|
|
("page-01.jpg", b"one"),
|
|
("page-02.jpg", b"two"),
|
|
],
|
|
session=async_session,
|
|
settings=settings,
|
|
)
|
|
|
|
call_count = 0
|
|
|
|
async def _fake_transcribe_document_image(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
settings=None,
|
|
provider=None,
|
|
) -> TranscriptionResult:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
_ = (prompt_name, settings, provider)
|
|
if call_count == 2:
|
|
raise RuntimeError("simulated page failure")
|
|
return TranscriptionResult(
|
|
text="Transcript for first page",
|
|
provider="openrouter",
|
|
model="test-model",
|
|
prompt_name="transcribe_document.md",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"transcription.services.workflows.transcribe_document_image",
|
|
_fake_transcribe_document_image,
|
|
)
|
|
|
|
services = _build_services(default_session_factory)
|
|
queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
assert queued_job is not None
|
|
|
|
await advance_job(job=queued_job, services=services, session=async_session)
|
|
job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
|
|
assert job.status == JobStatus.PARTIAL_SUCCESS
|
|
assert len(job.job_sources) == 2
|
|
statuses = {job_source.status for job_source in job.job_sources}
|
|
assert statuses == {JobSourceStatus.TRANSCRIBED, JobSourceStatus.FAILED}
|
|
assert any(job_source.error_detail is not None for job_source in job.job_sources)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_skips_already_transcribed_sources_on_resubmit(
|
|
self,
|
|
async_session,
|
|
default_session_factory,
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""Queued jobs only process non-transcribed JobSource records."""
|
|
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
|
document = Document(id=uuid4(), name="resubmit-filter-document")
|
|
async_session.add(document)
|
|
await async_session.commit()
|
|
|
|
create_result = await create_job_for_document(
|
|
document_id=document.id,
|
|
uploads=[
|
|
("page-01.jpg", b"one"),
|
|
("page-02.jpg", b"two"),
|
|
],
|
|
session=async_session,
|
|
settings=settings,
|
|
)
|
|
|
|
services = _build_services(default_session_factory)
|
|
job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
page_one = next(js for js in job.job_sources if js.source is not None and js.source.page_number == 1)
|
|
page_two = next(js for js in job.job_sources if js.source is not None and js.source.page_number == 2)
|
|
|
|
page_one.status = JobSourceStatus.TRANSCRIBED
|
|
page_one.raw_transcription = "existing transcript"
|
|
page_two.status = JobSourceStatus.PENDING
|
|
page_two.raw_transcription = None
|
|
await services.transcriptions.update_job_source(job_source=page_one, session=async_session)
|
|
await services.transcriptions.update_job_source(job_source=page_two, session=async_session)
|
|
await services.jobs.update_job_state(job_id=job.id, status=JobStatus.QUEUED, session=async_session)
|
|
await async_session.commit()
|
|
|
|
call_count = 0
|
|
|
|
async def _fake_transcribe_document_image(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
settings=None,
|
|
provider=None,
|
|
) -> TranscriptionResult:
|
|
nonlocal call_count
|
|
_ = (image_path, prompt_name, settings, provider)
|
|
call_count += 1
|
|
return TranscriptionResult(
|
|
text="new transcript",
|
|
provider="openrouter",
|
|
model="test-model",
|
|
prompt_name="transcribe_document.md",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"transcription.services.workflows.transcribe_document_image",
|
|
_fake_transcribe_document_image,
|
|
)
|
|
|
|
queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
assert queued_job is not None
|
|
await advance_job(job=queued_job, services=services, session=async_session)
|
|
|
|
refreshed = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
|
|
assert call_count == 1
|
|
statuses = {js.status for js in refreshed.job_sources}
|
|
assert statuses == {JobSourceStatus.TRANSCRIBED}
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestPipelineFailureFlow:
|
|
"""Verify end-to-end failure lifecycle behavior."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_then_worker_persists_failed_terminal_state(
|
|
self,
|
|
async_session,
|
|
default_session_factory,
|
|
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 = await create_upload_job(
|
|
filename="pipeline.jpg",
|
|
file_bytes=b"pipeline-bytes",
|
|
session=async_session,
|
|
settings=settings,
|
|
)
|
|
|
|
async def _fake_transcribe_document_image(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
settings=None,
|
|
provider=None,
|
|
) -> TranscriptionResult:
|
|
_ = (image_path, prompt_name, settings, provider)
|
|
raise RuntimeError("pipeline provider failure")
|
|
|
|
monkeypatch.setattr(
|
|
"transcription.services.workflows.transcribe_document_image",
|
|
_fake_transcribe_document_image,
|
|
)
|
|
|
|
services = _build_services(default_session_factory)
|
|
queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
|
|
processed = queued_job is not None
|
|
if queued_job is not None:
|
|
await advance_job(job=queued_job, services=services, session=async_session)
|
|
job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
|
|
|
|
assert processed is True
|
|
assert job is not None
|
|
assert job.status == JobStatus.FAILED
|
|
assert all(job_source.raw_transcription is None for job_source in job.job_sources)
|
|
assert any(job_source.error_detail is not None for job_source in job.job_sources)
|
|
error_detail = next(job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None)
|
|
assert "pipeline provider failure" in error_detail
|
|
assert "[internal_unexpected_error]" in error_detail
|
|
assert "error_id=" in error_detail
|