e178ae406d
- SQLModel tables: flow_definitions/versions/runs, approval_tasks - Alembic setup with autogenerate (sqlmodel types fixed via script.py.mako) - Settings extended (shared root .env), db engine/session helpers - pytest suite: conftest with isolated test DB, 5 model tests green
69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
"""Shared pytest fixtures.
|
|
|
|
Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
|
|
`kirche_einfach_test` database from the compose stack). The URL is mapped
|
|
onto DATABASE_URL *before* any app module is imported, so settings and the
|
|
alembic env both target the test database. Never point this at production.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
TEST_DB_URL = os.environ.get("TEST_DATABASE_URL")
|
|
if not TEST_DB_URL:
|
|
raise RuntimeError(
|
|
"TEST_DATABASE_URL is not set. Refusing to run tests without "
|
|
"an explicit test database (see .env.example)."
|
|
)
|
|
|
|
# Must precede all `app.*` imports: Settings is instantiated at import time.
|
|
os.environ["DATABASE_URL"] = TEST_DB_URL
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def db_engine():
|
|
"""Engine on the test DB, migrated to head once per session."""
|
|
from alembic import command as alembic_command
|
|
from alembic.config import Config
|
|
|
|
from app import db as app_db
|
|
|
|
app_db.reset_engine()
|
|
engine = app_db.get_engine()
|
|
|
|
cfg = Config()
|
|
cfg.set_main_option("script_location", str(BACKEND_DIR / "migrations"))
|
|
alembic_command.upgrade(cfg, "head")
|
|
|
|
yield engine
|
|
|
|
engine.dispose()
|
|
app_db.reset_engine()
|
|
|
|
|
|
@pytest.fixture()
|
|
def session(db_engine):
|
|
"""Function-scoped session; tables are truncated after each test."""
|
|
from sqlmodel import Session, SQLModel
|
|
|
|
with Session(db_engine) as s:
|
|
yield s
|
|
with db_engine.begin() as conn:
|
|
for table in reversed(SQLModel.metadata.sorted_tables):
|
|
conn.execute(table.delete())
|
|
|
|
|
|
@pytest.fixture()
|
|
async def client():
|
|
"""ASGI test client for the FastAPI app."""
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.main import app
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|