Enhancement: Dependency Override

[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": "test@example.com",
        "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": "test@example.com",
        "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.