test db setup

alembic/env.py

@@ -5,6 +5,7 @@ from dotenv import load_dotenv
 from sqlalchemy import engine_from_config, pool
 
 from alembic import context
+from src.db import get_set_engine
 from src.models_db import BaseModelDb
 
 load_dotenv()  # reads .env into os.environ
@@ -64,6 +65,9 @@ def run_migrations_online() -> None:
     and associate a connection with the context.
 
     """
+    if get_set_engine() is not None:
+        connectable = get_set_engine()
+    else:
         connectable = engine_from_config(
             config.get_section(config.config_ini_section, {}),
             prefix="sqlalchemy.",

src/db.py

@@ -9,6 +9,18 @@ from sqlalchemy.orm import Session, sessionmaker
 
 DATABASE_URL_DEFAULT = "sqlite:///finanz.db"
 
+# Optional: allow an external engine (e.g. for testing with in-memory SQLite)
+_engine: Engine | None = None
+
+
+def set_engine(engine: Engine) -> None:
+    global _engine
+    _engine = engine
+
+
+def get_set_engine() -> Engine | None:
+    return _engine
+
 
 def get_engine() -> Engine:
     url = os.environ.get("DATABASE_URL", DATABASE_URL_DEFAULT)

tests/conftest.py

@@ -0,0 +1,36 @@
+#!/bin/python
+
+import os
+from collections.abc import Generator
+
+os.environ["DATABASE_URL"] = "sqlite://"
+
+import pytest
+from alembic.config import Config
+from sqlalchemy import StaticPool, create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from alembic import command
+from src.db import set_engine
+
+
+@pytest.fixture()
+def db_session() -> Generator[Session]:
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+
+    set_engine(engine)
+
+    alembic_cfg = Config("alembic.ini")
+    command.upgrade(alembic_cfg, "head")
+
+    SessionLocal = sessionmaker(bind=engine)
+    session = SessionLocal()
+    try:
+        yield session
+    finally:
+        session.close()
+        engine.dispose()

tests/test_db.py

@@ -0,0 +1,10 @@
+#!/bin/python
+
+from sqlalchemy import inspect
+from sqlalchemy.orm import Session
+
+
+def test_db_reachable_and_not_empty(db_session: Session):
+    inspector = inspect(db_session.get_bind())
+    tables = inspector.get_table_names()
+    assert tables