generated from john/python-template
91 lines
3.7 KiB
Python
91 lines
3.7 KiB
Python
"""Tests for Step 3 library services (revisions, search, export)."""
|
|
|
|
from sqlmodel import select
|
|
import pytest
|
|
|
|
from transcription.models import Document, Job, JobStatus, Transcript, TranscriptRevision
|
|
from transcription.services.library import accept_revision, add_revision, export_transcripts, list_revisions, search_accepted_transcripts
|
|
|
|
|
|
def _create_job(session) -> Job:
|
|
document = Document(filename="letter.jpg", file_path="uploads/letter.jpg")
|
|
session.add(document)
|
|
session.commit()
|
|
session.refresh(document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
|
|
session.add(job)
|
|
session.commit()
|
|
session.refresh(job)
|
|
return job
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestRevisionHistoryBehavior:
|
|
"""Verify revision append/accept behavior."""
|
|
|
|
def test_add_revision_appends_incrementing_revision_numbers(self, session):
|
|
"""add_revision creates immutable incrementing revisions per job."""
|
|
job = _create_job(session)
|
|
|
|
r1 = add_revision(job_id=job.id, text="first", session=session)
|
|
r2 = add_revision(job_id=job.id, text="second", session=session)
|
|
|
|
revisions = list_revisions(job_id=job.id, session=session)
|
|
|
|
assert r1.revision_number == 1
|
|
assert r2.revision_number == 2
|
|
assert [revision.revision_number for revision in revisions] == [1, 2]
|
|
|
|
def test_accept_revision_marks_selected_revision_and_sets_job_completed(self, session):
|
|
"""accept_revision marks one revision accepted and transitions job to completed."""
|
|
job = _create_job(session)
|
|
r1 = add_revision(job_id=job.id, text="v1", session=session)
|
|
r2 = add_revision(job_id=job.id, text="v2", session=session)
|
|
|
|
accepted = accept_revision(revision_id=r2.id, session=session)
|
|
session.refresh(job)
|
|
|
|
all_revisions = list(session.exec(select(TranscriptRevision).where(TranscriptRevision.job_id == job.id)).all())
|
|
accepted_flags = {revision.id: revision.accepted for revision in all_revisions}
|
|
transcript = session.exec(select(Transcript).where(Transcript.job_id == job.id)).first()
|
|
|
|
assert accepted.id == r2.id
|
|
assert accepted_flags[r1.id] is False
|
|
assert accepted_flags[r2.id] is True
|
|
assert job.status == JobStatus.COMPLETED
|
|
assert transcript is not None
|
|
assert transcript.text == "v2"
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestSearchAndExportBehavior:
|
|
"""Verify accepted-only search and export semantics."""
|
|
|
|
def test_search_returns_only_accepted_revisions(self, session):
|
|
"""search_accepted_transcripts filters out non-accepted revisions."""
|
|
job = _create_job(session)
|
|
draft = add_revision(job_id=job.id, text="family archive draft", session=session)
|
|
accepted = add_revision(job_id=job.id, text="family archive final", accepted=True, session=session)
|
|
|
|
results = search_accepted_transcripts(query="archive", session=session)
|
|
|
|
assert results
|
|
result_ids = {result.id for result in results}
|
|
assert accepted.id in result_ids
|
|
assert draft.id not in result_ids
|
|
|
|
def test_export_returns_serializable_records_for_accepted_revisions(self, session):
|
|
"""export_transcripts returns expected fields for accepted-only export."""
|
|
job = _create_job(session)
|
|
accepted = add_revision(job_id=job.id, text="export me", accepted=True, session=session)
|
|
|
|
records = export_transcripts(accepted_only=True, session=session)
|
|
|
|
assert len(records) == 1
|
|
record = records[0]
|
|
assert record["job_id"] == str(job.id)
|
|
assert record["revision_id"] == str(accepted.id)
|
|
assert record["accepted"] is True
|
|
assert record["text"] == "export me"
|