generated from john/python-template
330 lines
13 KiB
Python
330 lines
13 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentPerson
|
|
from transcription.db.models import DocumentPersonRole
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import Person
|
|
from transcription.db.models import Source
|
|
from transcription.errors import ErrorCategory
|
|
from transcription.services.documents import DocumentDeleteBlockedError
|
|
from transcription.services.documents import DocumentService
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.people import PeopleError
|
|
from transcription.services.people import PeopleService
|
|
from transcription.services.sources import SourceDeleteBlockedError
|
|
from transcription.services.sources import SourceService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_people_service_handles_person_and_document_person_crud(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people_service = PeopleService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="person-doc"))
|
|
person = await people_service.create_person(Person(full_name="Ada Lovelace"))
|
|
|
|
assert document.document_type_id is None
|
|
|
|
link = await people_service.create_document_person(
|
|
DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
|
|
)
|
|
|
|
fetched = await people_service.read_document_person(link.id)
|
|
assert fetched.id == link.id
|
|
assert fetched.role == DocumentPersonRole.AUTHOR
|
|
assert fetched.role_id is not None
|
|
|
|
updated_link = await people_service.update_document_person(
|
|
DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role=DocumentPersonRole.RECIPIENT)
|
|
)
|
|
assert updated_link.role == DocumentPersonRole.RECIPIENT
|
|
assert updated_link.role_id is not None
|
|
|
|
listed = await people_service.list_document_people(document_id=document.id)
|
|
assert len(listed) == 1
|
|
|
|
people = await people_service.list_people()
|
|
assert len(people) == 1
|
|
|
|
await people_service.delete_document_person(updated_link)
|
|
assert len(await people_service.list_document_people(document_id=document.id)) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_people_service_normalizes_and_rejects_duplicate_family_search_ids(default_session_factory):
|
|
people_service = PeopleService(session_factory=default_session_factory)
|
|
|
|
created = await people_service.create_person(
|
|
Person(full_name="Hig Higgins", family_search_id=" g8t4-mdq ")
|
|
)
|
|
assert created.family_search_id == "G8T4-MDQ"
|
|
|
|
with pytest.raises(PeopleError) as duplicate:
|
|
await people_service.create_person(
|
|
Person(full_name="Duplicate Hig", family_search_id="G8T4-MDQ")
|
|
)
|
|
assert duplicate.value.category == ErrorCategory.CONFLICT
|
|
|
|
with pytest.raises(PeopleError) as malformed:
|
|
await people_service.create_person(
|
|
Person(full_name="Malformed", family_search_id="not-an-id")
|
|
)
|
|
assert malformed.value.category == ErrorCategory.VALIDATION
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcription_service_manages_source_crud(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
transcriptions = SourceService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="source-doc"))
|
|
source = await transcriptions.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="page-1.jpg",
|
|
filename="page-1.jpg",
|
|
file_path="uploads/page-1.jpg",
|
|
file_hash="7" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
|
|
fetched = await transcriptions.read_source(source.id)
|
|
assert fetched.id == source.id
|
|
|
|
source.page_number = 2
|
|
updated = await transcriptions.update_source(source)
|
|
assert updated.page_number == 2
|
|
|
|
listed = await transcriptions.list_sources(document_id=document.id)
|
|
assert len(listed) == 1
|
|
|
|
filtered = await transcriptions.query_sources(document_id=document.id, page_number=2)
|
|
assert len(filtered) == 1
|
|
|
|
await transcriptions.delete_source(updated)
|
|
assert len(await transcriptions.list_sources(document_id=document.id)) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_navigation_is_bounded_to_ordered_document(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
sources = SourceService(session_factory=default_session_factory)
|
|
document = await documents.create_document(Document(id=uuid4(), name="ordered"))
|
|
other = await documents.create_document(Document(id=uuid4(), name="other"))
|
|
|
|
first, second, third, _foreign = [
|
|
await sources.create_source(
|
|
Source(
|
|
document_id=document_id,
|
|
page_number=page_number,
|
|
upload_name=f"page-{page_number}.jpg",
|
|
filename=f"page-{page_number}.jpg",
|
|
file_path=f"uploads/page-{page_number}.jpg",
|
|
file_hash=str(page_number) * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
for document_id, page_number in [
|
|
(document.id, 1),
|
|
(document.id, 2),
|
|
(document.id, 3),
|
|
(other.id, 2),
|
|
]
|
|
]
|
|
|
|
first_navigation = await sources.read_source_navigation(first.id)
|
|
middle_navigation = await sources.read_source_navigation(second.id)
|
|
last_navigation = await sources.read_source_navigation(third.id)
|
|
|
|
assert (first_navigation.previous_id, first_navigation.next_id) == (None, second.id)
|
|
assert (middle_navigation.previous_id, middle_navigation.next_id) == (first.id, third.id)
|
|
assert (last_navigation.previous_id, last_navigation.next_id) == (second.id, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcription_service_job_source_crud_uses_caller_session(default_session_factory):
|
|
transcriptions = SourceService(session_factory=default_session_factory)
|
|
|
|
async with transcriptions._session_scope() as session:
|
|
document = Document(id=uuid4(), name="job-source-doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
|
|
job = Job(document_id=document.id)
|
|
session.add(job)
|
|
await session.flush()
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="job-source.jpg",
|
|
filename="job-source.jpg",
|
|
file_path="uploads/job-source.jpg",
|
|
file_hash="8" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
|
|
job_source = await transcriptions.create_job_source(
|
|
JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING),
|
|
session=session,
|
|
)
|
|
assert job_source.status == JobSourceStatus.PENDING
|
|
|
|
job_source.status = JobSourceStatus.TRANSCRIBED
|
|
updated = await transcriptions.update_job_source(job_source, session=session)
|
|
assert updated.status == JobSourceStatus.TRANSCRIBED
|
|
|
|
fetched = await transcriptions.read_job_source(job_source.id, session=session)
|
|
assert fetched.id == job_source.id
|
|
|
|
listed = await transcriptions.list_job_sources(job_id=job.id, session=session)
|
|
assert len(listed) == 1
|
|
|
|
await transcriptions.delete_job_source(updated, session=session)
|
|
await session.commit()
|
|
|
|
assert len(await transcriptions.list_job_sources(job_id=job.id)) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_document_detail_loads_linked_person_relationship(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
people_service = PeopleService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="detail-person-doc"))
|
|
person = await people_service.create_person(Person(full_name="Grace Hopper"))
|
|
await people_service.create_document_person(
|
|
DocumentPerson(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role=DocumentPersonRole.AUTHOR,
|
|
)
|
|
)
|
|
|
|
detail = await documents.read_document_detail(document.id)
|
|
|
|
assert len(detail.document_people) == 1
|
|
link = detail.document_people[0]
|
|
assert link.person is not None
|
|
assert link.person.full_name == "Grace Hopper"
|
|
assert link.role == DocumentPersonRole.AUTHOR
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_document_delete_is_blocked_with_source_and_job_dependencies(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = SourceService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="blocked-by-deps"))
|
|
job = await jobs.create_job(Job(document_id=document.id))
|
|
source = await transcriptions.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="blocked.jpg",
|
|
filename="blocked.jpg",
|
|
file_path="uploads/blocked.jpg",
|
|
file_hash="9" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
await transcriptions.create_job_source(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
|
|
with pytest.raises(DocumentDeleteBlockedError) as exc_info:
|
|
await documents.delete_document(document)
|
|
|
|
message = exc_info.value.message
|
|
assert "Sources" in message
|
|
assert "Jobs" in message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_delete_blocks_when_linked_to_multiple_jobs(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = SourceService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="multi-job-source-doc"))
|
|
job_one = await jobs.create_job(Job(document_id=document.id))
|
|
job_two = await jobs.create_job(Job(document_id=document.id))
|
|
|
|
source = await transcriptions.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="shared-page.jpg",
|
|
filename="shared-page.jpg",
|
|
file_path="uploads/shared-page.jpg",
|
|
file_hash="a" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
await transcriptions.create_job_source(
|
|
JobSource(job_id=job_one.id, source_id=source.id, status=JobSourceStatus.PENDING)
|
|
)
|
|
await transcriptions.create_job_source(
|
|
JobSource(job_id=job_two.id, source_id=source.id, status=JobSourceStatus.PENDING)
|
|
)
|
|
|
|
with pytest.raises(SourceDeleteBlockedError):
|
|
await transcriptions.delete_source_from_job_context(job_id=job_one.id, source_id=source.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_job_source_transcription_persists_provider_json_payloads(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = SourceService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="provider-payloads-doc"))
|
|
job = await jobs.create_job(Job(document_id=document.id))
|
|
source = await transcriptions.create_source(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="provider.jpg",
|
|
filename="provider.jpg",
|
|
file_path="uploads/provider.jpg",
|
|
file_hash="b" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
await transcriptions.create_job_source(
|
|
JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
|
|
)
|
|
|
|
metadata = {"finish_reason": "stop", "usage": {"input_tokens": 11, "output_tokens": 22, "total_tokens": 33}}
|
|
raw_payload = {"id": "resp_xyz", "choices": [{"message": {"content": "provider transcript"}}]}
|
|
|
|
await transcriptions.update_job_source_transcription(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
text="provider transcript",
|
|
ai_metadata=metadata,
|
|
raw_api_response=raw_payload,
|
|
provider="openrouter",
|
|
model="test-model",
|
|
)
|
|
|
|
stored_rows = await transcriptions.list_job_sources(job_id=job.id)
|
|
assert len(stored_rows) == 1
|
|
assert stored_rows[0].raw_transcription == "provider transcript"
|
|
assert stored_rows[0].ai_metadata == metadata
|
|
assert stored_rows[0].raw_api_response == raw_payload
|