generated from john/python-template
585 lines
21 KiB
Python
585 lines
21 KiB
Python
import asyncio
|
|
from datetime import UTC
|
|
from datetime import datetime
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import event
|
|
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import JobStatus
|
|
from transcription.db.models import Source
|
|
from transcription.services.documents import DocumentService
|
|
from transcription.services.evidence import EvidenceService
|
|
from transcription.services.jobs import JobCancelBlockedError
|
|
from transcription.services.jobs import JobDeleteBlockedError
|
|
from transcription.services.jobs import JobNotFoundError
|
|
from transcription.services.jobs import JobResubmitBlockedError
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.sources import SourceService
|
|
|
|
|
|
class TestJobService:
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_read_job(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="test-bundle")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
fetched = await job_service.read_job(job_id=job.id)
|
|
assert fetched.id == job.id
|
|
assert fetched.document is not None
|
|
assert fetched.document.id == document.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_job_state_updates_status_and_retry(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="test-bundle")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
updated = await job_service.update_job_state(
|
|
job_id=job.id,
|
|
status=JobStatus.PROCESSING,
|
|
retry_count_increment=1,
|
|
)
|
|
|
|
assert updated.status == JobStatus.PROCESSING
|
|
assert updated.retry_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_jobs_by_status(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="query-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
await job_service.create_job(job=Job(document_id=document.id, status=JobStatus.PROCESSING))
|
|
await job_service.create_job(job=Job(document_id=document.id, status=JobStatus.QUEUED))
|
|
|
|
result = await job_service.query_jobs(status=JobStatus.PROCESSING)
|
|
assert len(result) == 1
|
|
assert result[0].status == JobStatus.PROCESSING
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_jobs_by_source_filename(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="source-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="letter.jpg",
|
|
filename="stored-letter.jpg",
|
|
file_path="/uploads/stored-letter.jpg",
|
|
file_hash="a" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
result = await job_service.query_jobs(filename="stored-letter.jpg")
|
|
assert len(result) == 1
|
|
assert result[0].id == job.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_queued_job_claims_oldest_and_marks_processing(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="ordered-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
created_at = datetime.now(UTC)
|
|
first = Job(document_id=document.id, status=JobStatus.QUEUED, date_created=created_at)
|
|
second = Job(
|
|
document_id=document.id,
|
|
status=JobStatus.QUEUED,
|
|
date_created=created_at + timedelta(microseconds=1),
|
|
)
|
|
await job_service.create_job(job=first)
|
|
await job_service.create_job(job=second)
|
|
|
|
claimed = await job_service.claim_next_queued_job()
|
|
assert claimed is not None
|
|
assert claimed.id == first.id
|
|
assert claimed.status == JobStatus.PROCESSING
|
|
|
|
# The claim is exclusive: the same job is never handed out twice.
|
|
next_claim = await job_service.claim_next_queued_job()
|
|
assert next_claim is not None
|
|
assert next_claim.id == second.id
|
|
|
|
assert await job_service.claim_next_queued_job() is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_queued_job_emits_a_bounded_unadorned_query(
|
|
self,
|
|
job_service: JobService,
|
|
):
|
|
"""CRIT-01: the hot poll must not select a subgraph or scan the queue."""
|
|
statements: list[str] = []
|
|
|
|
async with job_service._session_scope() as session:
|
|
bind = session.get_bind()
|
|
|
|
def capture(_conn, _cursor, statement, *_rest):
|
|
statements.append(statement)
|
|
|
|
event.listen(bind, "before_cursor_execute", capture)
|
|
try:
|
|
await job_service.claim_next_queued_job(session=session)
|
|
finally:
|
|
event.remove(bind, "before_cursor_execute", capture)
|
|
|
|
claim_sql = [
|
|
item
|
|
for item in statements
|
|
if item.lstrip().upper().startswith(("SELECT", "UPDATE"))
|
|
]
|
|
assert len(claim_sql) == 1, claim_sql
|
|
assert "LIMIT" in claim_sql[0].upper()
|
|
assert "JOIN" not in claim_sql[0].upper()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_queued_job_is_atomic_on_sqlite(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = await document_service.create_document(Document(id=uuid4(), name="atomic-claim-doc"))
|
|
queued_job = await job_service.create_job(Job(document_id=document.id, status=JobStatus.QUEUED))
|
|
|
|
async def claim_once():
|
|
claimed = await job_service.claim_next_queued_job()
|
|
return claimed.id if claimed is not None else None
|
|
|
|
first_claim, second_claim = await asyncio.gather(claim_once(), claim_once())
|
|
|
|
assert [first_claim, second_claim].count(queued_job.id) == 1
|
|
assert [first_claim, second_claim].count(None) == 1
|
|
assert await job_service.claim_next_queued_job() is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_job_persists_provider_and_model(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="provider-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(
|
|
document_id=document.id,
|
|
provider="openrouter",
|
|
model="google/gemini-2.5-flash",
|
|
)
|
|
await job_service.create_job(job=job)
|
|
|
|
fetched = await job_service.read_job(job_id=job.id)
|
|
assert fetched.provider == "openrouter"
|
|
assert fetched.model == "google/gemini-2.5-flash"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_job_resolves_filename_from_linked_source(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="filename-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="page_001.png",
|
|
filename="stored_page_001.png",
|
|
file_path="/uploads/stored_page_001.png",
|
|
file_hash="b" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
fetched = await job_service.read_job(job_id=job.id)
|
|
assert fetched.filename == "stored_page_001.png"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_job_with_guardrails_blocks_processing_jobs(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="processing-delete-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.PROCESSING)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobDeleteBlockedError):
|
|
await job_service.delete_job_with_guardrails(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_job_with_guardrails_removes_jobsource_links(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="delete-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="delete-job-source.jpg",
|
|
filename="stored-delete-job-source.jpg",
|
|
file_path="/uploads/stored-delete-job-source.jpg",
|
|
file_hash="c" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
await job_service.delete_job_with_guardrails(job_id=job.id)
|
|
|
|
with pytest.raises(JobNotFoundError):
|
|
await job_service.read_job(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_job_and_evidence_removes_attempts_but_preserves_source(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
source_service = SourceService(session_factory=job_service.session_factory)
|
|
evidence_service = EvidenceService(session_factory=job_service.session_factory)
|
|
document = await document_service.create_document(Document(name="evidence-delete-doc"))
|
|
job = await job_service.create_job(Job(document_id=document.id, status=JobStatus.FAILED))
|
|
source = await source_service.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="evidence.jpg",
|
|
filename="evidence.jpg",
|
|
file_path="/uploads/evidence.jpg",
|
|
file_hash="d" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
await source_service.create_job_source(JobSource(job_id=job.id, source_id=source.id))
|
|
await source_service.update_job_source_transcription(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
text=None,
|
|
error_detail="fixture failure",
|
|
)
|
|
|
|
await job_service.delete_job_and_evidence(job_id=job.id)
|
|
|
|
with pytest.raises(JobNotFoundError):
|
|
await job_service.read_job(job_id=job.id)
|
|
assert await evidence_service.list_execution_attempts(job_id=job.id) == []
|
|
assert (await source_service.read_source(source.id)).id == source.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_job_marks_non_transcribed_sources_failed(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="cancel-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="cancel-1.jpg",
|
|
filename="stored-cancel-1.jpg",
|
|
file_path="/uploads/stored-cancel-1.jpg",
|
|
file_hash="d" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
source_two = Source(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
upload_name="cancel-2.jpg",
|
|
filename="stored-cancel-2.jpg",
|
|
file_path="/uploads/stored-cancel-2.jpg",
|
|
file_hash="e" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source_one)
|
|
session.add(source_two)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
cancelled = await job_service.cancel_job(job_id=job.id)
|
|
assert cancelled.status == JobStatus.FAILED
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
statuses = {item.status for item in refreshed.job_sources}
|
|
assert JobSourceStatus.TRANSCRIBED in statuses
|
|
assert JobSourceStatus.CANCELLED in statuses
|
|
assert JobSourceStatus.FAILED not in statuses
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_resets_only_failed(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="resubmit-1.jpg",
|
|
filename="stored-resubmit-1.jpg",
|
|
file_path="/uploads/stored-resubmit-1.jpg",
|
|
file_hash="f" * 64,
|
|
file_size_bytes=1,
|
|
raw_transcription="existing text",
|
|
)
|
|
source_two = Source(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
upload_name="resubmit-2.jpg",
|
|
filename="stored-resubmit-2.jpg",
|
|
file_path="/uploads/stored-resubmit-2.jpg",
|
|
file_hash="0" * 64,
|
|
file_size_bytes=1,
|
|
raw_transcription="done text",
|
|
)
|
|
session.add(source_one)
|
|
session.add(source_two)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.FAILED,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
count = await job_service.resubmit_failed_sources(job_id=job.id)
|
|
assert count == 1
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
assert refreshed.status == JobStatus.QUEUED
|
|
|
|
failed_entry = next(
|
|
item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 1
|
|
)
|
|
transcribed_entry = next(
|
|
item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 2
|
|
)
|
|
assert failed_entry.status == JobSourceStatus.PENDING
|
|
assert failed_entry.source is not None
|
|
assert failed_entry.source.raw_transcription == "existing text"
|
|
assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_blocks_when_only_pending_or_transcribed(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-no-failed-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="resubmit-pending.jpg",
|
|
filename="stored-resubmit-pending.jpg",
|
|
file_path="/uploads/stored-resubmit-pending.jpg",
|
|
file_hash="1" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
source_two = Source(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
upload_name="resubmit-done.jpg",
|
|
filename="stored-resubmit-done.jpg",
|
|
file_path="/uploads/stored-resubmit-done.jpg",
|
|
file_hash="2" * 64,
|
|
file_size_bytes=1,
|
|
raw_transcription="done text",
|
|
)
|
|
session.add(source_one)
|
|
session.add(source_two)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
with pytest.raises(JobResubmitBlockedError):
|
|
await job_service.resubmit_failed_sources(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_includes_cancelled(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
"""Cancel is recoverable: cancelled pages are re-attempted on resubmit."""
|
|
document = Document(id=uuid4(), name="resubmit-cancelled-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="resubmit-cancelled.jpg",
|
|
filename="stored-resubmit-cancelled.jpg",
|
|
file_path="/uploads/stored-resubmit-cancelled.jpg",
|
|
file_hash="3" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.CANCELLED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
assert await job_service.resubmit_failed_sources(job_id=job.id) == 1
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
assert refreshed.status == JobStatus.QUEUED
|
|
assert refreshed.job_sources[0].status == JobSourceStatus.PENDING
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_blocks_when_processing(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-blocked-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.PROCESSING)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobResubmitBlockedError):
|
|
await job_service.resubmit_failed_sources(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_job_blocks_transcribed_terminal_jobs(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="cancel-blocked-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobCancelBlockedError):
|
|
await job_service.cancel_job(job_id=job.id)
|