How to mock dependencies?

[kayprogrammer]

Summary

Can you please add a dependency override to the application. Something similar to fastapi's app.dependency_overrides.
This is important when trying to override a dependency while writing tests.

Basic Example

@pytest.fixture
def client(app):
    database = TestAsyncSessionLocal()
    def test_get_db(app):
        try:
            yield database
        finally:
            database.close()
    app.dependency_overrides['db'] = Provide(test_get_db) # nothing like dependency_overrides in starlite
    ...

Drawbacks and Impact

Here's what I tried:

def client(app):
    database = TestAsyncSessionLocal()
    def test_get_db(app):
        try:
            yield database
        finally:
            database.close()
    app.dependencies['db'] = Provide(test_get_db)
    ...

This doesn't work because handlers are already registered in main:app. So overriding dependencies like this after handlers have been registered changes nothing.

Unresolved questions

No response

[Goldziher]

Hi. It's basically already in place. The create_test_client helper allows you to pass any dependencies you want top level.

For controllers you can use attribute mocking.

What's missing?

[kayprogrammer]

Hi. It's basically already in place. The create_test_client helper allows you to pass any dependencies you want top level.

For controllers you can use attribute mocking.

What's missing?

Thanks for your reply. But the create_test_client was built upon TestClient, and I want to use AsyncTestClient.
Also, can you also show me an example on how I can use mocking for the controllers please. Thank you.

[provinzkraut]

Expanding on what @Goldziher said: It's unlikely that we'll support this for a number of reasons (which we actually cover here in our FastAPI migration guide).

The gist is that it's not actually needed and often a crude workaround that has more drawbacks than benefits. Overriding dependencies in such a way:

1. Breaks test isolation if you're not using an application factory, since it's modifying global state (the application object). You then need to keep track of this and make sure to roll back the changes you made
2. It does not offer any functional benefits over mocking / monkeypatching the dependency callable
3. It is less granular and can easily break. This is because you can define dependencies on every application layer, but overrides like this are global. Since overrides are already possible on the layers, you'd effectively need to override all possible overrides, thereby changing the way your application actually works, which means your tests aren't testing your actual application, but what you think your application works like

The solution to your problem is this:

@pytest.fixture
def database():
    database = TestAsyncSessionLocal()
    try:
        yield database
    finally:
        database.close()

@pytest.fixture
def client(app, mocker, database):
  mocker.patch("path.to.get_db", return_value=database)

This actually has several benefits:

1. You now have an independent database fixture you can use in your tests
2. You do not modify the way your application works
3. You can easily target the dependency on multiple layers should you override it somewhere
4. No special functionality is involved, so you can access all the pytest features like cleanup, parametrisation, marking, selection, etc.

If you do need to actually swap out the whole dependency function because you're performing some additional logic in there you can easily adapt this pattern:

def get_database(app):
    database = TestAsyncSessionLocal()
    try:
        yield database
    finally:
        database.close()


@pytest.fixture
def client(app, mocker, database):
  mocker.patch("path.to.get_db", new=get_database)

Just to illustrate what the proper way to override dependencies like you want would look like in contrast:

# add a separate fixture for the database to ensure it's always closed
# no matter where errors might occur
@pytest.fixture
def database()
    database = TestAsyncSessionLocal()
    try:
        yield database
    finally:
        if database.is_open:
          database.close()

@pytest.fixture
def client(app, database):
    def test_get_db(app):
        try:
            yield database
        finally:
            database.close()

    try:
      app.dependency_overrides['db'] = Provide(test_get_db)
    finally:
      app.dependency_overrides = {} # ensure to reset overrides at the end

[provinzkraut]

Thanks for your reply. But the create_test_client was built upon TestClient, and I want to use AsyncTestClient.

create_async_test_client has the same signature as create_test_client.

[kayprogrammer]

Thanks for your reply. I tried your solution using mock, but it doesn't work. The test still seems to be using my original db instead of the test_db. Here's how my app looks in the main.py:

