generated from john/python-template
240 lines
9.3 KiB
Python
240 lines
9.3 KiB
Python
"""Integration tests for document and relationship API routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Generator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import select
|
|
|
|
from transcription.api.errors import register_error_handlers
|
|
from transcription.api.documents_api import get_document_service
|
|
from transcription.api.documents_api import get_people_service
|
|
from transcription.api.documents_api import router
|
|
from transcription.config import Settings
|
|
from transcription.config import SqliteSettings
|
|
from transcription.db import create_all
|
|
from transcription.db.engine import get_database_url
|
|
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
|
|
from transcription.services.people import PeopleService
|
|
|
|
|
|
def _seed_document_and_person(
|
|
*, db_url: str, document_name: str = "API Doc", person_name: str = "API Person"
|
|
) -> tuple[UUID, UUID]:
|
|
async def _seed() -> tuple[UUID, UUID]:
|
|
async with session_scope(database_url=db_url) as session:
|
|
document = Document(name=document_name)
|
|
tokens = [token for token in person_name.split() if token]
|
|
given_names = " ".join(tokens[:-1]) if len(tokens) >= 2 else person_name
|
|
last_name = tokens[-1] if len(tokens) >= 2 else person_name
|
|
person = Person(given_names=given_names, last_name=last_name)
|
|
session.add(document)
|
|
session.add(person)
|
|
await session.commit()
|
|
await session.refresh(document)
|
|
await session.refresh(person)
|
|
return document.id, person.id
|
|
|
|
return asyncio.run(_seed())
|
|
|
|
|
|
async def _document_type_id(*, db_url: str, label: str) -> UUID:
|
|
async with session_scope(database_url=db_url) as session:
|
|
document_type = (await session.exec(select(DocumentType).where(DocumentType.label == label))).one()
|
|
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 _api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestClient, str]]:
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / db_filename)),
|
|
environment="test",
|
|
bootstrap_schema_on_startup=True,
|
|
upload_dir=tmp_path / "uploads",
|
|
prompt_dir=tmp_path / "prompts",
|
|
)
|
|
db_url = get_database_url(settings)
|
|
session_factory = None
|
|
|
|
async def _bootstrap() -> None:
|
|
nonlocal session_factory
|
|
from transcription.db.session import get_session_factory
|
|
|
|
session_factory = get_session_factory(database_url=db_url)
|
|
await create_all(engine=get_engine(db_url))
|
|
|
|
asyncio.run(_bootstrap())
|
|
|
|
service = DocumentService(session_factory=session_factory)
|
|
people_service = PeopleService(session_factory=session_factory)
|
|
app = FastAPI()
|
|
register_error_handlers(app)
|
|
app.include_router(router)
|
|
app.dependency_overrides[get_document_service] = lambda: service
|
|
app.dependency_overrides[get_people_service] = lambda: people_service
|
|
|
|
try:
|
|
with TestClient(app) as client:
|
|
yield client, db_url
|
|
finally:
|
|
asyncio.run(dispose_session_factory(db_url))
|
|
|
|
|
|
def test_list_document_types_returns_seeded_registry(tmp_path):
|
|
with _api_client(tmp_path, db_filename="api-types.db") as (client, _db_url):
|
|
response = client.get("/api/document-types")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
labels = {item["label"] for item in payload}
|
|
assert {"Book", "Letter", "Postcard", "Photo", "Journal", "Form"}.issubset(labels)
|
|
|
|
|
|
def test_list_person_roles_returns_seeded_registry(tmp_path):
|
|
with _api_client(tmp_path, db_filename="api-roles.db") as (client, _db_url):
|
|
response = client.get("/api/person-roles")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
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 _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="Form"))
|
|
response = client.put(
|
|
f"/api/documents/{document_id}/type",
|
|
json={"document_type_id": str(type_id)},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["document_id"] == str(document_id)
|
|
assert payload["document_type_id"] == str(type_id)
|
|
|
|
|
|
def test_document_type_payload_requires_uuid_only(tmp_path):
|
|
with _api_client(tmp_path, db_filename="api-doc-type-validation.db") as (client, db_url):
|
|
document_id, _ = _seed_document_and_person(db_url=db_url)
|
|
|
|
missing = client.put(f"/api/documents/{document_id}/type", json={})
|
|
invalid = client.put(
|
|
f"/api/documents/{document_id}/type",
|
|
json={"document_type_id": "record"},
|
|
)
|
|
unexpected = client.put(
|
|
f"/api/documents/{document_id}/type",
|
|
json={"document_type_id": str(UUID(int=1)), "ignored": True},
|
|
)
|
|
|
|
assert missing.status_code == 422
|
|
assert invalid.status_code == 422
|
|
assert unexpected.status_code == 422
|
|
|
|
|
|
def test_document_people_role_aware_write_read_and_delete(tmp_path):
|
|
with _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/documents/{document_id}/people",
|
|
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_id"] == str(author_id)
|
|
assert created["role_label"] == "Author"
|
|
|
|
link_id = created["id"]
|
|
update_response = client.patch(
|
|
f"/api/document-people/{link_id}",
|
|
json={"role_id": str(recipient_id)},
|
|
)
|
|
assert update_response.status_code == 200
|
|
updated = update_response.json()
|
|
assert updated["role_id"] == str(recipient_id)
|
|
assert updated["role_label"] == "Recipient"
|
|
|
|
list_response = client.get(f"/api/documents/{document_id}/people")
|
|
assert list_response.status_code == 200
|
|
links = list_response.json()["links"]
|
|
assert len(links) == 1
|
|
assert links[0]["role_id"] == str(recipient_id)
|
|
|
|
delete_response = client.delete(f"/api/document-people/{link_id}")
|
|
assert delete_response.status_code == 204
|
|
|
|
list_after_delete = client.get(f"/api/documents/{document_id}/people")
|
|
assert list_after_delete.status_code == 200
|
|
assert list_after_delete.json()["links"] == []
|
|
|
|
|
|
def test_document_person_link_requires_role_id(tmp_path):
|
|
with _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)
|
|
|
|
response = client.post(
|
|
f"/api/documents/{document_id}/people",
|
|
json={"person_id": str(person_id)},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
payload = response.json()
|
|
assert "detail" in payload
|
|
assert isinstance(payload["detail"], list)
|
|
missing_role = [
|
|
item
|
|
for item in payload["detail"]
|
|
if isinstance(item, dict) and item.get("loc") == ["body", "role_id"] and item.get("type") == "missing"
|
|
]
|
|
assert missing_role
|
|
|
|
|
|
def test_duplicate_document_person_link_returns_conflict_envelope(tmp_path):
|
|
with _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/documents/{document_id}/people",
|
|
json={"person_id": str(person_id), "role_id": str(author_id)},
|
|
)
|
|
assert first.status_code == 200
|
|
|
|
second = client.post(
|
|
f"/api/documents/{document_id}/people",
|
|
json={"person_id": str(person_id), "role_id": str(recipient_id)},
|
|
)
|
|
|
|
assert second.status_code == 409
|
|
payload = second.json()
|
|
assert payload["category"] == "conflict"
|