bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

V4.4 Complete

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-15 14:30:33 -05:00
parent 63373bf24d
commit 7db4df1729
32 changed files with 1529 additions and 716 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/test_v4_documents.py
+27 -15
View File
@@ -24,6 +24,7 @@ from transcription.db.engine import get_engine
from transcription.db.models import Document
from transcription.db.models import DocumentType
from transcription.db.models import Person
from transcription.db.models import PersonRole
from transcription.db.session import dispose_session_factory
from transcription.db.session import session_scope
from transcription.services.documents import DocumentService
@@ -53,6 +54,12 @@ async def _document_type_id(*, db_url: str, label: str) -> UUID:
return document_type.id
async def _person_role_id(*, db_url: str, semantic_key: str) -> UUID:
async with session_scope(database_url=db_url) as session:
role = (await session.exec(select(PersonRole).where(PersonRole.semantic_key == semantic_key))).one()
return role.id
@contextmanager
def _v4_api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestClient, str]]:
settings = Settings(
@@ -97,7 +104,7 @@ def test_list_document_types_returns_seeded_registry(tmp_path):
assert response.status_code == 200
payload = response.json()
labels = {item["label"] for item in payload}
assert {"Letter", "Record", "Memo"}.issubset(labels)
assert {"Book", "Letter", "Postcard", "Photo", "Journal", "Form"}.issubset(labels)
def test_list_person_roles_returns_seeded_registry(tmp_path):
@@ -106,14 +113,15 @@ def test_list_person_roles_returns_seeded_registry(tmp_path):
assert response.status_code == 200
payload = response.json()
codes = {item["code"] for item in payload}
assert {"author", "recipient", "mentioned"}.issubset(codes)
labels = {item["label"] for item in payload}
assert {"Author", "Recipient", "Mentioned"}.issubset(labels)
assert all("code" not in item and "semantic_key" not in item for item in payload)
def test_set_document_type_by_id_updates_canonical_field(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-doc-type.db") as (client, db_url):
document_id, _ = _seed_document_and_person(db_url=db_url)
type_id = asyncio.run(_document_type_id(db_url=db_url, label="Record"))
type_id = asyncio.run(_document_type_id(db_url=db_url, label="Form"))
response = client.put(
f"/api/v4/documents/{document_id}/type",
json={"document_type_id": str(type_id)},
@@ -147,32 +155,35 @@ def test_document_type_payload_requires_uuid_only(tmp_path):
def test_document_people_role_aware_write_read_and_delete(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-links.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
author_id = asyncio.run(_person_role_id(db_url=db_url, semantic_key="author"))
recipient_id = asyncio.run(_person_role_id(db_url=db_url, semantic_key="recipient"))
create_response = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
json={"person_id": str(person_id), "role_id": str(author_id)},
)
assert create_response.status_code == 200
created = create_response.json()
assert created["document_id"] == str(document_id)
assert created["person_id"] == str(person_id)
assert created["role_code"] == "author"
assert created["role_id"] is not None
assert created["role_id"] == str(author_id)
assert created["role_label"] == "Author"
link_id = created["id"]
update_response = client.patch(
f"/api/v4/document-people/{link_id}",
json={"role_code": "recipient"},
json={"role_id": str(recipient_id)},
)
assert update_response.status_code == 200
updated = update_response.json()
assert updated["role_code"] == "recipient"
assert updated["role_id"] == str(recipient_id)
assert updated["role_label"] == "Recipient"
list_response = client.get(f"/api/v4/documents/{document_id}/people")
assert list_response.status_code == 200
links = list_response.json()["links"]
assert len(links) == 1
assert links[0]["role_code"] == "recipient"
assert links[0]["role_id"] == str(recipient_id)
delete_response = client.delete(f"/api/v4/document-people/{link_id}")
assert delete_response.status_code == 204
@@ -182,7 +193,7 @@ def test_document_people_role_aware_write_read_and_delete(tmp_path):
assert list_after_delete.json()["links"] == []
def test_document_person_link_defaults_to_author_when_role_is_omitted(tmp_path):
def test_document_person_link_requires_role_id(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-default-role.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
@@ -191,23 +202,24 @@ def test_document_person_link_defaults_to_author_when_role_is_omitted(tmp_path):
json={"person_id": str(person_id)},
)
assert response.status_code == 200
assert response.json()["role_code"] == "author"
assert response.status_code == 422
def test_duplicate_document_person_link_returns_conflict_envelope(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-dup.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
author_id = asyncio.run(_person_role_id(db_url=db_url, semantic_key="author"))
recipient_id = asyncio.run(_person_role_id(db_url=db_url, semantic_key="recipient"))
first = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
json={"person_id": str(person_id), "role_id": str(author_id)},
)
assert first.status_code == 200
second = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
json={"person_id": str(person_id), "role_id": str(recipient_id)},
)
assert second.status_code == 409
tests/services/test_document_service.py
+13 -20
View File
@@ -5,15 +5,12 @@ from datetime import datetime
from uuid import uuid4
import pytest
from sqlmodel import select
from transcription.config import Settings
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 Person
from transcription.db.models import PersonRole
from transcription.db.models import Source
from transcription.services.documents import DocumentDeleteBlockedError
from transcription.services.documents import DocumentError
@@ -132,11 +129,12 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp
)
)
person = await people_service.create_person(Person(full_name="Linked Person"))
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=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
@@ -144,7 +142,7 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp
assert len(links_before_delete) == 1
assert links_before_delete[0].role_id is not None
assert links_before_delete[0].role_ref is not None
assert links_before_delete[0].role_ref.code == "author"
assert links_before_delete[0].role_ref.label == "Author"
document_dir = service.settings.upload_dir / "documents" / str(document.id)
document_dir.mkdir(parents=True, exist_ok=True)
@@ -196,11 +194,12 @@ async def test_read_person_detail_loads_document_links(default_session_factory):
)
)
person = await people_service.create_person(Person(full_name="Linked Person"))
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=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
@@ -243,11 +242,12 @@ async def test_delete_person_removes_links_when_linked_documents_exist(default_s
)
)
person = await service.create_person(Person(full_name="Blocked Person"))
author_role = await service.create_person_role(label="Author")
await service.create_document_person(
DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
@@ -285,19 +285,19 @@ async def test_create_document_uses_existing_document_type_registry(default_sess
@pytest.mark.asyncio
async def test_update_document_person_sets_role_id_from_legacy_role(default_session_factory):
async def test_update_document_person_changes_role_id(default_session_factory):
documents_service = DocumentService(session_factory=default_session_factory)
service = PeopleService(session_factory=default_session_factory)
document = await documents_service.create_document(
Document(id=uuid4(), name="role-sync-doc", document_type="letter")
)
document = await documents_service.create_document(Document(id=uuid4(), name="role-sync-doc"))
person = await service.create_person(Person(full_name="Role Sync Person"))
author_role = await service.create_person_role(label="Author")
recipient_role = await service.create_person_role(label="Recipient")
link = await service.create_document_person(
DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
@@ -306,15 +306,8 @@ async def test_update_document_person_sets_role_id_from_legacy_role(default_sess
id=link.id,
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.RECIPIENT,
role_id=None,
role_id=recipient_role.id,
)
)
assert updated.role == DocumentPersonRole.RECIPIENT
assert updated.role_id is not None
async with service._session_scope() as session:
recipient_role = (await session.exec(select(PersonRole).where(PersonRole.code == "recipient"))).first()
assert recipient_role is not None
assert updated.role_id == recipient_role.id
tests/services/test_settings_services.py
+24 -28
View File
@@ -6,8 +6,8 @@ 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 Person
from transcription.db.models import PersonRole
from transcription.errors import ErrorCategory
from transcription.services.documents import DocumentService
from transcription.services.documents import DocumentTypeError
@@ -73,24 +73,27 @@ async def test_document_type_delete_allows_unreferenced_and_blocks_referenced(de
@pytest.mark.asyncio
async def test_person_role_maintenance_orders_by_label_then_code(default_session_factory):
async def test_person_role_maintenance_orders_by_normalized_label(default_session_factory):
service = PeopleService(session_factory=default_session_factory)
second = await service.create_person_role(code="witness", label="Witness")
first = await service.create_person_role(code="author", label="Author")
second = await service.create_person_role(label="Witness")
first = await service.create_person_role(label="Archivist")
updated = await service.update_person_role(second.id, label="Attestor", is_active=False)
assert updated.code == "witness"
assert [item.id for item in await service.list_person_roles(active_only=False)] == [second.id, first.id]
assert updated.semantic_key is None
assert [item.id for item in await service.list_person_roles(active_only=False)] == [first.id, second.id]
assert [item.id for item in await service.list_person_roles()] == [first.id]
summaries = {item.id: item for item in await service.list_person_role_summaries()}
assert summaries[second.id].link_count == 0
assert summaries[second.id].is_built_in is False
@pytest.mark.asyncio
async def test_person_role_delete_allows_unreferenced_and_blocks_referenced(default_session_factory):
documents = DocumentService(session_factory=default_session_factory)
people = PeopleService(session_factory=default_session_factory)
unused = await people.create_person_role(code="witness", label="Witness")
referenced = await people.create_person_role(code="author", label="Author")
unused = await people.create_person_role(label="Witness")
referenced = await people.create_person_role(label="Creator")
document = await documents.create_document(Document(name="Role document"))
person = await people.create_person(Person(full_name="Role Person"))
await people.create_document_person(
@@ -98,7 +101,6 @@ async def test_person_role_delete_allows_unreferenced_and_blocks_referenced(defa
document_id=document.id,
person_id=person.id,
role_id=referenced.id,
role=DocumentPersonRole.AUTHOR,
)
)
@@ -113,12 +115,12 @@ async def test_person_role_delete_allows_unreferenced_and_blocks_referenced(defa
@pytest.mark.asyncio
async def test_person_role_duplicate_code_is_conflict(default_session_factory):
async def test_person_role_duplicate_normalized_label_is_conflict(default_session_factory):
service = PeopleService(session_factory=default_session_factory)
await service.create_person_role(code="author", label="Author")
await service.create_person_role(label="Witness")
with pytest.raises(PersonRoleError) as caught:
await service.create_person_role(code=" AUTHOR ", label="Duplicate")
await service.create_person_role(label=" witness ")
assert caught.value.category == ErrorCategory.CONFLICT
@@ -127,7 +129,7 @@ async def test_person_role_duplicate_code_is_conflict(default_session_factory):
async def test_custom_person_role_can_be_used_for_document_link(default_session_factory):
documents = DocumentService(session_factory=default_session_factory)
people = PeopleService(session_factory=default_session_factory)
role = await people.create_person_role(code="witness", label="Witness")
role = await people.create_person_role(label="Witness")
document = await documents.create_document(Document(name="Witnessed document"))
person = await people.create_person(Person(full_name="Archive Witness"))
@@ -138,30 +140,24 @@ async def test_custom_person_role_can_be_used_for_document_link(default_session_
)
loaded = await people.list_document_people(document_id=document.id)
assert link.role == "witness"
assert link.role_id == role.id
assert loaded[0].role_ref is not None
assert loaded[0].role_ref.code == "witness"
assert loaded[0].role_ref.label == "Witness"
@pytest.mark.asyncio
async def test_custom_person_role_delete_blocks_legacy_only_reference(default_session_factory):
documents = DocumentService(session_factory=default_session_factory)
async def test_built_in_person_role_cannot_be_deleted(default_session_factory):
people = PeopleService(session_factory=default_session_factory)
role = await people.create_person_role(code="witness", label="Witness")
document = await documents.create_document(Document(name="Legacy role document"))
person = await people.create_person(Person(full_name="Legacy Witness"))
async with people._session_scope() as session:
session.add(
DocumentPerson(
document_id=document.id,
person_id=person.id,
role="witness",
role_id=None,
)
role = PersonRole(
semantic_key="author",
label="Author",
normalized_label="author",
)
session.add(role)
await session.commit()
await session.refresh(role)
assert await people.is_person_role_referenced(role.id) is True
with pytest.raises(PersonRoleError) as caught:
await people.delete_person_role(role.id)
tests/services/test_v2_crud.py
+12 -18
View File
@@ -4,7 +4,6 @@ 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
@@ -27,23 +26,23 @@ async def test_people_service_handles_person_and_document_person_crud(default_se
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=DocumentPersonRole.AUTHOR)
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 == DocumentPersonRole.AUTHOR
assert fetched.role_id is not None
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=DocumentPersonRole.RECIPIENT)
DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role_id=recipient_role.id)
)
assert updated_link.role == DocumentPersonRole.RECIPIENT
assert updated_link.role_id is not None
assert updated_link.role_id == recipient_role.id
listed = await people_service.list_document_people(document_id=document.id)
assert len(listed) == 1
@@ -59,21 +58,15 @@ async def test_people_service_handles_person_and_document_person_crud(default_se
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 ")
)
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")
)
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")
)
await people_service.create_person(Person(full_name="Malformed", family_search_id="not-an-id"))
assert malformed.value.category == ErrorCategory.VALIDATION
@@ -202,11 +195,12 @@ async def test_document_detail_loads_linked_person_relationship(default_session_
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=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
@@ -216,7 +210,7 @@ async def test_document_detail_loads_linked_person_relationship(default_session_
link = detail.document_people[0]
assert link.person is not None
assert link.person.full_name == "Grace Hopper"
assert link.role == DocumentPersonRole.AUTHOR
assert link.role_id == author_role.id
@pytest.mark.asyncio
tests/services/test_v44_workflows.py
+166
View File
@@ -0,0 +1,166 @@
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)],
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)],
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)],
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 = await documents.create_document(Document(name="Print Me", notes="Archive note"))
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.COMPLETED,
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 [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 == "completed"
tests/test_db.py
+4 -93
View File
@@ -1,7 +1,5 @@
"""Tests for the database runtime and V2 schema bootstrap behavior."""
from uuid import uuid4
import pytest
from sqlalchemy import inspect
from sqlalchemy import text
@@ -98,11 +96,11 @@ async def test_create_all_seeds_default_registry_rows(tmp_path):
try:
await create_all(engine=runtime.engine)
async with AsyncSession(runtime.engine, expire_on_commit=False) as session:
role_codes = set((await session.exec(select(PersonRole.code))).all())
type_labels = set((await session.exec(select(DocumentType.label))).all())
role_keys = set((await session.exec(select(PersonRole.semantic_key))).all())
type_keys = set((await session.exec(select(DocumentType.semantic_key))).all())
assert {"author", "recipient", "mentioned"}.issubset(role_codes)
assert {"Letter", "Record", "Memo"}.issubset(type_labels)
assert {"author", "recipient", "mentioned"}.issubset(role_keys)
assert {"book", "letter", "postcard", "photo", "journal", "form"}.issubset(type_keys)
finally:
await dispose_database_runtime()
@@ -137,93 +135,6 @@ async def test_create_all_upgrades_existing_person_table_for_family_search(tmp_p
await dispose_database_runtime()
@pytest.mark.asyncio
async def test_upgrade_migrates_document_types_to_uuid_only_identity(tmp_path):
settings = Settings(
openrouter_api_key="test-key",
database=SqliteSettings(path=str(tmp_path / "type-upgrade.db")),
environment="test",
)
runtime = initialize_database_runtime(settings=settings)
type_id = uuid4().hex
document_id = uuid4().hex
try:
async with runtime.engine.begin() as connection:
await connection.execute(
text(
"CREATE TABLE document_type ("
"id CHAR(32) PRIMARY KEY NOT NULL, "
"code VARCHAR NOT NULL, "
"label VARCHAR NOT NULL, "
"is_active BOOLEAN NOT NULL, "
"sort_order INTEGER NOT NULL, "
"created_at DATETIME NOT NULL, "
"updated_at DATETIME NOT NULL"
")"
)
)
await connection.execute(text("CREATE UNIQUE INDEX ix_document_type_code ON document_type (code)"))
await connection.execute(
text(
"CREATE TABLE document ("
"id CHAR(32) PRIMARY KEY NOT NULL, "
"name VARCHAR NOT NULL, "
"document_type_id CHAR(32), "
"document_type VARCHAR, "
"document_date DATE, "
"document_date_raw VARCHAR, "
"location_created VARCHAR, "
"notes VARCHAR, "
"archive_identifier VARCHAR, "
"created_at DATETIME NOT NULL, "
"updated_at DATETIME NOT NULL"
")"
)
)
await connection.execute(
text(
"INSERT INTO document_type "
"(id, code, label, is_active, sort_order, created_at, updated_at) "
"VALUES (:id, 'letter', 'Letter', 1, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
),
{"id": type_id},
)
await connection.execute(
text(
"INSERT INTO document "
"(id, name, document_type_id, document_type, created_at, updated_at) "
"VALUES (:id, 'Legacy Letter', NULL, 'letter', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
),
{"id": document_id},
)
await upgrade_schema(engine=runtime.engine)
async with runtime.engine.connect() as connection:
type_columns, document_columns, migrated_type_id, normalized_label = await connection.run_sync(
lambda sync_connection: (
{column["name"] for column in inspect(sync_connection).get_columns("document_type")},
{column["name"] for column in inspect(sync_connection).get_columns("document")},
sync_connection.execute(
text("SELECT document_type_id FROM document WHERE id = :id"),
{"id": document_id},
).scalar_one(),
sync_connection.execute(
text("SELECT normalized_label FROM document_type WHERE id = :id"),
{"id": type_id},
).scalar_one(),
)
)
assert {"code", "sort_order"}.isdisjoint(type_columns)
assert "document_type" not in document_columns
assert migrated_type_id == type_id
assert normalized_label == "letter"
finally:
await dispose_database_runtime()
@pytest.mark.asyncio
async def test_v42_upgrade_adds_evidence_tables_without_rewriting_legacy_snapshot(tmp_path):
settings = Settings(
tests/test_models.py
+5 -9
View File
@@ -7,7 +7,6 @@ from sqlalchemy.exc import IntegrityError
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.db.models import DocumentType
from transcription.db.models import Job
from transcription.db.models import JobSource
@@ -35,8 +34,8 @@ def _persist_document_type(session, *, label: str = "Letter") -> DocumentType:
return document_type
def _persist_person_role(session, *, code: str = "author", label: str = "Author") -> PersonRole:
role = PersonRole(code=code, label=label)
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)
@@ -195,7 +194,6 @@ class TestPersonAndDocumentPersonModel:
first = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(first)
@@ -204,7 +202,6 @@ class TestPersonAndDocumentPersonModel:
duplicate = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(duplicate)
@@ -244,7 +241,6 @@ class TestRelationships:
link = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(link)
@@ -257,9 +253,9 @@ class TestRelationships:
class TestRegistryModels:
def test_person_role_code_is_unique(self, session):
_persist_person_role(session, code="mentioned", label="Mentioned")
duplicate = PersonRole(code="mentioned", label="Mentioned Again")
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()
tests/ui/test_documents_page.py
+7 -6
View File
@@ -9,10 +9,10 @@ from sqlmodel import select
from transcription.db import session_scope
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.db.models import DocumentType
from transcription.db.models import Job
from transcription.db.models import Person
from transcription.db.models import PersonRole
from transcription.db.models import Source
# --- Helper Fixtures ---
@@ -23,6 +23,7 @@ async def seed_person_and_document():
"""Seed a Person and Document linked by DocumentPerson role."""
async with session_scope() as session:
letter_type = (await session.exec(select(DocumentType).where(DocumentType.label == "Letter"))).one()
author_role = (await session.exec(select(PersonRole).where(PersonRole.semantic_key == "author"))).one()
person = Person(full_name="Zenna Cochran")
session.add(person)
await session.flush()
@@ -38,7 +39,7 @@ async def seed_person_and_document():
link = DocumentPerson(
document_id=doc.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
session.add(link)
await session.commit()
@@ -92,7 +93,7 @@ class TestDocumentsPageRendering:
assert response.status_code == 200
assert "Create Document" in response.text
assert "Document name" in response.text
assert "Linked People by Role" in response.text
assert "Linked People" in response.text
assert "Document type" in response.text
@pytest.mark.asyncio
@@ -132,7 +133,7 @@ class TestDocumentsPageRendering:
_, client = app_client
async with session_scope() as session:
doc = Document(name="Doc With Job", document_type="letter")
doc = Document(name="Doc With Job")
session.add(doc)
await session.flush()
@@ -165,7 +166,7 @@ class TestDocumentsPageRendering:
_, client = app_client
async with session_scope() as session:
doc = Document(name="Doc With Source", document_type="letter")
doc = Document(name="Doc With Source")
session.add(doc)
await session.flush()
@@ -194,7 +195,7 @@ class TestDocumentsPageRendering:
_, client = app_client
async with session_scope() as session:
doc = Document(name="Orphan Document", document_type="note")
doc = Document(name="Orphan Document")
session.add(doc)
await session.commit()
doc_id = str(doc.id)
tests/ui/test_jobs_page.py
+3 -3
View File
@@ -15,7 +15,7 @@ from transcription.db.models import JobStatus
async def seed_document_with_unlinked_job():
"""Seed a document and a queued job for testing route actions."""
async with session_scope() as session:
document = Document(name="Test Archival Letter", document_type="letter")
document = Document(name="Test Archival Letter")
session.add(document)
await session.flush()
@@ -72,7 +72,7 @@ class TestJobsPageRendering:
_, client = app_client
async with session_scope() as session:
doc = Document(name="Preselected Journal Entry", document_type="journal")
doc = Document(name="Preselected Journal Entry")
session.add(doc)
await session.commit()
doc_id = str(doc.id)
@@ -133,7 +133,7 @@ class TestJobsPageRendering:
_, client = app_client
async with session_scope() as session:
doc = Document(name="Processing Doc", document_type="letter")
doc = Document(name="Processing Doc")
session.add(doc)
await session.flush()
job = Job(document_id=doc.id, status=JobStatus.PROCESSING)
tests/ui/test_people_page.py
+6 -4
View File
@@ -4,12 +4,13 @@ from datetime import date
from uuid import uuid4
import pytest
from sqlmodel import select
from transcription.db import session_scope
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.db.models import Person
from transcription.db.models import PersonRole
@pytest.mark.integration
@@ -122,8 +123,9 @@ class TestPeoplePageRendering:
_, client = app_client
async with session_scope() as session:
author_role = (await session.exec(select(PersonRole).where(PersonRole.semantic_key == "author"))).one()
person = Person(full_name="Linked Person")
document = Document(name="Linked Document", document_type="letter")
document = Document(name="Linked Document")
session.add_all([person, document])
await session.flush()
@@ -131,7 +133,7 @@ class TestPeoplePageRendering:
DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=author_role.id,
)
)
await session.commit()
@@ -141,7 +143,7 @@ class TestPeoplePageRendering:
assert response.status_code == 200
assert "Linked Document" in response.text
assert "Role: author" in response.text
assert "Role: Author" in response.text
def test_person_detail_page_handles_invalid_id(self, app_client):
_, client = app_client
tests/ui/test_print_preview_page.py
+103
View File
@@ -0,0 +1,103 @@
from pathlib import Path
import pytest
from transcription.db import session_scope
from transcription.db.models import Document
from transcription.db.models import Source
from transcription.ui.pages.print_preview_page import reflow_transcription
def test_reflow_transcription_preserves_paragraph_boundaries():
assert reflow_transcription("first line\nsecond line\n\nnext paragraph") == [
"first line second line",
"next paragraph",
]
@pytest.mark.integration
@pytest.mark.asyncio
async def test_document_print_preview_and_safe_media_route(app_client):
app, client = app_client
media_path = app.state.settings.upload_dir / "documents" / "print-page.png"
pdf_path = app.state.settings.upload_dir / "documents" / "print-page.pdf"
media_path.parent.mkdir(parents=True, exist_ok=True)
media_path.write_bytes(b"\x89PNG\r\n\x1a\n")
pdf_path.write_bytes(b"%PDF-1.4\n%%EOF")
async with session_scope() as session:
document = Document(name="<Print & Preserve>", notes="<script>unsafe()</script>")
session.add(document)
await session.flush()
source = Source(
document_id=document.id,
page_number=1,
upload_name="print-page.png",
filename="print-page.png",
file_path=str(media_path),
file_hash="a" * 64,
file_size_bytes=media_path.stat().st_size,
raw_transcription="line one\nline two",
)
session.add(source)
session.add(
Source(
document_id=document.id,
page_number=2,
upload_name="print-page.pdf",
filename="print-page.pdf",
file_path=str(pdf_path),
file_hash="c" * 64,
file_size_bytes=pdf_path.stat().st_size,
raw_transcription="PDF source",
)
)
await session.commit()
document_id = document.id
source_id = source.id
response = client.get(f"/ui/documents/{document_id}/print")
assert response.status_code == 200
assert "Print &amp; Preserve" in response.text
assert "&lt;script&gt;unsafe()&lt;/script&gt;" in response.text
assert "Facsimile" in response.text
assert "Text only" in response.text
assert "print-page-break" in response.text
assert "print-source-pdf" in response.text
assert str(media_path) not in response.text
assert str(pdf_path) not in response.text
media_response = client.get(f"/api/v4/documents/{document_id}/sources/{source_id}/media")
assert media_response.status_code == 200
assert media_response.headers["content-type"] == "image/png"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_document_source_media_rejects_cross_document_access(app_client):
app, client = app_client
media_path = Path(app.state.settings.upload_dir) / "documents" / "other.png"
media_path.parent.mkdir(parents=True, exist_ok=True)
media_path.write_bytes(b"\x89PNG\r\n\x1a\n")
async with session_scope() as session:
owner = Document(name="Owner")
other = Document(name="Other")
session.add_all([owner, other])
await session.flush()
source = Source(
document_id=owner.id,
page_number=1,
upload_name="other.png",
filename="other.png",
file_path=str(media_path),
file_hash="b" * 64,
file_size_bytes=media_path.stat().st_size,
)
session.add(source)
await session.commit()
other_id = other.id
source_id = source.id
response = client.get(f"/api/v4/documents/{other_id}/sources/{source_id}/media")
assert response.status_code == 404
tests/ui/test_sources_page.py
+4 -4
View File
@@ -80,7 +80,7 @@ class TestSourcesPageRendering:
_, client = app_client
async with session_scope() as session:
document = Document(name="Source Document", document_type="letter")
document = Document(name="Source Document")
session.add(document)
await session.flush()
session.add(
@@ -108,8 +108,8 @@ class TestSourcesPageRendering:
_, client = app_client
async with session_scope() as session:
target = Document(name="Target", document_type="letter")
other = Document(name="Other", document_type="record")
target = Document(name="Target")
other = Document(name="Other")
session.add_all([target, other])
await session.flush()
@@ -274,7 +274,7 @@ class TestSourcesPageRendering:
_, client = app_client
async with session_scope() as session:
document = Document(name="Unlinked Source Doc", document_type="memo")
document = Document(name="Unlinked Source Doc")
session.add(document)
await session.flush()
source = Source(