generated from john/python-template
174 lines
6.7 KiB
Python
174 lines
6.7 KiB
Python
from datetime import UTC
|
|
from datetime import datetime
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentPerson
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobStatus
|
|
from transcription.db.models import Person
|
|
from transcription.db.models import PersonRole
|
|
from transcription.db.models import Source
|
|
from transcription.services.documents import DocumentService
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.people import DocumentPersonInput
|
|
from transcription.services.people import PeopleError
|
|
from transcription.services.people import PeopleService
|
|
from transcription.services.sources import SourceService
|
|
from transcription.services.workflows import create_document_with_people
|
|
from transcription.services.workflows import update_document_with_people
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_document_with_people_rolls_back_on_invalid_person(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people = PeopleService(session_factory=default_session_factory)
|
|
role = await people.create_person_role(label="Witness")
|
|
|
|
with pytest.raises(PeopleError):
|
|
await create_document_with_people(
|
|
document=Document(name="Must roll back"),
|
|
links=[DocumentPersonInput(person_id=uuid4(), role_id=role.id)],
|
|
tag_labels=[],
|
|
documents=documents,
|
|
people=people,
|
|
)
|
|
|
|
assert await documents.query_documents(name="Must roll back") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_document_with_people_rolls_back_document_and_links(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people = PeopleService(session_factory=default_session_factory)
|
|
role = await people.create_person_role(label="Witness")
|
|
inactive = await people.create_person_role(label="Former Witness", is_active=False)
|
|
person = await people.create_person(Person(full_name="Archive Witness"))
|
|
document = await create_document_with_people(
|
|
document=Document(name="Original name"),
|
|
links=[DocumentPersonInput(person_id=person.id, role_id=role.id)],
|
|
tag_labels=[],
|
|
documents=documents,
|
|
people=people,
|
|
)
|
|
assert [item.name for item in await documents.list_documents()] == ["Original name"]
|
|
|
|
candidate = Document(
|
|
id=document.id,
|
|
name="Changed name",
|
|
created_at=document.created_at,
|
|
updated_at=document.updated_at,
|
|
)
|
|
with pytest.raises(PeopleError):
|
|
await update_document_with_people(
|
|
document=candidate,
|
|
links=[DocumentPersonInput(person_id=person.id, role_id=inactive.id)],
|
|
tag_labels=[],
|
|
documents=documents,
|
|
people=people,
|
|
)
|
|
|
|
persisted_documents = await documents.list_documents()
|
|
links = await people.list_document_people(document_id=document.id)
|
|
assert [item.name for item in persisted_documents] == ["Original name"]
|
|
assert len(links) == 1
|
|
assert links[0].role_id == role.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_link_writes_reject_new_inactive_role_assignments(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people = PeopleService(session_factory=default_session_factory)
|
|
active = await people.create_person_role(label="Witness")
|
|
inactive = await people.create_person_role(label="Former Witness", is_active=False)
|
|
person = await people.create_person(Person(full_name="Archive Witness"))
|
|
document = await documents.create_document(Document(name="Role rules"))
|
|
link = await people.add_document_person_link(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role_id=active.id,
|
|
)
|
|
|
|
with pytest.raises(PeopleError, match="Inactive Person Role"):
|
|
await people.set_document_person_role(
|
|
document_person_id=link.id,
|
|
role_id=inactive.id,
|
|
)
|
|
|
|
unchanged = await people.set_document_person_role(
|
|
document_person_id=link.id,
|
|
role_id=active.id,
|
|
)
|
|
assert unchanged.role_id == active.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_document_print_projection_uses_semantic_author_and_current_text(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people = PeopleService(session_factory=default_session_factory)
|
|
sources = SourceService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
document_type = await documents.create_document_type(label="Print Type")
|
|
document = await documents.create_document(
|
|
Document(name="Print Me", notes="Archive note", document_type_id=document_type.id)
|
|
)
|
|
person = await people.create_person(Person(full_name="Historic Author"))
|
|
|
|
async with people._session_scope() as session:
|
|
author = PersonRole(
|
|
semantic_key="author",
|
|
label="Creator",
|
|
normalized_label="creator",
|
|
)
|
|
session.add(author)
|
|
await session.flush()
|
|
session.add(DocumentPerson(document_id=document.id, person_id=person.id, role_id=author.id))
|
|
await session.commit()
|
|
|
|
await sources.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
upload_name="page-2.png",
|
|
filename="page-2.png",
|
|
file_path="managed/page-2.png",
|
|
file_hash="2" * 64,
|
|
file_size_bytes=2,
|
|
raw_transcription="raw second",
|
|
revised_text="revised second",
|
|
)
|
|
)
|
|
await sources.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="page-1.png",
|
|
filename="page-1.png",
|
|
file_path="managed/page-1.png",
|
|
file_hash="1" * 64,
|
|
file_size_bytes=1,
|
|
raw_transcription="raw first",
|
|
)
|
|
)
|
|
await jobs.create_job(
|
|
Job(
|
|
document_id=document.id,
|
|
status=JobStatus.TRANSCRIBED,
|
|
provider="openrouter",
|
|
model="model-a",
|
|
prompt_name="transcribe_document.md",
|
|
date_created=datetime(2026, 1, 1, tzinfo=UTC),
|
|
)
|
|
)
|
|
|
|
projection = await documents.read_document_print_projection(document.id)
|
|
|
|
assert projection.authors == ("Historic Author",)
|
|
assert projection.document_type == "Print Type"
|
|
assert [source.page_number for source in projection.sources] == [1, 2]
|
|
assert [source.current_text for source in projection.sources] == ["raw first", "revised second"]
|
|
assert [source.media_type for source in projection.sources] == ["image/png", "image/png"]
|
|
assert projection.jobs[0].status == "transcribed"
|