generated from john/python-template
pruning
This commit is contained in:
John Lancaster
@@ -1,137 +0,0 @@
|
||||
"""Tests for transcription.services.transcription."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from transcription.config import Settings
|
||||
from transcription.providers.base import ProviderError, TranscriptionResult
|
||||
from transcription.services.transcription import (
|
||||
PromptLoadError,
|
||||
TranscriptionError,
|
||||
load_image_payload,
|
||||
load_prompt_text,
|
||||
transcribe_document_image,
|
||||
)
|
||||
|
||||
|
||||
class _FakeProvider:
|
||||
def __init__(self, *, result: TranscriptionResult | None = None, error: Exception | None = None):
|
||||
self._result = result or TranscriptionResult(
|
||||
text="Transcript output",
|
||||
provider="openrouter",
|
||||
model="test-model",
|
||||
)
|
||||
self._error = error
|
||||
self.calls: list[dict[str, object]] = []
|
||||
|
||||
def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
|
||||
self.calls.append(
|
||||
{
|
||||
"prompt_text": prompt_text,
|
||||
"image_bytes": image_bytes,
|
||||
"mime_type": mime_type,
|
||||
}
|
||||
)
|
||||
if self._error:
|
||||
raise self._error
|
||||
return self._result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestPromptLoading:
|
||||
"""Verify prompt artifact loading and validation."""
|
||||
|
||||
def test_loads_prompt_text_from_prompt_dir(self, tmp_path: Path):
|
||||
"""Prompt loader returns canonical prompt text from configured prompt directory."""
|
||||
prompt_dir = tmp_path / "prompts"
|
||||
prompt_dir.mkdir()
|
||||
prompt_file = prompt_dir / "transcribe_document.md"
|
||||
prompt_file.write_text("Prompt body", encoding="utf-8")
|
||||
|
||||
settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
|
||||
text = load_prompt_text(settings=settings)
|
||||
|
||||
assert text == "Prompt body"
|
||||
|
||||
def test_missing_prompt_raises_error(self, tmp_path: Path):
|
||||
"""Prompt loader raises PromptLoadError when the file is missing."""
|
||||
prompt_dir = tmp_path / "prompts"
|
||||
prompt_dir.mkdir()
|
||||
|
||||
settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
|
||||
with pytest.raises(PromptLoadError) as exc_info:
|
||||
load_prompt_text(settings=settings)
|
||||
|
||||
assert exc_info.value.category.value == "infrastructure_persistent_error"
|
||||
assert "verify prompt_dir" in exc_info.value.suggestion.lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestImageLoading:
|
||||
"""Verify local image payload loading and mime detection."""
|
||||
|
||||
def test_load_image_payload_reads_bytes_and_mime_type(self, tmp_path: Path):
|
||||
"""Image loader returns file bytes and a detected MIME type for supported files."""
|
||||
image_path = tmp_path / "sample.png"
|
||||
image_bytes = b"\x89PNG\r\n\x1a\n"
|
||||
image_path.write_bytes(image_bytes)
|
||||
|
||||
loaded_bytes, mime_type = load_image_payload(image_path)
|
||||
|
||||
assert loaded_bytes == image_bytes
|
||||
assert mime_type == "image/png"
|
||||
|
||||
def test_missing_image_raises_error(self, tmp_path: Path):
|
||||
"""Image loader raises TranscriptionError when image file does not exist."""
|
||||
missing = tmp_path / "missing.png"
|
||||
|
||||
with pytest.raises(TranscriptionError) as exc_info:
|
||||
load_image_payload(missing)
|
||||
|
||||
assert exc_info.value.category.value == "not_found_error"
|
||||
assert "verify" in exc_info.value.suggestion.lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestTranscriptionService:
|
||||
"""Verify service orchestration across prompt, image, and provider calls."""
|
||||
|
||||
def test_transcribe_document_image_calls_provider_once(self, tmp_path: Path):
|
||||
"""Service loads prompt and image, then invokes provider exactly once."""
|
||||
prompt_dir = tmp_path / "prompts"
|
||||
prompt_dir.mkdir()
|
||||
(prompt_dir / "transcribe_document.md").write_text("Prompt body", encoding="utf-8")
|
||||
|
||||
image_path = tmp_path / "document.jpg"
|
||||
image_path.write_bytes(b"jpeg-bytes")
|
||||
|
||||
settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
|
||||
provider = _FakeProvider()
|
||||
|
||||
result = transcribe_document_image(image_path, settings=settings, provider=provider)
|
||||
|
||||
assert result.text == "Transcript output"
|
||||
assert len(provider.calls) == 1
|
||||
assert provider.calls[0]["prompt_text"] == "Prompt body"
|
||||
assert provider.calls[0]["image_bytes"] == b"jpeg-bytes"
|
||||
assert provider.calls[0]["mime_type"] == "image/jpeg"
|
||||
|
||||
def test_provider_error_is_wrapped(self, tmp_path: Path):
|
||||
"""Service wraps provider failures in TranscriptionError."""
|
||||
prompt_dir = tmp_path / "prompts"
|
||||
prompt_dir.mkdir()
|
||||
(prompt_dir / "transcribe_document.md").write_text("Prompt body", encoding="utf-8")
|
||||
|
||||
image_path = tmp_path / "document.png"
|
||||
image_path.write_bytes(b"png-bytes")
|
||||
|
||||
settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
|
||||
provider = _FakeProvider(error=ProviderError("upstream failure"))
|
||||
|
||||
with pytest.raises(TranscriptionError) as exc_info:
|
||||
transcribe_document_image(image_path, settings=settings, provider=provider)
|
||||
|
||||
assert exc_info.value.category.value == "external_provider_error"
|
||||
assert exc_info.value.retriable is True
|
||||
assert "retry" in exc_info.value.suggestion.lower()
|
||||
@@ -1,104 +0,0 @@
|
||||
"""Tests for transcription.services.upload."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from transcription.config import Settings
|
||||
from transcription.models import Document, Job, JobStatus
|
||||
from transcription.services.upload import UploadError, create_upload_job
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUploadValidation:
|
||||
"""Verify upload validation behavior."""
|
||||
|
||||
def test_rejects_empty_bytes(self, session, tmp_path: Path):
|
||||
"""create_upload_job rejects an empty upload payload."""
|
||||
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
||||
with pytest.raises(UploadError) as exc_info:
|
||||
create_upload_job(
|
||||
filename="letter.jpg",
|
||||
file_bytes=b"",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
assert exc_info.value.category.value == "validation_error"
|
||||
assert "non-empty" in exc_info.value.suggestion.lower()
|
||||
|
||||
def test_rejects_unsupported_extension(self, session, tmp_path: Path):
|
||||
"""create_upload_job rejects unsupported filename extensions."""
|
||||
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
||||
with pytest.raises(UploadError) as exc_info:
|
||||
create_upload_job(
|
||||
filename="notes.txt",
|
||||
file_bytes=b"content",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
assert exc_info.value.category.value == "user_input_error"
|
||||
assert "jpg" in exc_info.value.suggestion.lower()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestUploadPersistence:
|
||||
"""Verify upload file and record persistence behavior."""
|
||||
|
||||
def test_writes_file_and_creates_records(self, session, tmp_path: Path):
|
||||
"""create_upload_job writes file and creates document/job records."""
|
||||
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
||||
|
||||
result = create_upload_job(
|
||||
filename="letter.jpg",
|
||||
file_bytes=b"image-bytes",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
assert result.stored_path.exists()
|
||||
assert result.stored_path.read_bytes() == b"image-bytes"
|
||||
|
||||
document = session.get(Document, result.document_id)
|
||||
job = session.get(Job, result.job_id)
|
||||
assert document is not None
|
||||
assert job is not None
|
||||
assert document.filename == "letter.jpg"
|
||||
assert document.file_path == str(result.stored_path)
|
||||
|
||||
def test_uses_unique_stored_filename(self, session, tmp_path: Path):
|
||||
"""create_upload_job stores uploads with unique filenames."""
|
||||
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
||||
|
||||
first = create_upload_job(
|
||||
filename="duplicate.jpg",
|
||||
file_bytes=b"first",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
second = create_upload_job(
|
||||
filename="duplicate.jpg",
|
||||
file_bytes=b"second",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
assert first.stored_path != second.stored_path
|
||||
assert first.stored_path.exists()
|
||||
assert second.stored_path.exists()
|
||||
|
||||
def test_sets_job_status_queued(self, session, tmp_path: Path):
|
||||
"""create_upload_job persists a job with queued status."""
|
||||
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
|
||||
|
||||
result = create_upload_job(
|
||||
filename="queued.pdf",
|
||||
file_bytes=b"%PDF-1.4",
|
||||
session=session,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
job = session.get(Job, result.job_id)
|
||||
assert job is not None
|
||||
assert job.status == JobStatus.QUEUED
|
||||
@@ -1,232 +0,0 @@
|
||||
"""Tests for transcription.worker."""
|
||||
|
||||
from pathlib import Path
|
||||
from threading import Event
|
||||
|
||||
import pytest
|
||||
from sqlmodel import select
|
||||
|
||||
from transcription.config import Settings
|
||||
from transcription.errors import AppError, ErrorCategory
|
||||
from transcription.models import Document, Job, JobStatus, Transcript
|
||||
from transcription.providers.base import TranscriptionResult
|
||||
from transcription.worker import process_next_queued_job, run_worker_loop
|
||||
|
||||
|
||||
def _create_queued_job(session, *, filename: str = "doc.jpg", file_path: str = "uploads/doc.jpg") -> Job:
|
||||
document = Document(filename=filename, file_path=file_path)
|
||||
session.add(document)
|
||||
session.commit()
|
||||
session.refresh(document)
|
||||
|
||||
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
||||
session.add(job)
|
||||
session.commit()
|
||||
session.refresh(job)
|
||||
return job
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestWorkerQueueBehavior:
|
||||
"""Verify worker behavior when selecting queued jobs."""
|
||||
|
||||
def test_returns_false_when_queue_empty(self, session):
|
||||
"""process_next_queued_job returns False when there are no queued jobs."""
|
||||
processed = process_next_queued_job(session=session)
|
||||
assert processed is False
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestWorkerSuccessPath:
|
||||
"""Verify worker success-path lifecycle transitions and transcript persistence."""
|
||||
|
||||
def test_transitions_processing_to_transcribed(self, session, monkeypatch, tmp_path: Path):
|
||||
"""process_next_queued_job transitions queued jobs to transcribed on success."""
|
||||
job = _create_queued_job(session)
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
return TranscriptionResult(text="ok", provider="openrouter", model="test-model")
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
|
||||
)
|
||||
|
||||
processed = process_next_queued_job(session=session)
|
||||
session.refresh(job)
|
||||
|
||||
assert processed is True
|
||||
assert job.status == JobStatus.TRANSCRIBED
|
||||
|
||||
def test_persists_transcript_text_on_success(self, session, monkeypatch, tmp_path: Path):
|
||||
"""process_next_queued_job stores transcript text for successful jobs."""
|
||||
job = _create_queued_job(session)
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
return TranscriptionResult(text="Transcript body", provider="openrouter", model="test-model")
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
|
||||
)
|
||||
|
||||
process_next_queued_job(session=session)
|
||||
|
||||
transcript = session.exec(
|
||||
select(Transcript).where(Transcript.job_id == job.id)
|
||||
).first()
|
||||
assert transcript is not None
|
||||
assert transcript.text == "Transcript body"
|
||||
assert transcript.error_detail is None
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestWorkerFailurePath:
|
||||
"""Verify worker failure-path lifecycle transitions and error persistence."""
|
||||
|
||||
def test_sets_failed_and_error_detail_on_failure(self, session, monkeypatch, tmp_path: Path):
|
||||
"""process_next_queued_job marks failed and stores error detail on exception."""
|
||||
job = _create_queued_job(session)
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
raise RuntimeError("provider failure")
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
|
||||
)
|
||||
|
||||
processed = process_next_queued_job(session=session)
|
||||
session.refresh(job)
|
||||
transcript = session.exec(
|
||||
select(Transcript).where(Transcript.job_id == job.id)
|
||||
).first()
|
||||
|
||||
assert processed is True
|
||||
assert job.status == JobStatus.FAILED
|
||||
assert transcript is not None
|
||||
assert transcript.text is None
|
||||
assert "provider failure" in transcript.error_detail
|
||||
assert "[internal_unexpected_error]" in transcript.error_detail
|
||||
assert "error_id=" in transcript.error_detail
|
||||
assert "suggestion=" in transcript.error_detail
|
||||
|
||||
def test_updates_existing_transcript_if_present(self, session, monkeypatch, tmp_path: Path):
|
||||
"""process_next_queued_job updates existing transcript instead of duplicating."""
|
||||
job = _create_queued_job(session)
|
||||
existing = Transcript(job_id=job.id, text="old", error_detail=None)
|
||||
session.add(existing)
|
||||
session.commit()
|
||||
session.refresh(existing)
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
raise RuntimeError("provider failure")
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(openrouter_api_key="test-key", prompt_dir=tmp_path),
|
||||
)
|
||||
|
||||
process_next_queued_job(session=session)
|
||||
|
||||
transcripts = session.exec(
|
||||
select(Transcript).where(Transcript.job_id == job.id)
|
||||
).all()
|
||||
assert len(transcripts) == 1
|
||||
assert transcripts[0].id == existing.id
|
||||
assert transcripts[0].text is None
|
||||
assert "provider failure" in transcripts[0].error_detail
|
||||
assert "[internal_unexpected_error]" in transcripts[0].error_detail
|
||||
assert "error_id=" in transcripts[0].error_detail
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestWorkerRetryBehavior:
|
||||
"""Verify worker retry and terminal failure policies."""
|
||||
|
||||
def test_retriable_failure_requeues_until_limit(self, session, monkeypatch, tmp_path: Path):
|
||||
"""Retriable failures requeue jobs while retry budget remains."""
|
||||
job = _create_queued_job(session)
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
raise AppError(
|
||||
"temporary upstream outage",
|
||||
category=ErrorCategory.EXTERNAL_PROVIDER,
|
||||
suggestion="Retry from jobs page.",
|
||||
retriable=True,
|
||||
)
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(
|
||||
openrouter_api_key="test-key",
|
||||
prompt_dir=tmp_path,
|
||||
worker_max_retries=1,
|
||||
worker_retry_backoff_seconds=0.0,
|
||||
),
|
||||
)
|
||||
|
||||
processed = process_next_queued_job(session=session)
|
||||
session.refresh(job)
|
||||
|
||||
assert processed is True
|
||||
assert job.status == JobStatus.QUEUED
|
||||
assert job.retry_count == 1
|
||||
|
||||
def test_retriable_failure_exhaustion_sets_failed(self, session, monkeypatch, tmp_path: Path):
|
||||
"""Retriable failures transition to failed when retry budget is exhausted."""
|
||||
job = _create_queued_job(session)
|
||||
job.retry_count = 1
|
||||
session.add(job)
|
||||
session.commit()
|
||||
|
||||
def _fake_transcribe(_path):
|
||||
raise AppError(
|
||||
"temporary upstream outage",
|
||||
category=ErrorCategory.EXTERNAL_PROVIDER,
|
||||
suggestion="Retry from jobs page.",
|
||||
retriable=True,
|
||||
)
|
||||
|
||||
monkeypatch.setattr("transcription.worker.transcribe_document_image", _fake_transcribe)
|
||||
monkeypatch.setattr(
|
||||
"transcription.worker.get_settings",
|
||||
lambda: Settings(
|
||||
openrouter_api_key="test-key",
|
||||
prompt_dir=tmp_path,
|
||||
worker_max_retries=1,
|
||||
worker_retry_backoff_seconds=0.0,
|
||||
),
|
||||
)
|
||||
|
||||
process_next_queued_job(session=session)
|
||||
session.refresh(job)
|
||||
|
||||
assert job.status == JobStatus.FAILED
|
||||
assert job.retry_count == 1
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestWorkerLoopControl:
|
||||
"""Verify worker loop start/stop behavior."""
|
||||
|
||||
def test_stops_when_stop_event_is_set(self, monkeypatch):
|
||||
"""run_worker_loop exits when a stop event is set."""
|
||||
stop_event = Event()
|
||||
stop_event.set()
|
||||
|
||||
called = {"value": False}
|
||||
|
||||
def _fake_process_next_queued_job(**_kwargs):
|
||||
called["value"] = True
|
||||
return False
|
||||
|
||||
monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
|
||||
|
||||
run_worker_loop(stop_event=stop_event, poll_interval_seconds=0.01)
|
||||
assert called["value"] is False
|
||||
Reference in New Issue
Block a user