generated from john/python-template
122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
"""Integration tests for end-to-end upload and worker pipeline behavior."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobStatus
|
|
from transcription.providers.base import TranscriptionResult
|
|
from transcription.services import ServiceBundle
|
|
from transcription.services.store import create_upload_job
|
|
from transcription.services.workflows import advance_job
|
|
|
|
|
|
@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, 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 = ServiceBundle()
|
|
queued_job = await services.jobs.read_next_queued_job(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 async_session.get(Job, upload_result.job_id)
|
|
|
|
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.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, 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 = ServiceBundle()
|
|
queued_job = await services.jobs.read_next_queued_job(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 async_session.get(Job, upload_result.job_id)
|
|
|
|
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
|