app = Starlite(
        dependencies={"db": Provide(get_db)},
        route_handlers=all_routers,
        openapi_config=openapi_config,
        template_config=template_config,
       
        exception_handlers={
            ValidationException: validation_exception_handler,
            HTTPException: http_exception_handler,
            HTTP_500_INTERNAL_SERVER_ERROR: internal_server_error_handler,
        },
    )

Please help

[provinzkraut]

Can you provide an example of your app, the get_db dependency and the test case that's not working as expected? It's impossible to diagnose your issue just from the information you provided.

[kayprogrammer]

Can you provide an example of your app, the get_db dependency and the test case that's not working as expected? It's impossible to diagnose your issue just from the information you provided.

app.main.py

from starlite import (
    Starlite,
    TemplateConfig,
    get,
    Provide,
    OpenAPIConfig,
    OpenAPIController,
    HTTPException,
    ValidationException,
)

from starlite.status_codes import HTTP_500_INTERNAL_SERVER_ERROR
from starlite.contrib.jinja import JinjaTemplateEngine

from app.core.config import settings
from app.core.database import get_db
from app.common.exception_handlers import (
    validation_exception_handler,
    http_exception_handler,
    internal_server_error_handler,
)
from app.api.routers import all_routers

from pathlib import Path


class MyOpenAPIController(OpenAPIController):
    path = "/"


def create_app() -> Starlite:
    openapi_config = OpenAPIConfig(
        title=settings.PROJECT_NAME,
        version="3.0.0",
        description="A simple bidding API built with Litestar",
        security=[{"BearerToken": []}],
        openapi_controller=MyOpenAPIController,
        root_schema_site="swagger",
    )

    template_config = TemplateConfig(
        directory=Path("app/templates"),
        engine=JinjaTemplateEngine,
    )

    app = Starlite(
        dependencies={"db": Provide(get_db)},
        route_handlers=all_routers,
        openapi_config=openapi_config,
        template_config=template_config,
       
        exception_handlers={
            ValidationException: validation_exception_handler,
            HTTPException: http_exception_handler,
            HTTP_500_INTERNAL_SERVER_ERROR: internal_server_error_handler,
        },
    )
    return app

app = create_app()

@get(
    "/ping",
    tags=["HealthCheck"],
    summary="API Health Check",
    description="This endpoint checks the health of the API",
)
async def healthcheck() -> dict[str, str]:
    return {"success": "pong!"}


app.register(healthcheck, add_to_openapi_schema=True)

app.api.tests.conftest.py

from starlite import AsyncTestClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.core.database import Base
from app.core.config import settings
from app.core.security import get_password_hash
from app.db.managers.accounts import user_manager
from app.main import app

import pytest_asyncio
import asyncio

TEST_DATABASE = f"{settings.SQLALCHEMY_DATABASE_URL}_test"

engine = create_async_engine(TEST_DATABASE, future=True)

TestAsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
)


