diff --git a/tests/services/test_transcription.py b/tests/services/test_transcription.py deleted file mode 100644 index c98059a..0000000 --- a/tests/services/test_transcription.py +++ /dev/null @@ -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() diff --git a/tests/services/test_upload.py b/tests/services/test_upload.py deleted file mode 100644 index 854d4d5..0000000 --- a/tests/services/test_upload.py +++ /dev/null @@ -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 diff --git a/tests/services/test_worker.py b/tests/services/test_worker.py deleted file mode 100644 index c27a261..0000000 --- a/tests/services/test_worker.py +++ /dev/null @@ -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