generated from john/python-template
162 lines
4.9 KiB
Python
162 lines
4.9 KiB
Python
"""Tests for transcription.models — Document, Source, Job, Revision persistence and relationships."""
|
|
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from transcription.models import Document, Job, JobStatus, Revision, Source
|
|
|
|
|
|
def _make_document(**overrides) -> Document:
|
|
defaults = {"name": "letter bundle"}
|
|
defaults.update(overrides)
|
|
return Document(**defaults)
|
|
|
|
|
|
def _persist_document(session) -> Document:
|
|
document = _make_document()
|
|
session.add(document)
|
|
session.commit()
|
|
session.refresh(document)
|
|
return document
|
|
|
|
|
|
def _persist_job(session, document: Document) -> Job:
|
|
job = Job(document_id=document.id)
|
|
session.add(job)
|
|
session.commit()
|
|
session.refresh(job)
|
|
return job
|
|
|
|
|
|
def _persist_source(session, document: Document, job: Job, **overrides) -> Source:
|
|
defaults = {
|
|
"document_id": document.id,
|
|
"job_id": job.id,
|
|
"upload_name": "letter.jpg",
|
|
"filename": "stored-letter.jpg",
|
|
"file_path": "/uploads/stored-letter.jpg",
|
|
}
|
|
defaults.update(overrides)
|
|
source = Source(**defaults)
|
|
session.add(source)
|
|
session.commit()
|
|
session.refresh(source)
|
|
return source
|
|
|
|
|
|
class TestDocumentModel:
|
|
def test_can_be_persisted(self, session):
|
|
document = _persist_document(session)
|
|
fetched = session.get(Document, document.id)
|
|
assert fetched is not None
|
|
assert fetched.name == "letter bundle"
|
|
|
|
def test_defaults_are_populated(self, session):
|
|
document = _persist_document(session)
|
|
assert isinstance(document.id, UUID)
|
|
|
|
|
|
class TestJobModel:
|
|
def test_can_be_created_for_document(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
fetched = session.get(Job, job.id)
|
|
assert fetched is not None
|
|
assert fetched.document_id == document.id
|
|
|
|
def test_defaults_are_populated(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
assert job.status == JobStatus.QUEUED
|
|
assert job.retry_count == 0
|
|
assert job.date_created is not None
|
|
assert job.date_updated is not None
|
|
|
|
def test_transitions_to_transcribed(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
|
|
job.status = JobStatus.PROCESSING
|
|
session.add(job)
|
|
session.commit()
|
|
session.refresh(job)
|
|
|
|
job.status = JobStatus.TRANSCRIBED
|
|
session.add(job)
|
|
session.commit()
|
|
session.refresh(job)
|
|
|
|
assert job.status == JobStatus.TRANSCRIBED
|
|
|
|
|
|
class TestSourceModel:
|
|
def test_can_be_created_for_document_and_job(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document, job)
|
|
|
|
fetched = session.get(Source, source.id)
|
|
assert fetched is not None
|
|
assert fetched.document_id == document.id
|
|
assert fetched.job_id == job.id
|
|
assert fetched.date_uploaded is not None
|
|
|
|
|
|
class TestRevisionModel:
|
|
def test_revision_persists_for_source(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document, job)
|
|
|
|
revision = Revision(source_id=source.id, text="Edited revision text")
|
|
session.add(revision)
|
|
session.commit()
|
|
session.refresh(revision)
|
|
|
|
fetched = session.get(Revision, revision.id)
|
|
assert fetched is not None
|
|
assert fetched.text == "Edited revision text"
|
|
assert fetched.date_created is not None
|
|
|
|
def test_source_id_is_unique(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document, job)
|
|
|
|
first = Revision(source_id=source.id, text="First")
|
|
session.add(first)
|
|
session.commit()
|
|
|
|
duplicate = Revision(source_id=source.id, text="Duplicate")
|
|
session.add(duplicate)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
|
|
class TestRelationships:
|
|
def test_document_exposes_jobs_and_sources(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
_persist_source(session, document, job)
|
|
|
|
session.refresh(document)
|
|
assert len(document.jobs) == 1
|
|
assert len(document.sources) == 1
|
|
|
|
def test_source_exposes_optional_single_revision(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document, job)
|
|
|
|
assert source.revision is None
|
|
|
|
revision = Revision(source_id=source.id, text="Edited")
|
|
session.add(revision)
|
|
session.commit()
|
|
|
|
session.refresh(source)
|
|
assert source.revision is not None
|
|
assert source.revision.text == "Edited"
|