[Python] How to mock the Database (SQLModel + FastAPI)

I want to test my GraphQL endpoint. To do this, I have to mock the database.

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from contextlib import asynccontextmanager
import os


class Database:
def __init__(self) -> None:
self.engine = None

def connect(self) -> None:
if not self.engine:
self.engine = create_async_engine(os.environ.get("DATABASE_URL", "sqlite:///db.sqlite")
)
else:
raise ConnectionAbortedError("Database is already connected")

@asynccontextmanager
async def get_async_session(self):
if not self.engine:
raise ConnectionError("Database is not connected")

async with AsyncSession(self.engine) as session:
try:
yield session
finally:
await session.close()


I've tried to mock create_engine in my test,

async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine


in the app fixture

@pytest.fixture(scope="session", autouse=True)
def app() -> FastAPI:
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)

with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine

from daytistics.main import app

return app


and I've to mock the Session

async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.Session", autospec=True) as mock_session_class:
mock_session_class.return_value = Session(engine)


but each time it was written to the production database.

I don't if it's useful, but this is my app factory:

def create_app() -> FastAPI:
dotenv.load_dotenv()
app = FastAPI()

register_dependencies()

with container.sync_context() as ctx:
db = ctx.resolve(Database)
db.connect()

graphql_router = create_graphql_router()
app.include_router(graphql_router, prefix="/graphql")

return app


app = create_app()


Update:

I've tried to use a monkeypatch and it did not work either:

@pytest.fixture(scope="function")
async def mock_db(monkeypatch):
test_engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)

async with test_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)

def mock_connect(self):
self.engine = test_engine

@asynccontextmanager
async def mock_get_async_session(self):
async with AsyncSession(test_engine) as session:
try:
yield session
finally:
await session.close()

monkeypatch.setattr(Database, "connect", mock_connect)
monkeypatch.setattr(Database, "get_async_session", mock_get_async_session)

return test_engine

Continue reading...