generated from john/python-template
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""Tests for the database runtime and V2 schema bootstrap behavior."""
|
|
|
|
import warnings
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy import text
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.exc import SAWarning
|
|
from sqlmodel import SQLModel
|
|
from sqlmodel import select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from transcription.config import Settings
|
|
from transcription.config import SqliteSettings
|
|
from transcription.db import create_all
|
|
from transcription.db import dispose_database_runtime
|
|
from transcription.db import initialize_database_runtime
|
|
from transcription.db import normalize_legacy_status_spellings
|
|
from transcription.db import reconcile_legacy_job_source_columns
|
|
from transcription.db import session_scope
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentType
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import PersonRole
|
|
from transcription.db.models import Source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_all_creates_expected_tables(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "schema.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
await create_all(engine=runtime.engine)
|
|
async with runtime.engine.connect() as conn:
|
|
table_names = set(await conn.run_sync(lambda c: inspect(c).get_table_names()))
|
|
|
|
assert "document" in table_names
|
|
assert "document_type" in table_names
|
|
assert "person" in table_names
|
|
assert "person_role" in table_names
|
|
assert "document_person" in table_names
|
|
assert "job" in table_names
|
|
assert "source" in table_names
|
|
assert "job_source" in table_names
|
|
assert "execution_attempt" in table_names
|
|
assert "revision" not in table_names
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_yields_async_session(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "session.db")),
|
|
environment="test",
|
|
)
|
|
initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
async with session_scope(settings=settings) as session:
|
|
assert session is not None
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_rejects_reinitialization_for_different_database(tmp_path):
|
|
"""An existing process runtime cannot silently switch database targets."""
|
|
first_settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "first.db")),
|
|
environment="test",
|
|
)
|
|
second_settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "second.db")),
|
|
environment="test",
|
|
)
|
|
initialize_database_runtime(settings=first_settings)
|
|
|
|
try:
|
|
with pytest.raises(RuntimeError, match="already initialized for a different database"):
|
|
initialize_database_runtime(settings=second_settings)
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_all_seeds_default_registry_rows(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "seed.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
await create_all(engine=runtime.engine)
|
|
async with AsyncSession(runtime.engine, expire_on_commit=False) as session:
|
|
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_keys)
|
|
assert {"book", "letter", "postcard", "photo", "journal", "form"}.issubset(type_keys)
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_all_declares_hot_path_indexes(tmp_path):
|
|
"""Worker and detail-page filters must be index-backed in a freshly created schema."""
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "indexes.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
await create_all(engine=runtime.engine)
|
|
async with runtime.engine.connect() as connection:
|
|
|
|
def collect(sync_connection) -> tuple[dict[str, list[list[str]]], list[list[str]]]:
|
|
database = inspect(sync_connection)
|
|
indexes = {
|
|
table: [index["column_names"] for index in database.get_indexes(table)]
|
|
for table in ("job", "source", "job_source", "document", "document_person")
|
|
}
|
|
job_source_unique = [
|
|
constraint["column_names"]
|
|
for constraint in database.get_unique_constraints("job_source")
|
|
]
|
|
return indexes, job_source_unique
|
|
|
|
indexes, job_source_unique = await connection.run_sync(collect)
|
|
|
|
assert ["status", "date_created"] in indexes["job"]
|
|
assert ["document_id"] in indexes["job"]
|
|
assert ["document_id"] in indexes["source"]
|
|
assert ["preferred_execution_attempt_id"] in indexes["source"]
|
|
assert ["job_id"] in indexes["job_source"]
|
|
assert ["source_id"] in indexes["job_source"]
|
|
assert ["job_id", "source_id"] in job_source_unique
|
|
assert ["document_type_id"] in indexes["document"]
|
|
for column in ("document_id", "person_id", "role_id"):
|
|
assert [column] in indexes["document_person"]
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalize_legacy_status_spellings_repairs_job_source_status_rows(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "legacy-status.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
await create_all(engine=runtime.engine)
|
|
async with AsyncSession(runtime.engine, expire_on_commit=False) as session:
|
|
document = Document(name="legacy-status-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="legacy.jpg",
|
|
filename="legacy.jpg",
|
|
file_path="uploads/legacy.jpg",
|
|
file_hash="a" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
job_source = JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
|
|
session.add(job_source)
|
|
await session.commit()
|
|
await session.refresh(job_source)
|
|
|
|
async with runtime.engine.begin() as connection:
|
|
await connection.execute(
|
|
text('update "job_source" set status = :status where status = :expected'),
|
|
{"status": "TRANSCRIBED", "expected": JobSourceStatus.PENDING.value},
|
|
)
|
|
|
|
fixed_rows = await normalize_legacy_status_spellings(engine=runtime.engine)
|
|
assert fixed_rows == 1
|
|
|
|
async with runtime.engine.connect() as connection:
|
|
status = (
|
|
await connection.execute(
|
|
text('select status from "job_source"'),
|
|
)
|
|
).scalar_one()
|
|
assert status == "transcribed"
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconcile_legacy_job_source_columns_drops_executed_at(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "legacy-column.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
await create_all(engine=runtime.engine)
|
|
async with runtime.engine.begin() as connection:
|
|
await connection.execute(text("PRAGMA foreign_keys=OFF"))
|
|
await connection.execute(text('alter table "job_source" rename to "job_source_current"'))
|
|
await connection.execute(
|
|
text(
|
|
'create table "job_source" ('
|
|
'id char(32) not null primary key, '
|
|
'job_id char(32) not null, '
|
|
'source_id char(32) not null, '
|
|
'status varchar(11) not null, '
|
|
'executed_at datetime not null, '
|
|
'constraint "uq_job_source_job_source" unique ("job_id", "source_id"), '
|
|
'foreign key("job_id") references "job" ("id"), '
|
|
'foreign key("source_id") references "source" ("id")'
|
|
")"
|
|
)
|
|
)
|
|
await connection.execute(text('drop table "job_source_current"'))
|
|
await connection.execute(text("PRAGMA foreign_keys=ON"))
|
|
|
|
dropped = await reconcile_legacy_job_source_columns(engine=runtime.engine)
|
|
assert dropped == 1
|
|
|
|
async with runtime.engine.connect() as connection:
|
|
columns = await connection.run_sync(lambda c: [col["name"] for col in inspect(c).get_columns("job_source")])
|
|
assert "executed_at" not in columns
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
def test_metadata_has_no_unresolvable_table_cycle():
|
|
"""create_all must be able to order every table, including on PostgreSQL."""
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("error", SAWarning)
|
|
ordered = [table.name for table in SQLModel.metadata.sorted_tables]
|
|
|
|
assert ordered.index("source") < ordered.index("execution_attempt")
|
|
|
|
|
|
def test_preferred_execution_attempt_id_column_matches_model_declaration():
|
|
"""The self-referential provenance column is a real UUID, not an opaque CHAR(32)."""
|
|
column = SQLModel.metadata.tables["source"].c["preferred_execution_attempt_id"]
|
|
|
|
assert isinstance(column.type, sa.Uuid)
|
|
assert column.type.compile(dialect=postgresql.dialect()) == "UUID"
|
|
assert Source.model_fields["preferred_execution_attempt_id"].annotation is not None
|
|
|
|
foreign_key = next(iter(column.foreign_keys))
|
|
assert foreign_key.use_alter is True
|
|
assert foreign_key.column is SQLModel.metadata.tables["execution_attempt"].c["id"]
|
|
|
|
|
|
def test_bootstrap_policy_production_defaults_false():
|
|
settings = Settings(openrouter_api_key="test-key", environment="production")
|
|
assert settings.should_bootstrap_schema is False
|
|
|
|
|
|
def test_bootstrap_policy_development_defaults_true():
|
|
settings = Settings(openrouter_api_key="test-key", environment="development")
|
|
assert settings.should_bootstrap_schema is True
|
|
|
|
|
|
def test_bootstrap_policy_explicit_override_true():
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="production",
|
|
bootstrap_schema_on_startup=True,
|
|
)
|
|
assert settings.should_bootstrap_schema is True
|