generated from john/python-template
92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
"""Tests for source revision behavior in TranscriptionService."""
|
|
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
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.jobs import JobService
|
|
from transcription.services.transcription import TranscriptionService
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestTranscriptionServiceRevisionUpsert:
|
|
"""Verify page-level source revision semantics."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_revision_creates_new_revision(self, default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = TranscriptionService(session_factory=default_session_factory)
|
|
|
|
document = Document(id=uuid4(), name="revision-create")
|
|
await documents.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
|
|
await jobs.create_job(job=job)
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="source.jpg",
|
|
filename="source.jpg",
|
|
file_path="uploads/source.jpg",
|
|
)
|
|
async with transcriptions._session_scope() as session:
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
|
|
await session.commit()
|
|
await session.refresh(source)
|
|
|
|
revision = await transcriptions.upsert_revision_for_source(source_id=source.id, text="User revision")
|
|
fetched = await transcriptions.read_revision_by_source(source.id)
|
|
|
|
assert revision.id == source.id
|
|
assert revision.revised_text == "User revision"
|
|
assert fetched is not None
|
|
assert fetched.id == source.id
|
|
assert fetched.revised_text == "User revision"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_revision_updates_existing_single_revision(self, default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = TranscriptionService(session_factory=default_session_factory)
|
|
|
|
document = Document(id=uuid4(), name="revision-update")
|
|
await documents.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
|
|
await jobs.create_job(job=job)
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="source.jpg",
|
|
filename="source.jpg",
|
|
file_path="uploads/source.jpg",
|
|
)
|
|
async with transcriptions._session_scope() as session:
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
|
|
await session.commit()
|
|
await session.refresh(source)
|
|
|
|
first = await transcriptions.upsert_revision_for_source(source_id=source.id, text="Revision v1")
|
|
second = await transcriptions.upsert_revision_for_source(source_id=source.id, text="Revision v2")
|
|
revisions = await transcriptions.list_revisions_by_job(job.id)
|
|
|
|
assert first.id == second.id
|
|
assert second.revised_text == "Revision v2"
|
|
assert len(revisions) == 1
|
|
assert revisions[0].id == first.id
|
|
assert revisions[0].revised_text == "Revision v2"
|