generated from john/python-template
126 lines
4.7 KiB
Python
126 lines
4.7 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.services.documents import DocumentService
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.transcription import TranscriptionService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_document_service_handles_person_and_document_person_crud(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
|
|
document = await documents.create_document(Document(id=uuid4(), name="person-doc"))
|
|
person = await documents.create_person(Person(full_name="Ada Lovelace"))
|
|
|
|
link = await documents.create_document_person(
|
|
DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
|
|
)
|
|
|
|
fetched = await documents.read_document_person(link.id)
|
|
assert fetched.id == link.id
|
|
assert fetched.role == DocumentPersonRole.AUTHOR
|
|
|
|
updated_link = await documents.update_document_person(
|
|
DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role=DocumentPersonRole.RECIPIENT)
|
|
)
|
|
assert updated_link.role == DocumentPersonRole.RECIPIENT
|
|
|
|
listed = await documents.list_document_people(document_id=document.id)
|
|
assert len(listed) == 1
|
|
|
|
people = await documents.list_people()
|
|
assert len(people) == 1
|
|
|
|
await documents.delete_document_person(updated_link)
|
|
assert len(await documents.list_document_people(document_id=document.id)) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcription_service_manages_source_crud(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
transcriptions = TranscriptionService(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",
|
|
)
|
|
)
|
|
|
|
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_transcription_service_job_source_crud_uses_caller_session(default_session_factory):
|
|
documents = DocumentService(session_factory=default_session_factory)
|
|
jobs = JobService(session_factory=default_session_factory)
|
|
transcriptions = TranscriptionService(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",
|
|
)
|
|
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
|