generated from john/python-template
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
"""Shared test fixtures.
|
|
|
|
Every test gets a fresh in-memory SQLite database so tests are
|
|
isolated, fast, and leave no artifacts on disk.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlmodel import Session
|
|
from sqlmodel import SQLModel
|
|
from sqlmodel import create_engine
|
|
from sqlmodel.pool import StaticPool
|
|
|
|
from transcription.config import Settings
|
|
from transcription.config import SqliteSettings
|
|
from transcription.db.engine import get_database_url
|
|
from transcription.db.engine import get_engine
|
|
from transcription.db.session import dispose_session_factory
|
|
from transcription.db.session import get_session_factory
|
|
from transcription.db.session import session_scope
|
|
from transcription.services.documents import DocumentService
|
|
from transcription.services.jobs import JobService
|
|
|
|
|
|
@pytest.fixture
|
|
def session():
|
|
"""Provide a clean synchronous database session for sync tests."""
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as sync_session:
|
|
yield sync_session
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def default_settings(tmp_path):
|
|
"""Provide default settings for tests."""
|
|
database_path = tmp_path / "tests.db"
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=SqliteSettings(path=str(database_path)),
|
|
environment="test",
|
|
)
|
|
db_url = get_database_url(settings)
|
|
if Path(str(get_engine(database_url=db_url).url.database)).resolve() != database_path.resolve():
|
|
raise RuntimeError(f"Refusing to initialize destructive test fixtures against {db_url}")
|
|
await dispose_session_factory(db_url)
|
|
engine = get_engine(database_url=db_url)
|
|
|
|
# Cached in-memory engines persist across tests; reset schema per test for isolation.
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(SQLModel.metadata.drop_all)
|
|
await connection.run_sync(SQLModel.metadata.create_all)
|
|
|
|
yield settings
|
|
await dispose_session_factory(db_url)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_session(default_settings: Settings):
|
|
"""Provide a clean asynchronous database session for async tests."""
|
|
db_url = get_database_url(default_settings)
|
|
async with session_scope(database_url=db_url) as async_session:
|
|
yield async_session
|
|
|
|
|
|
@pytest.fixture
|
|
def default_session_factory(default_settings: Settings):
|
|
"""Provide a base fixture for tests that require database access."""
|
|
db_url = get_database_url(default_settings)
|
|
session_factory = get_session_factory(database_url=db_url)
|
|
return session_factory
|
|
|
|
|
|
@pytest.fixture
|
|
def job_service(default_session_factory) -> JobService:
|
|
"""Provide a JobService instance for testing."""
|
|
return JobService(session_factory=default_session_factory)
|
|
|
|
|
|
@pytest.fixture
|
|
def document_service(default_session_factory) -> DocumentService:
|
|
"""Provide a DocumentService instance for testing."""
|
|
return DocumentService(session_factory=default_session_factory)
|