generated from john/python-template
updates
This commit is contained in:
+12
-8
@@ -13,11 +13,12 @@ from sqlmodel.pool import StaticPool
|
||||
|
||||
from transcription.config import Settings
|
||||
from transcription.config import get_settings
|
||||
from transcription.db.engine import get_database_url
|
||||
from transcription.db.engine import get_engine
|
||||
from transcription.db.operations import create_all
|
||||
from transcription.db.runtime import dispose_database_runtime
|
||||
from transcription.db.runtime import get_engine
|
||||
from transcription.db.runtime import get_session
|
||||
from transcription.db.runtime import get_session_factory
|
||||
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
|
||||
|
||||
@@ -39,23 +40,26 @@ def session():
|
||||
async def default_settings():
|
||||
"""Provide default settings for tests."""
|
||||
settings = get_settings(database_url="sqlite:///:memory:")
|
||||
await create_all(engine=get_engine(settings=settings))
|
||||
db_url = get_database_url(settings)
|
||||
await create_all(engine=get_engine(database_url=db_url))
|
||||
return settings
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def async_session(default_settings: Settings):
|
||||
"""Provide a clean asynchronous database session for async tests."""
|
||||
async with get_session(settings=default_settings) as async_session:
|
||||
db_url = get_database_url(default_settings)
|
||||
async with session_scope(database_url=db_url) as async_session:
|
||||
yield async_session
|
||||
|
||||
await dispose_database_runtime()
|
||||
await dispose_session_factory(db_url)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def default_session_factory(default_settings: Settings):
|
||||
"""Provide a base fixture for tests that require database access."""
|
||||
session_factory = get_session_factory(settings=default_settings)
|
||||
db_url = get_database_url(default_settings)
|
||||
session_factory = get_session_factory(database_url=db_url)
|
||||
return session_factory
|
||||
|
||||
|
||||
|
||||
+5
-4
@@ -4,17 +4,18 @@ import pytest
|
||||
from sqlalchemy import inspect
|
||||
|
||||
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 get_session
|
||||
from transcription.db import initialize_database_runtime
|
||||
from transcription.db import session_scope
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_all_creates_expected_tables(tmp_path):
|
||||
settings = Settings(
|
||||
openrouter_api_key="test-key",
|
||||
database_url=f"sqlite:///{tmp_path / 'schema.db'}",
|
||||
database=SqliteSettings(path=str(tmp_path / "schema.db")),
|
||||
environment="test",
|
||||
)
|
||||
runtime = initialize_database_runtime(settings=settings)
|
||||
@@ -36,13 +37,13 @@ async def test_create_all_creates_expected_tables(tmp_path):
|
||||
async def test_get_session_yields_async_session(tmp_path):
|
||||
settings = Settings(
|
||||
openrouter_api_key="test-key",
|
||||
database_url=f"sqlite:///{tmp_path / 'session.db'}",
|
||||
database=SqliteSettings(path=str(tmp_path / "session.db")),
|
||||
environment="test",
|
||||
)
|
||||
initialize_database_runtime(settings=settings)
|
||||
|
||||
try:
|
||||
async with get_session(settings=settings) as session:
|
||||
async with session_scope(settings=settings) as session:
|
||||
assert session is not None
|
||||
finally:
|
||||
await dispose_database_runtime()
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Generator
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
@@ -14,10 +15,10 @@ from sqlmodel import delete
|
||||
|
||||
from transcription.app import create_app
|
||||
from transcription.config import Settings
|
||||
from transcription.config import _settings
|
||||
from transcription.config import SqliteSettings
|
||||
from transcription.db import create_all
|
||||
from transcription.db import get_session
|
||||
from transcription.db import initialize_database_runtime
|
||||
from transcription.db import session_scope
|
||||
from transcription.db.models import Document
|
||||
from transcription.db.models import Job
|
||||
from transcription.db.models import JobStatus
|
||||
@@ -28,18 +29,17 @@ RevisionSeed = str
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def app_client(tmp_path_factory: pytest.TempPathFactory) -> tuple[FastAPI, TestClient]:
|
||||
def app_client(tmp_path_factory: pytest.TempPathFactory) -> Generator[tuple[FastAPI, TestClient]]:
|
||||
"""Provide a real application and test client backed by in-memory SQLite."""
|
||||
tmp_path = tmp_path_factory.mktemp("ui")
|
||||
settings = Settings(
|
||||
openrouter_api_key="test-key",
|
||||
database_url="sqlite:///:memory:",
|
||||
database=SqliteSettings(path=":memory:"),
|
||||
environment="test",
|
||||
bootstrap_schema_on_startup=True,
|
||||
upload_dir=tmp_path / "uploads",
|
||||
prompt_dir=tmp_path / "prompts",
|
||||
)
|
||||
_settings.set(settings)
|
||||
|
||||
app = create_app()
|
||||
app.state.runtime = initialize_database_runtime(settings=settings)
|
||||
@@ -54,7 +54,7 @@ def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None:
|
||||
app, _ = app_client
|
||||
|
||||
async def _clear() -> None:
|
||||
async with get_session(session_factory=app.state.runtime.session_factory) as session:
|
||||
async with session_scope() as session:
|
||||
await session.exec(delete(Revision))
|
||||
await session.exec(delete(Source))
|
||||
await session.exec(delete(Job))
|
||||
@@ -80,7 +80,7 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
|
||||
source_file: Path | None = None,
|
||||
) -> UUID:
|
||||
async def _insert() -> UUID:
|
||||
async with get_session(session_factory=app.state.runtime.session_factory) as session:
|
||||
async with session_scope() as session:
|
||||
stored_path = app.state.settings.upload_dir / filename
|
||||
stored_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
source_path = source_file or fixtures_dir / "small_png.png"
|
||||
|
||||
Reference in New Issue
Block a user