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