@pytest_asyncio.fixture(scope='session')
def event_loop():
    """Overrides pytest default function scoped event loop"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest_asyncio.fixture
async def database():
    db = TestAsyncSessionLocal()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    try:
        yield db
    finally:
        db.close()

@pytest_asyncio.fixture
async def client(database, mocker):
    mocker.patch("app.core.database.get_db", return_value=database)
    async with AsyncTestClient(app) as client:
        yield client

@pytest_asyncio.fixture
async def test_user(database):
    user_dict = {
        "first_name": "Test",
        "last_name": "Name",
        "email": "[email protected]",
        "password": get_password_hash("testpassword"),
    }
    user = await user_manager.create(database, user_dict)
    return user


@pytest_asyncio.fixture
async def verified_user(database):
    user_dict = {
        "first_name": "Test",
        "last_name": "Name",
        "email": "[email protected]",
        "password": get_password_hash("testpassword"),
        "is_email_verified": True,
    }
    user = await user_manager.create(database, user_dict)
    return user

app.core.database.py

from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from sqlalchemy.ext.asyncio import AsyncSession

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from .config import settings

sqlalchemy_config = SQLAlchemyConfig(
    connection_string=settings.SQLALCHEMY_DATABASE_URL,
    dependency_key="db",
)

sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()

AsyncSessionLocal = sessionmaker(
    bind=sqlalchemy_config.engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
)


async def get_db():
    db = AsyncSessionLocal()
    try:
        yield db
    finally:
        await db.close()

Thank you.

[provinzkraut]

A couple of things here, but let me just say that this isn't something Litestar specific but rather about how scopes, imports and mocking works.

1. You are misusing the application factory pattern. Having a create_app function and then immediately calling it in the same file is functionally equivalent to simply creating the app in the file directly.
2. The issue you're seeing is because of the order of execution. What happens in your code is:
3. You import app. This will also set the dependencies on that object
4. You mock app.core.database.get_db. Since the application object has already been created, this will not change the get_db function you passed to it

To fix it you can either:

1. Mock get_db before you import app (requires the least amount of effort from your side)
2. Fix your broken patterns by not calling create_app immediately and instead importing it and calling it after you mocked get_db

The second approach is the preferred one and one of the reasons we don't provide dependency overrides: It encourages bad practices. As you've seen, the fact that Litestar does not provide you with such capabilities has brought to light a deeper issue in your code.

Examples

Here are two simplified examples that you can apply to your existing code:

Importing the app locally

dependencies.py

def some_dependency() -> str:
  return "foo"

main.py

from starlite import Starlite, Provide, get
from dependencies import some_dependency


@get()
def handler(dep: str) -> str:
  return dep


app = Starlite([handler], dependencies={"dep": Provide(some_dependency)})

test.py

from starlite import TestClient


def new_dependency():
    return "bar"


def test(mocker):
    mocker.patch("dependencies.some_dependency", new=new_dependency)
    from main import app

    with TestClient(app) as client:
        assert client.get("/").text == "bar"

Using an application factory

dependencies.py

def some_dependency() -> str:
  return "foo"

main.py

from starlite import Starlite, Provide, get
from dependencies import some_dependency


@get()
def handler(dep: str) -> str:
  return dep

def create_app() -> Starlite:
  return Starlite([handler], dependencies={"dep": Provide(some_dependency)})

test.py

from starlite import TestClient
from main import create_app
import pytest


def new_dependency():
    return "bar"

@pytest.fixture(autouse=True)
def mock_db():
    return mocker.patch("dependencies.some_dependency", new=new_dependency)


@pytest.fixture
def app():
  return create_app()


def test(app):
    with TestClient(app) as client:
        assert client.get("/").text == "bar"

[kayprogrammer]

A couple of things here, but let me just say that this isn't something Litestar specific but rather about how scopes, imports and mocking works.

1. You are misusing the application factory pattern. Having a `create_app` function and then immediately calling it in the same file is functionally equivalent to simply creating the app in the file directly.

2. The issue you're seeing is because of the order of execution. What happens in your code is:

3. You import `app`. This will also set the dependencies on that object

4. You mock `app.core.database.get_db`. Since the application object has already been created, this will not change the `get_db` function you passed to it

To fix it you can either:

1. Mock `get_db` before you import `app` (requires the least amount of effort from your side)

2. Fix your broken patterns by not calling `create_app` immediately and instead importing it and calling it _after_ you mocked `get_db`

The second approach is the preferred one and one of the reasons we don't provide dependency overrides: It encourages bad practices. As you've seen, the fact that Litestar does not provide you with such capabilities has brought to light a deeper issue in your code.

Examples

Here are two simplified examples that you can apply to your existing code:

Importing the app locally

dependencies.py

def some_dependency() -> str:
  return "foo"

main.py

from starlite import Starlite, Provide, get
from dependencies import some_dependency


@get()
def handler(dep: str) -> str:
  return dep


app = Starlite([handler], dependencies={"dep": Provide(some_dependency)})

test.py

from starlite import TestClient


def new_dependency():
    return "bar"


def test(mocker):
    mocker.patch("dependencies.some_dependency", new=new_dependency)
    from main import app

    with TestClient(app) as client:
        assert client.get("/").text == "bar"

Using an application factory

dependencies.py

def some_dependency() -> str:
  return "foo"

main.py

from starlite import Starlite, Provide, get
from dependencies import some_dependency


@get()
def handler(dep: str) -> str:
  return dep

def create_app() -> Starlite:
  return Starlite([handler], dependencies={"dep": Provide(some_dependency)})

test.py

from starlite import TestClient
from main import create_app
import pytest


def new_dependency():
    return "bar"

@pytest.fixture(autouse=True)
def mock_db():
    return mocker.patch("dependencies.some_dependency", new=new_dependency)


@pytest.fixture
def app():
  return create_app()


def test(app):
    with TestClient(app) as client:
        assert client.get("/").text == "bar"

Thank you very much sir. With your help, I've finally been able to resolve my issue.
Much appreciated. 🙂

[euri10]

the solution here explains quite well the reasonning, still the test fails for mainly 2 reasons:

the create_app function should import locally the dependency, I think it's not possible otherwise (not 100% sure but never seen it in action )

def create_app() -> Litestar:
    from deps.dependencies import some_dependency
    return Litestar([handler], dependencies={"dep": Provide(some_dependency)})

using a context manager to patch should be preferred.

see the diff here from a non-passing test to a passing one: euri10/mock_dependency_litestar@9434736

[euri10]

added an example on how to mock a httpx client dependency that was put in the state in the repo above ;)

[bachya]

The accepted answer has a challenge: you effectively lose the ability to test the actual dependency's internals (problematic if you care about test coverage).

I'm running into this with a mocked DB connection—MRE below:

dependencies.py

from litestar.datastructures import State
from litestar.di import Provide
from sqlalchemy.ext.asyncio import AsyncSession


async def get_db_session_dependency(state: State) -> AsyncSession:
    """Define a dependency injector to get a DB session.

    Args:
    ----
        state: The Litestar app state.

    Yields:
    ------
        The session.

    """
    sessionmaker = state.db_sessionmaker
    raised_exc = None

    async with sessionmaker() as session:
        try:
            yield session
        except Exception as exc:  # noqa: BLE001
            await session.rollback()
            raised_exc = exc
        finally:
            await session.close()

    # If we encountered an exception inside the yielded dependency, we raise here to
    # ensure it properly propagates up to our handlers:
    if raised_exc:
        raise raised_exc


DBSessionDependency = Provide(get_db_session_dependency)

app.py

"""Define the API."""

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from litestar import Litestar, Router
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from .dependencies import DBSessionDependency

DB_CONNECTION_STRING: Final[str] = "postgresql+psycopg://postgres:postgres@logs-db:5432/my-db"


@asynccontextmanager
async def lifespan_db_sessionmaker(app: Litestar) -> AsyncGenerator[None, None]:
    """Define a lifespan context manager for creating a DB sessionmaker.

    Args:
    ----
        app: The Litestar app.

    """
    engine = create_async_engine(DB_CONNECTION_STRING)
    app.state.db_sessionmaker = async_sessionmaker(autocommit=False, bind=engine)

    try:
        yield
    finally:
        await engine.dispose()


def create_app() -> Litestar:
    """Create the app.

    Returns
    -------
        The app.

    """
    # Create the API router:
    api_router = Router(
        path="/api",
        route_handlers=[
            Router(
                path="/v1",
                route_handlers=[HEALTH_ROUTER],
            )
        ],
    )

    # Create the app:
    app = Litestar(
        dependencies={
            "db_session": DBSessionDependency,
        },
        lifespan=[lifespan_db_sessionmaker],
    )

    return app

conftest.py

from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, MagicMock, Mock, patch

from litestar.testing import AsyncTestClient
import pytest
import pytest_asyncio

@pytest.fixture(name="patch_sqlalchemy")
def _patch_sqlalchemy_fixture(
    mock_sqlalchemy_sessionmaker: MagicMock,
) -> Generator[None, None, None]:
    """Define a patch for SQLAlchemy.

    Args:
    ----
        mock_sqlalchemy_sessionmaker: The mock SQLAlchemy session maker.

    """
    with patch(
        "src.app.create_async_engine", return_value=Mock(dispose=AsyncMock())
    ), patch("src.app.async_sessionmaker", return_value=mock_sqlalchemy_sessionmaker):
        yield


@pytest_asyncio.fixture(name="http_client")
async def http_client_fixture(
    patch_sqlalchemy: Generator[None, None, None],
) -> AsyncGenerator[AsyncTestClient, None]:
    """Define a test HTTP client.

    Args:
    ----
        patch_sqlalchemy: A patch for SQLAlchemy.

    Yields:
    ------
        The HTTP client.

    """
    app = create_app()
    async with AsyncTestClient(app=app) as client:
        yield client


@pytest.fixture(name="mock_sqlalchemy_sessionmaker")
def mock_sqlalchemy_sessionmaker_fixture(mock_sqlalchemy_session: Mock) -> MagicMock:
    """Define a mock SQLAlchemy session maker.

    Args:
    ----
        mock_sqlalchemy_session: The mock SQLAlchemy session.

    Returns:
    -------
        The mock SQLAlchemy session maker.

    """
    return MagicMock(
        return_value=MagicMock(
            __aenter__=AsyncMock(return_value=mock_sqlalchemy_session),
            __aexit__=AsyncMock(),
        )
    )


@pytest.fixture(name="mock_sqlalchemy_session")
def mock_sqlalchemy_session_fixture(
    mock_sqlalchemy_session_commit: AsyncMock,
    mock_sqlalchemy_session_execute: AsyncMock,
    mock_sqlalchemy_session_rollback: AsyncMock,
) -> Mock:
    """Define a mock SQLAlchemy session.

    Args:
    ----
        mock_sqlalchemy_session_commit: The mock SQLAlchemy session commit coroutine.
        mock_sqlalchemy_session_execute: The mock SQLAlchemy session execute coroutine.
        mock_sqlalchemy_session_rollback: The mock SQLAlchemy session rollback
            coroutine.

    Returns:
    -------
        The mock SQLAlchemy session.

    """
    return Mock(
        add=Mock(),
        close=AsyncMock(),
        delete=AsyncMock(),
        execute=mock_sqlalchemy_session_execute,
        commit=mock_sqlalchemy_session_commit,
        refresh=AsyncMock(),
        rollback=mock_sqlalchemy_session_rollback,
    )

In production, my dependency returns a sqlalchemy.ext.asyncio.AsyncSession, whereas in tests, the dependency returns a Mock that simulates a session. As expected, my tests fail because Litestar is getting something different (a Mock) than it expects (an AsyncSession):

[{'message': "Unsupported type: <class 'unittest.mock.Mock'>", 'key': 'db_session'}]

If I merely create a replacement dependency that returns a Mock, I'm bypassing all of the logic that makes the AsyncSession.

I know there are ways around this:

• Dependency(skip_validation=True) (although, that affects my route whether testing or in production)
• Refactoring my dependency so that it wraps SQLAlchemy stuff into its internals, such that mocking won't influence the final output

...but it would be great if skip_validation could be applied to the Litestar object. That way, when I create it in my tests (via my factory method), I can turn off validation at that point.

[AlexPetul]

Litestar checks the type via isinstance function (at least in current version 2.15.2), so you can use spec or spec_set parameters on your mock to pass this check. In this case you don't have to skip validation. For example:

mock = AsyncMock(spec=AsyncSession).