generated from john/python-template
176 lines
6.5 KiB
Python
176 lines
6.5 KiB
Python
"""Integration tests for additive V4 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 transcription.api.errors import register_error_handlers
|
|
from transcription.api.v4_documents import get_document_service
|
|
from transcription.api.v4_documents import get_people_service
|
|
from transcription.api.v4_documents 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 Person
|
|
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)
|
|
person = Person(full_name=person_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())
|
|
|
|
|
|
@contextmanager
|
|
def _v4_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 _v4_api_client(tmp_path, db_filename="api-types.db") as (client, _db_url):
|
|
response = client.get("/api/v4/document-types")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
codes = {item["code"] for item in payload}
|
|
assert {"letter", "record", "memo"}.issubset(codes)
|
|
|
|
|
|
def test_list_person_roles_returns_seeded_registry(tmp_path):
|
|
with _v4_api_client(tmp_path, db_filename="api-roles.db") as (client, _db_url):
|
|
response = client.get("/api/v4/person-roles")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
codes = {item["code"] for item in payload}
|
|
assert {"author", "recipient", "mentioned"}.issubset(codes)
|
|
|
|
|
|
def test_set_document_type_by_code_updates_canonical_fields(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)
|
|
response = client.put(
|
|
f"/api/v4/documents/{document_id}/type",
|
|
json={"document_type_code": "record"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["document_id"] == str(document_id)
|
|
assert payload["document_type_id"] is not None
|
|
assert payload["document_type_code"] == "record"
|
|
|
|
|
|
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)
|
|
|
|
create_response = client.post(
|
|
f"/api/v4/documents/{document_id}/people",
|
|
json={"person_id": str(person_id), "role_code": "author"},
|
|
)
|
|
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
|
|
|
|
link_id = created["id"]
|
|
update_response = client.patch(
|
|
f"/api/v4/document-people/{link_id}",
|
|
json={"role_code": "recipient"},
|
|
)
|
|
assert update_response.status_code == 200
|
|
updated = update_response.json()
|
|
assert updated["role_code"] == "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"
|
|
|
|
delete_response = client.delete(f"/api/v4/document-people/{link_id}")
|
|
assert delete_response.status_code == 204
|
|
|
|
list_after_delete = client.get(f"/api/v4/documents/{document_id}/people")
|
|
assert list_after_delete.status_code == 200
|
|
assert list_after_delete.json()["links"] == []
|
|
|
|
|
|
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)
|
|
|
|
first = client.post(
|
|
f"/api/v4/documents/{document_id}/people",
|
|
json={"person_id": str(person_id), "role_code": "author"},
|
|
)
|
|
assert first.status_code == 200
|
|
|
|
second = client.post(
|
|
f"/api/v4/documents/{document_id}/people",
|
|
json={"person_id": str(person_id), "role_code": "author"},
|
|
)
|
|
|
|
assert second.status_code == 409
|
|
payload = second.json()
|
|
assert payload["category"] == "conflict_error"
|