transcription/tests/test_models.py

"""Tests for transcription.models — Document, Source, Job, Revision persistence and relationships."""

from uuid import UUID

import pytest
from sqlalchemy.exc import IntegrityError

from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobStatus
from transcription.db.models import Revision
from transcription.db.models import 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"