transcription/tests/test_db.py

"""Tests for the database runtime and V2 schema bootstrap behavior."""

import warnings

import pytest
import sqlalchemy as sa
from sqlalchemy import inspect
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 session_scope
from transcription.db.models import DocumentType
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) -> dict[str, list[list[str]]]:
                database = inspect(sync_connection)
                return {
                    table: [index["column_names"] for index in database.get_indexes(table)]
                    for table in ("job", "source", "job_source", "document", "document_person")
                }

            indexes = 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 ["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()


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