generated from john/python-template
126 lines
4.4 KiB
Python
126 lines
4.4 KiB
Python
"""Tests for transcription.services.upload."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from transcription.config import Settings
|
|
from transcription.models import Document
|
|
from transcription.models import Job
|
|
from transcription.models import JobStatus
|
|
from transcription.services.upload import UploadError
|
|
from transcription.services.upload import 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()
|
|
|
|
def test_rejects_payload_exceeding_max_upload_bytes(self, session, tmp_path: Path):
|
|
"""create_upload_job rejects payloads above configured size limit."""
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
upload_dir=tmp_path,
|
|
max_upload_bytes=3,
|
|
)
|
|
with pytest.raises(UploadError) as exc_info:
|
|
create_upload_job(
|
|
filename="scan.jpg",
|
|
file_bytes=b"1234",
|
|
session=session,
|
|
settings=settings,
|
|
)
|
|
|
|
assert exc_info.value.category.value == "user_input_error"
|
|
assert "smaller file" 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
|