Testing with dishka

Testing you code does not always require the whole application to be started. You can have unit tests for separate components and even integration tests which check only specific links. In many cases you do not need IoC-container: you create objects with a power of Dependency Injection and not framework.

For other cases which require calling functions located on application boundaries you need a container. These cases include testing you view functions with mocks of business logic and testing the application as a whole. Comparing to a production mode you will still have same implementations for some classes and others will be replaced with mocks. Luckily, in dishka your container is not an implicit global thing and can be replaced easily.

There are many options to make providers with mock objects. If you are using pytest then you can

• use fixtures to configure mocks and then pass those objects to a provider

• create mocks in a provider and retrieve them in pytest fixtures from a container

The main limitation here is that a container itself cannot be adjusted after creation. You can configure providers whenever you want before you make a container. Once it is created dependency graph is build and validated, and all you can do is to provide context data when entering a scope.

Example

Imagine, you have a service built with FastAPI:

from sqlite3 import Connection
from typing import Annotated

from fastapi import FastAPI, APIRouter

from dishka.integrations.fastapi import FromDishka, inject

router = APIRouter()


@router.get("/")
@inject
async def index(connection: FromDishka[Connection]) -> str:
    connection.execute("select 1")
    return "Ok"


app = FastAPI()
app.include_router(router)

And a container:

from collections.abc import Iterable
from sqlite3 import connect, Connection

from dishka import Provider, Scope, provide, make_async_container
from dishka.integrations.fastapi import setup_dishka


class ConnectionProvider(Provider):
    def __init__(self, uri):
        super().__init__()
        self.uri = uri

    @provide(scope=Scope.REQUEST)
    def get_connection(self) -> Iterable[Connection]:
        conn = connect(self.uri)
        yield conn
        conn.close()


container = make_async_container(ConnectionProvider("sqlite:///"))
setup_dishka(container, app)

First of all - split your application factory and container setup.

from fastapi import FastAPI

from dishka import make_async_container
from dishka.integrations.fastapi import setup_dishka


def create_app() -> FastAPI:
    app = FastAPI()
    app.include_router(router)
    return app


def create_production_app():
    app = create_app()
    container = make_async_container(ConnectionProvider("sqlite:///"))
    setup_dishka(container, app)
    return app

Create a provider with you mock objects. You can still use production providers and override dependencies in a new one. Or you can build container only with new providers. It depends on the structure of your application and type of a test.

from sqlite3 import Connection
from unittest.mock import Mock

import pytest
import pytest_asyncio
from fastapi.testclient import TestClient

from dishka import Provider, Scope, provide, make_async_container
from dishka.integrations.fastapi import setup_dishka


class MockConnectionProvider(Provider):
    @provide(scope=Scope.APP)
    def get_connection(self) -> Connection:
        connection = Mock()
        connection.execute = Mock(return_value="1")
        return connection


@pytest.fixture
def container():
    container = make_async_container(MockConnectionProvider())
    yield container
    container.close()


@pytest.fixture
def client(container):
    app = create_app()
    setup_dishka(container, app)
    with TestClient(app) as client:
        yield client


@pytest_asyncio.fixture
async def connection(container):
    return await container.get(Connection)

Write tests.

from unittest.mock import Mock


async def test_controller(client: TestClient, connection: Mock):
    response = client.get("/")
    assert response.status_code == 200
    connection.execute.assertCalled()

Bringing all together

from collections.abc import Iterable
from sqlite3 import Connection, connect
from typing import Annotated
from unittest.mock import Mock

import pytest
import pytest_asyncio
from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient

from dishka import Provider, Scope, make_async_container, provide
from dishka.integrations.fastapi import FromDishka, inject, setup_dishka

router = APIRouter()


@router.get("/")
@inject
async def index(connection: FromDishka[Connection]) -> str:
    connection.execute("select 1")
    return "Ok"


def create_app() -> FastAPI:
    app = FastAPI()
    app.include_router(router)
    return app


class ConnectionProvider(Provider):
    def __init__(self, uri):
        super().__init__()
        self.uri = uri

    @provide(scope=Scope.REQUEST)
    def get_connection(self) -> Iterable[Connection]:
        conn = connect(self.uri)
        yield conn
        conn.close()


def create_production_app():
    app = create_app()
    container = make_async_container(ConnectionProvider("sqlite:///"))
    setup_dishka(container, app)
    return app


class MockConnectionProvider(Provider):
    @provide(scope=Scope.APP)
    def get_connection(self) -> Connection:
        connection = Mock()
        connection.execute = Mock(return_value="1")
        return connection


@pytest_asyncio.fixture
async def container():
    container = make_async_container(MockConnectionProvider())
    yield container
    await container.close()


@pytest.fixture
def client(container):
    app = create_app()
    setup_dishka(container, app)
    with TestClient(app) as client:
        yield client


@pytest_asyncio.fixture
async def connection(container):
    return await container.get(Connection)


@pytest.mark.asyncio
async def test_controller(client: TestClient, connection: Mock):
    response = client.get("/")
    assert response.status_code == 200
    connection.execute.assert_called()