generated from john/python-template
230 lines
8.4 KiB
Python
230 lines
8.4 KiB
Python
"""Retranscription candidate and promotion tests."""
|
|
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobPurpose
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import Source
|
|
from transcription.services import ServiceBundle
|
|
from transcription.services.errors import CandidatePromotionError
|
|
from transcription.services.workflows import create_source_retranscription_job
|
|
|
|
|
|
def _services(default_session_factory, settings: Settings) -> ServiceBundle:
|
|
return ServiceBundle.from_session_factory(default_session_factory, settings=settings)
|
|
|
|
|
|
async def _seed_source(services: ServiceBundle) -> Source:
|
|
document = await services.documents.create_document(Document(name="candidate source"))
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="page.jpg",
|
|
filename="page.jpg",
|
|
file_path="page.jpg",
|
|
file_hash="a" * 64,
|
|
file_size_bytes=1,
|
|
revised_text="human revision",
|
|
)
|
|
return await services.sources.create_source(source)
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_first_success_is_preferred_and_later_success_remains_candidate(default_session_factory):
|
|
settings = Settings(openrouter_api_key="test-key", provider_models=None)
|
|
services = _services(default_session_factory, settings)
|
|
source = await _seed_source(services)
|
|
|
|
first_job = await services.jobs.create_job(Job(document_id=source.document_id))
|
|
second_job = await services.jobs.create_job(Job(document_id=source.document_id))
|
|
await services.sources.create_job_source(JobSource(job_id=first_job.id, source_id=source.id))
|
|
await services.sources.create_job_source(JobSource(job_id=second_job.id, source_id=source.id))
|
|
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=first_job.id,
|
|
source_id=source.id,
|
|
text="first result",
|
|
provider="fixture",
|
|
model="model-a",
|
|
)
|
|
selected = await services.sources.read_source(source.id)
|
|
first_attempt_id = selected.preferred_execution_attempt_id
|
|
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=second_job.id,
|
|
source_id=source.id,
|
|
text="candidate result",
|
|
provider="fixture",
|
|
model="model-b",
|
|
)
|
|
unchanged = await services.sources.read_source(source.id)
|
|
attempts = await services.evidence.list_execution_attempts(source_id=source.id)
|
|
|
|
assert unchanged.raw_transcription == "first result"
|
|
assert unchanged.preferred_execution_attempt_id == first_attempt_id
|
|
assert {attempt.raw_transcription for attempt in attempts} == {"first result", "candidate result"}
|
|
|
|
candidate = next(attempt for attempt in attempts if attempt.raw_transcription == "candidate result")
|
|
promoted = await services.evidence.promote_machine_attempt(
|
|
source_id=source.id,
|
|
execution_attempt_id=candidate.id,
|
|
)
|
|
assert promoted.raw_transcription == "candidate result"
|
|
assert promoted.preferred_execution_attempt_id == candidate.id
|
|
assert promoted.revised_text == "human revision"
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_promotion_rejects_unrelated_attempt(default_session_factory):
|
|
settings = Settings(openrouter_api_key="test-key", provider_models=None)
|
|
services = _services(default_session_factory, settings)
|
|
source = await _seed_source(services)
|
|
|
|
with pytest.raises(CandidatePromotionError) as exc_info:
|
|
await services.evidence.promote_machine_attempt(
|
|
source_id=source.id,
|
|
execution_attempt_id=uuid4(),
|
|
)
|
|
error = exc_info.value
|
|
assert error.category.value == "validation_error"
|
|
assert "successful transcription attempt" in error.message
|
|
assert "Select an available successful candidate" in error.suggestion
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_retranscription_job_locks_source_and_frozen_model(default_session_factory, tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
prompt_dir=tmp_path,
|
|
provider_model="vendor/default",
|
|
provider_models=["vendor/alternate"],
|
|
)
|
|
(tmp_path / settings.default_prompt_name).write_text("Transcribe verbatim.", encoding="utf-8")
|
|
services = _services(default_session_factory, settings)
|
|
source = await _seed_source(services)
|
|
|
|
job = await create_source_retranscription_job(
|
|
source_id=source.id,
|
|
model="vendor/alternate",
|
|
services=services,
|
|
settings=settings,
|
|
)
|
|
loaded = await services.jobs.read_job(job.id)
|
|
|
|
assert loaded.purpose == JobPurpose.RETRANSCRIPTION
|
|
assert loaded.document_id == source.document_id
|
|
assert loaded.provider == "openrouter"
|
|
assert loaded.model == "vendor/alternate"
|
|
assert loaded.user_prompt == "Transcribe verbatim."
|
|
assert [link.source_id for link in loaded.job_sources] == [source.id]
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_promoting_candidate_does_not_mutate_execution_attempt_history(default_session_factory):
|
|
settings = Settings(openrouter_api_key="test-key", provider_models=None)
|
|
services = _services(default_session_factory, settings)
|
|
source = await _seed_source(services)
|
|
|
|
first_job = await services.jobs.create_job(Job(document_id=source.document_id))
|
|
second_job = await services.jobs.create_job(Job(document_id=source.document_id))
|
|
await services.sources.create_job_source(JobSource(job_id=first_job.id, source_id=source.id))
|
|
await services.sources.create_job_source(JobSource(job_id=second_job.id, source_id=source.id))
|
|
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=first_job.id,
|
|
source_id=source.id,
|
|
text="baseline",
|
|
provider="fixture",
|
|
model="model-a",
|
|
)
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=second_job.id,
|
|
source_id=source.id,
|
|
text="candidate",
|
|
provider="fixture",
|
|
model="model-b",
|
|
)
|
|
|
|
before = [
|
|
(
|
|
attempt.id,
|
|
attempt.job_id,
|
|
attempt.source_id,
|
|
attempt.attempt_number,
|
|
attempt.status.value,
|
|
attempt.raw_transcription,
|
|
attempt.error_category,
|
|
attempt.error_detail,
|
|
attempt.failure_phase,
|
|
)
|
|
for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
|
|
]
|
|
|
|
candidate = next(
|
|
attempt
|
|
for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
|
|
if attempt.raw_transcription == "candidate"
|
|
)
|
|
await services.evidence.promote_machine_attempt(source_id=source.id, execution_attempt_id=candidate.id)
|
|
|
|
after = [
|
|
(
|
|
attempt.id,
|
|
attempt.job_id,
|
|
attempt.source_id,
|
|
attempt.attempt_number,
|
|
attempt.status.value,
|
|
attempt.raw_transcription,
|
|
attempt.error_category,
|
|
attempt.error_detail,
|
|
attempt.failure_phase,
|
|
)
|
|
for attempt in await services.evidence.list_execution_attempts(source_id=source.id)
|
|
]
|
|
|
|
assert after == before
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_retry_appends_new_attempt_instead_of_rewriting_history(default_session_factory):
|
|
settings = Settings(openrouter_api_key="test-key", provider_models=None)
|
|
services = _services(default_session_factory, settings)
|
|
source = await _seed_source(services)
|
|
|
|
job = await services.jobs.create_job(Job(document_id=source.document_id))
|
|
await services.sources.create_job_source(JobSource(job_id=job.id, source_id=source.id))
|
|
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
text=None,
|
|
error_detail="timeout",
|
|
error_category="timeout",
|
|
provider="fixture",
|
|
model="model-a",
|
|
failure_phase="provider_call",
|
|
)
|
|
await services.sources.update_job_source_transcription(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
text="retry-success",
|
|
provider="fixture",
|
|
model="model-a",
|
|
)
|
|
|
|
attempts = list(await services.evidence.list_execution_attempts(source_id=source.id))
|
|
assert len(attempts) == 2
|
|
assert [attempt.attempt_number for attempt in attempts] == [1, 2]
|
|
assert attempts[0].raw_transcription is None
|
|
assert attempts[1].raw_transcription == "retry-success"
|