generated from john/python-template
142 lines
5.4 KiB
Python
142 lines
5.4 KiB
Python
"""Tests for transcription.db — schema bootstrap and session factory."""
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlmodel import SQLModel
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from sqlmodel.pool import StaticPool
|
|
|
|
|
|
def _in_memory_engine():
|
|
"""Create a fresh in-memory SQLite engine for isolated db tests."""
|
|
return create_async_engine(
|
|
"sqlite+aiosqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
|
|
|
|
class TestSchemaBootstrap:
|
|
"""Verify create_all produces the expected table set."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_all_creates_expected_tables(self):
|
|
"""After create_all(), document, job, and transcript tables exist."""
|
|
engine = _in_memory_engine()
|
|
# Ensure models are imported so metadata is populated
|
|
import transcription.db as db_module
|
|
from transcription.models import Document # noqa: F401
|
|
from transcription.models import Job # noqa: F401
|
|
from transcription.models import Transcript # noqa: F401
|
|
|
|
await db_module.create_all(engine=engine)
|
|
|
|
async with engine.connect() as connection:
|
|
table_names = set(await connection.run_sync(lambda conn: inspect(conn).get_table_names()))
|
|
await engine.dispose()
|
|
assert "document" in table_names
|
|
assert "job" in table_names
|
|
assert "transcript" in table_names
|
|
|
|
|
|
class TestSessionFactory:
|
|
"""Verify get_session yields and cleans up sessions."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_yields_session(self):
|
|
"""get_session() yields a usable Session object."""
|
|
engine = _in_memory_engine()
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(SQLModel.metadata.create_all)
|
|
|
|
import transcription.db as db_module
|
|
|
|
factory = db_module.async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with db_module.get_session(session_factory=factory) as session:
|
|
assert isinstance(session, AsyncSession)
|
|
await engine.dispose()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_is_closed_after_generator_exit(self):
|
|
"""After the context manager exits, the session is closed."""
|
|
engine = _in_memory_engine()
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(SQLModel.metadata.create_all)
|
|
|
|
import transcription.db as db_module
|
|
|
|
factory = db_module.async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with db_module.get_session(session_factory=factory) as session:
|
|
# Session is usable inside the context
|
|
await session.exec(text("SELECT 1"))
|
|
captured = session
|
|
|
|
assert captured.sync_session._transaction is None
|
|
await engine.dispose()
|
|
|
|
|
|
class TestBootstrapPolicy:
|
|
"""Verify schema bootstrap policy defaults and overrides."""
|
|
|
|
def test_production_defaults_to_no_bootstrap(self):
|
|
"""Production defaults to explicit non-bootstrap startup behavior."""
|
|
from transcription.config import Settings
|
|
from transcription.db import should_bootstrap_schema
|
|
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="production",
|
|
bootstrap_schema_on_startup=None,
|
|
)
|
|
assert should_bootstrap_schema(settings) is False
|
|
|
|
def test_development_defaults_to_bootstrap(self):
|
|
"""Development defaults to schema bootstrap for local workflows."""
|
|
from transcription.config import Settings
|
|
from transcription.db import should_bootstrap_schema
|
|
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="development",
|
|
bootstrap_schema_on_startup=None,
|
|
)
|
|
assert should_bootstrap_schema(settings) is True
|
|
|
|
def test_explicit_override_wins(self):
|
|
"""Explicit bootstrap_schema_on_startup overrides environment default."""
|
|
from transcription.config import Settings
|
|
from transcription.db import should_bootstrap_schema
|
|
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
environment="production",
|
|
bootstrap_schema_on_startup=True,
|
|
)
|
|
assert should_bootstrap_schema(settings) is True
|
|
|
|
|
|
class TestRuntimeSqliteEngineConfig:
|
|
"""Verify sqlite runtime engine configuration from settings."""
|
|
|
|
def test_in_memory_sqlite_uses_static_pool_by_default(self):
|
|
"""In-memory sqlite URLs get StaticPool for consistent connections."""
|
|
from transcription.config import Settings
|
|
from transcription.db.runtime import _build_engine
|
|
|
|
settings = Settings(openrouter_api_key="test-key", database_url="sqlite://")
|
|
engine = _build_engine(settings)
|
|
|
|
assert engine.sync_engine.pool.__class__.__name__ == "StaticPool"
|
|
|
|
def test_file_sqlite_does_not_force_static_pool(self):
|
|
"""File-backed sqlite URLs keep default pool behavior."""
|
|
from transcription.config import Settings
|
|
from transcription.db.runtime import _build_engine
|
|
|
|
settings = Settings(openrouter_api_key="test-key", database_url="sqlite:///./runtime-test.db")
|
|
engine = _build_engine(settings)
|
|
|
|
assert engine.sync_engine.pool.__class__.__name__ != "StaticPool"
|