generated from john/python-template
132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""Tests for the database runtime and V2 schema bootstrap behavior."""
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy import text
|
|
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
|
|
|
|
|
|
@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 "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_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_codes = set((await session.exec(select(PersonRole.code))).all())
|
|
type_codes = set((await session.exec(select(DocumentType.code))).all())
|
|
|
|
assert {"author", "recipient", "mentioned"}.issubset(role_codes)
|
|
assert {"letter", "record", "memo"}.issubset(type_codes)
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_all_upgrades_existing_person_table_for_family_search(tmp_path):
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(tmp_path / "upgrade.db")),
|
|
environment="test",
|
|
)
|
|
runtime = initialize_database_runtime(settings=settings)
|
|
|
|
try:
|
|
async with runtime.engine.begin() as connection:
|
|
await connection.execute(
|
|
text("CREATE TABLE person (id CHAR(32) PRIMARY KEY NOT NULL, full_name VARCHAR NOT NULL)")
|
|
)
|
|
|
|
await create_all(engine=runtime.engine)
|
|
async with runtime.engine.connect() as connection:
|
|
columns, indexes = await connection.run_sync(
|
|
lambda sync_connection: (
|
|
{column["name"] for column in inspect(sync_connection).get_columns("person")},
|
|
inspect(sync_connection).get_indexes("person"),
|
|
)
|
|
)
|
|
|
|
assert "family_search_id" in columns
|
|
assert any(
|
|
index["column_names"] == ["family_search_id"] and index["unique"] for index in indexes
|
|
)
|
|
finally:
|
|
await dispose_database_runtime()
|
|
|
|
|
|
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
|