generated from john/python-template
312 lines
10 KiB
Python
312 lines
10 KiB
Python
"""Tests for the V2 SQLModel persistence layer and relationships."""
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentPerson
|
|
from transcription.db.models import DocumentType
|
|
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 Person
|
|
from transcription.db.models import PersonRole
|
|
from transcription.db.models import Source
|
|
|
|
|
|
def _make_document(**overrides: Any) -> Document:
|
|
defaults: dict[str, Any] = {
|
|
"name": "letter bundle",
|
|
"notes": "Family correspondence",
|
|
}
|
|
defaults.update(overrides)
|
|
return Document(**defaults)
|
|
|
|
|
|
def _persist_document_type(session, *, label: str = "Letter") -> DocumentType:
|
|
document_type = DocumentType(label=label, normalized_label=label.strip().casefold())
|
|
session.add(document_type)
|
|
session.commit()
|
|
session.refresh(document_type)
|
|
return document_type
|
|
|
|
|
|
def _persist_person_role(session, *, label: str = "Author") -> PersonRole:
|
|
role = PersonRole(label=label, normalized_label=label.strip().casefold())
|
|
session.add(role)
|
|
session.commit()
|
|
session.refresh(role)
|
|
return role
|
|
|
|
|
|
def _persist_document(session) -> Document:
|
|
document_type = _persist_document_type(session)
|
|
document = _make_document(document_type_id=document_type.id)
|
|
session.add(document)
|
|
session.commit()
|
|
session.refresh(document)
|
|
return document
|
|
|
|
|
|
def _persist_person(session, **overrides: Any) -> Person:
|
|
defaults: dict[str, Any] = {"given_names": "Ada", "last_name": "Lovelace"}
|
|
defaults.update(overrides)
|
|
person = Person(**defaults)
|
|
session.add(person)
|
|
session.commit()
|
|
session.refresh(person)
|
|
return person
|
|
|
|
|
|
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, *, page_number: int = 1, **overrides: Any) -> Source:
|
|
defaults: dict[str, Any] = {
|
|
"document_id": document.id,
|
|
"page_number": page_number,
|
|
"upload_name": "letter.jpg",
|
|
"filename": "stored-letter.jpg",
|
|
"file_path": "/uploads/stored-letter.jpg",
|
|
"raw_transcription": "Original machine text",
|
|
}
|
|
defaults.update(overrides)
|
|
defaults["file_hash"] = "a" * 64
|
|
defaults["file_size_bytes"] = 123
|
|
source = Source(**defaults)
|
|
session.add(source)
|
|
session.commit()
|
|
session.refresh(source)
|
|
return source
|
|
|
|
|
|
def _persist_job_source(session, job: Job, source: Source, **overrides: Any) -> JobSource:
|
|
defaults: dict[str, Any] = {
|
|
"job_id": job.id,
|
|
"source_id": source.id,
|
|
"status": JobSourceStatus.PENDING,
|
|
}
|
|
defaults.update(overrides)
|
|
job_source = JobSource(**defaults)
|
|
session.add(job_source)
|
|
session.commit()
|
|
session.refresh(job_source)
|
|
return job_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)
|
|
assert document.created_at is not None
|
|
assert document.updated_at is not None
|
|
assert document.created_at.tzinfo is None
|
|
assert document.updated_at.tzinfo is None
|
|
|
|
def test_can_reference_document_type_registry(self, session):
|
|
document_type = _persist_document_type(session, label="Record")
|
|
document = _make_document(document_type_id=document_type.id)
|
|
session.add(document)
|
|
session.commit()
|
|
session.refresh(document)
|
|
assert document.document_type_id == document_type.id
|
|
|
|
|
|
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
|
|
assert job.date_created.tzinfo is None
|
|
assert job.date_updated.tzinfo is 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
|
|
|
|
def test_status_enums_match_v4_lifecycle_contract(self):
|
|
assert [status.value for status in JobStatus] == [
|
|
"queued",
|
|
"processing",
|
|
"transcribed",
|
|
"partial_success",
|
|
"failed",
|
|
]
|
|
assert [status.value for status in JobSourceStatus] == [
|
|
"pending",
|
|
"transcribed",
|
|
"failed",
|
|
"cancelled",
|
|
]
|
|
|
|
|
|
class TestSourceModel:
|
|
def test_can_be_created_for_document(self, session):
|
|
document = _persist_document(session)
|
|
source = _persist_source(session, document)
|
|
|
|
fetched = session.get(Source, source.id)
|
|
assert fetched is not None
|
|
assert fetched.document_id == document.id
|
|
assert fetched.page_number == 1
|
|
assert fetched.date_uploaded is not None
|
|
assert fetched.date_uploaded.tzinfo is None
|
|
|
|
def test_revised_text_is_supported(self, session):
|
|
document = _persist_document(session)
|
|
source = _persist_source(session, document, revised_text="Edited output")
|
|
|
|
fetched = session.get(Source, source.id)
|
|
assert fetched is not None
|
|
assert fetched.revised_text == "Edited output"
|
|
|
|
|
|
class TestPersonAndDocumentPersonModel:
|
|
def test_family_search_id_is_unique_when_present(self, session):
|
|
session.add(Person(given_names="First", last_name="Person", family_search_id="G8T4-MDQ"))
|
|
session.commit()
|
|
|
|
session.add(Person(given_names="Second", last_name="Person", family_search_id="G8T4-MDQ"))
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
def test_document_person_role_is_unique_per_document_person(self, session):
|
|
document = _persist_document(session)
|
|
person = _persist_person(session)
|
|
person_role = _persist_person_role(session)
|
|
|
|
first = DocumentPerson(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role_id=person_role.id,
|
|
)
|
|
session.add(first)
|
|
session.commit()
|
|
|
|
duplicate = DocumentPerson(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role_id=person_role.id,
|
|
)
|
|
session.add(duplicate)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
|
|
class TestJobSourceModel:
|
|
def test_job_source_is_a_queue_row_not_an_evidence_row(self, session):
|
|
"""job_source carries only queue state; evidence lives on execution_attempt."""
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document)
|
|
job_source = _persist_job_source(session, job, source)
|
|
|
|
fetched = session.get(JobSource, job_source.id)
|
|
assert fetched is not None
|
|
assert fetched.status == JobSourceStatus.PENDING
|
|
assert set(JobSource.model_fields) == {"id", "job_id", "source_id", "status"}
|
|
|
|
def test_job_source_membership_is_unique_per_job_and_source(self, session):
|
|
document = _persist_document(session)
|
|
job = _persist_job(session, document)
|
|
source = _persist_source(session, document)
|
|
_persist_job_source(session, job, source)
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
|
|
class TestRelationships:
|
|
def test_document_exposes_jobs_sources_and_people(self, session):
|
|
document = _persist_document(session)
|
|
_persist_job(session, document)
|
|
_persist_source(session, document)
|
|
person = _persist_person(session)
|
|
person_role = _persist_person_role(session)
|
|
|
|
link = DocumentPerson(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role_id=person_role.id,
|
|
)
|
|
session.add(link)
|
|
session.commit()
|
|
|
|
session.refresh(document, attribute_names=["jobs", "sources", "document_people"])
|
|
assert len(document.jobs) == 1
|
|
assert len(document.sources) == 1
|
|
assert len(document.document_people) == 1
|
|
|
|
|
|
class TestRegistryModels:
|
|
def test_person_role_normalized_label_is_unique(self, session):
|
|
_persist_person_role(session, label="Mentioned")
|
|
duplicate = PersonRole(label=" mentioned ", normalized_label="mentioned")
|
|
session.add(duplicate)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
def test_document_type_normalized_label_is_unique(self, session):
|
|
_persist_document_type(session, label="Journal")
|
|
duplicate = DocumentType(label=" journal ", normalized_label="journal")
|
|
session.add(duplicate)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
|
|
def test_no_relationship_declares_an_implicit_eager_load():
|
|
"""CRIT-02 guard: eager loading is a per-query decision, never a model default."""
|
|
from sqlmodel import SQLModel
|
|
|
|
offenders = {
|
|
f"{mapper.class_.__name__}.{relationship.key}": relationship.lazy
|
|
for mapper in SQLModel._sa_registry.mappers
|
|
for relationship in mapper.relationships
|
|
if relationship.lazy not in {"raise", "noload"}
|
|
}
|
|
assert offenders == {}, f"Relationships must not preload by default: {offenders}"
|