generated from john/python-template
zoltan57andCopilot App
66e2dce465
Baseline was 207 diagnostics. Two real bugs were hiding in the noise: - tools/run_destructive_tests.py imported ctypes.wintypes at module scope, which raises on non-Windows, and called fcntl unconditionally. The Windows and POSIX implementations now live under a module-level sys.platform split. - tests/ui/test_sources_page.py constructed Source(...) without document_id. Structural fixes, not suppressions: - New src/transcription/db/loading.py owns the SQLModel-field to QueryableAttribute reinterpretation via orm_attribute()/selectinload()/ defer(). This removed 42 "# pyright: ignore[reportArgumentType]" comments across documents/jobs/people/sources. Its docstring records that selectinload(A.b, B.c) is NOT equivalent to the chained form: varargs applies the selectin strategy only to the last path element, which under lazy="raise" raises InvalidRequestError at render time. - db/session.py transaction_scope no longer accepts or yields AsyncSessionTransaction. No caller ever passed one, sessionmaker.begin() yields an AsyncSession, and the dead branch was latently buggy because services call .exec(). Cleared 7 workflows.py diagnostics. - services/registry.py RegistryService is bound by a new RegistryEntry Protocol instead of bare SQLModel, so the shared implementation can read id/label/normalized_label/is_active. Cleared 9 diagnostics. - Column expressions in sources.py/jobs.py/test_store.py wrap in sqlmodel col(), the idiom already used in registry.py. - read_source_navigation wraps its literal tuple bounds in literal(). - normalization.py narrows with isinstance(image, TiffImageFile) rather than comparing image.format, since tag_v2 is TIFF-only. - linked_people.render uses @ui.refreshable_method, the NiceGUI API for bound methods. - The OpenRouter capturing client re-raises ResponseNotRead when the response stream is not async rather than mis-wrapping it. Tooling gate: - New .pre-commit-config.yaml runs ruff check and ty check as blocking hooks. No pre-commit config previously existed. Negative-tested: injecting a type error fails both hooks. - The last two "# pyright: ignore" comments (config.py) are removed; ty does not honor pyright directives. One "# ty: ignore" remains, in tests/test_prompts.py, where the test deliberately assigns to a frozen field to assert ValidationError. - asyncio_default_fixture_loop_scope is pinned to "function" so pytest-asyncio behavior does not shift on upgrade. Verification: ruff check clean, ty check reports 0 diagnostics, 292 passed and 4 skipped, pre-commit passes and demonstrably fails on a regression, and tools/run_destructive_tests.py runs on Windows. Co-authored-by: Copilot App <[email protected]>
328 lines
13 KiB
Python
328 lines
13 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
from pydantic import JsonValue
|
|
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentPerson
|
|
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"))
|
|
author_role = await people_service.create_person_role(label="Author")
|
|
recipient_role = await people_service.create_person_role(label="Recipient")
|
|
|
|
assert document.document_type_id is None
|
|
|
|
link = await people_service.create_document_person(
|
|
DocumentPerson(document_id=document.id, person_id=person.id, role_id=author_role.id)
|
|
)
|
|
|
|
fetched = await people_service.read_document_person(link.id)
|
|
assert fetched.id == link.id
|
|
assert fetched.role_id == author_role.id
|
|
|
|
updated_link = await people_service.update_document_person(
|
|
DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role_id=recipient_role.id)
|
|
)
|
|
assert updated_link.role_id == recipient_role.id
|
|
|
|
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"))
|
|
author_role = await people_service.create_person_role(label="Author")
|
|
await people_service.create_document_person(
|
|
DocumentPerson(
|
|
document_id=document.id,
|
|
person_id=person.id,
|
|
role_id=author_role.id,
|
|
)
|
|
)
|
|
|
|
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_id == author_role.id
|
|
|
|
|
|
@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: dict[str, JsonValue] = {
|
|
"finish_reason": "stop",
|
|
"usage": {"input_tokens": 11, "output_tokens": 22, "total_tokens": 33},
|
|
}
|
|
raw_payload: dict[str, JsonValue] = {"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
|