Testing a Python Backend: Unit, Integration, and Service Tests

How do you test a backend that accepts both HTTP requests and AMQP messages? I tested the same application through both interfaces, using a real test database and an in-memory message broker. This article compares unit, integration, and service tests and explains where mocks help and where they make tests fragile.

This is an English translation of my article originally published in Russian on Habr: Practical Testing of Python Applications.

This is the second part of the series. Part 1 explains the backend architecture. Here, I share how I tested that application and how its architecture made the process easier.

I have divided the application's tests into three categories:

• Unit tests
• Integration tests
• Service tests

We will use pytest and Faker to write the tests. These are key libraries for testing in Python, and I strongly recommend getting familiar with them if you have not used them before.

Jump to a section:

• Introduction to Mock Objects
• Writing Unit Tests
• Writing Integration Tests
• Writing Service Tests
• Conclusion

Introduction to Mock Objects

Mock objects can seem complicated to developers who are just starting out. Many articles mix up mock objects and monkeypatching, but when we use Dependency Injection (DI), we do not need monkeypatching in our tests.

Moreover, if your tests require monkeypatching, that is a bad sign. Your code probably uses global variables. You can read about their drawbacks here.

What Is a Mock Object?

A mock object is a special object that simulates the behavior of a real object during testing.

Imagine we are writing a function that retrieves a weather forecast in kelvins and converts the temperature to degrees Celsius:

from typing import Any

def get_celsius_temp(weather_service: Any) -> str:
    kelvin_temp: float = weather_service.get_temp()
    return kelvin_temp - 273.15

Without mock objects, testing this function would require:

• A working weather service. If it is unavailable, the tests will fail.
• A city with a constant temperature of +15°C. No such city exists.
• A city with a constant temperature of −15°C. That does not exist either.

Using a mock object solves these problems and lets us test the function as it is:

from unittest.mock import Mock

def test_positive_temp() -> None:
    weather_svc = Mock()
    weather_svc.get_temp = Mock(return_value=288.15)
    assert get_celsius_temp(weather_svc) == 15

def test_negative_temp() -> None:
    weather_svc = Mock()
    weather_svc.get_temp = Mock(return_value=258.15)
    assert get_celsius_temp(weather_svc) == -15

Types of Mock Objects in Python

The unittest library provides several useful classes:

• Mock — the base class
• MagicMock — an extension of Mock with magic methods implemented
• AsyncMock — an extension of MagicMock with async/await support

What Mock Objects Can Do

Mock objects can:

• Record call history and arguments
• Count calls
• Raise exceptions through side_effect

from unittest.mock import Mock

m = Mock(return_value=42)

result1 = m(foo="bar")
result2 = m("test", 1, 3)

print(m.mock_calls)  # [call(foo='bar'), call('test', 1, 3)]
print(m.call_count)  # 2
print(m.called)  # True
print(result1, result2)  # 42, 42

z = Mock(side_effect=Exception("Mocked exception"))

z()  # Exception: Mocked exception

Checking Calls

Mock objects provide convenient methods for checking how they were used:

from unittest.mock import Mock

x = Mock()
x(foo="bar")

x.assert_called()  # Checks that the object was called
x.assert_called_with(foo="bar")  # Checks the call arguments
x.assert_called_once()  # Checks that it was called exactly once

The create_autospec Function

The create_autospec function creates a mock object that checks method signatures:

from unittest.mock import create_autospec

class Foo:
    def bar(self, some_id: int): ...

mocked_foo = create_autospec(Foo)

mocked_foo.bar(some_id=1)  # OK
mocked_foo.baar()  # AttributeError: Mock object has no attribute 'baar'
mocked_foo.bar()   # TypeError: missing a required argument: 'some_id'

Writing Unit Tests

Unit tests check individual parts of a program, such as functions, methods, and classes, in isolation from the rest of the code. They are usually the most numerous and fragile tests in a project. In our example, we will write unit tests for the business logic implemented in the interactors.

Here is an example of the interactor that retrieves a book, along with its test:

# book_club/application/interactors.py
...

class GetBookInteractor:
    def __init__(self, book_gateway: interfaces.BookReader) -> None:
        self._book_gateway = book_gateway

    async def __call__(self, uuid: str) -> entities.BookDM | None:
        return await self._book_gateway.read_by_uuid(uuid)


# tests/test_application.py
...
@pytest.fixture
def get_book_interactor() -> GetBookInteractor:
    book_gateway = create_autospec(interfaces.BookReader)
    return GetBookInteractor(book_gateway)


@pytest.mark.parametrize("uuid", [str(uuid4()), str(uuid4())])
async def test_get_book(get_book_interactor: GetBookInteractor, uuid: str) -> None:
    result = await get_book_interactor(uuid=uuid)
    get_book_interactor._book_gateway.read_by_uuid.assert_awaited_once_with(
        uuid=uuid
    )
    assert result == get_book_interactor._book_gateway.read_by_uuid.return_value

In the fixture, I create an interactor, but replace the real BookGateway, which accesses the database, with a mock object that conforms to the BookReader interface. This lets us check that the interactor works correctly.

Now let us look at the interactor that saves a book:

# book_club/application/interactors.py
...

class NewBookInteractor:
    def __init__(
        self,
        db_session: interfaces.DBSession,
        book_gateway: interfaces.BookSaver,
        uuid_generator: interfaces.UUIDGenerator,
    ) -> None:
        self._db_session = db_session
        self._book_gateway = book_gateway
        self._uuid_generator = uuid_generator

    async def __call__(self, dto: NewBookDTO) -> str:
        uuid = str(self._uuid_generator())
        book = entities.BookDM(
            uuid=uuid, title=dto.title, pages=dto.pages, is_read=dto.is_read
        )

        await self._book_gateway.save(book)
        await self._db_session.commit()
        return uuid


# tests/test_application.py
...

@pytest.fixture
def new_book_interactor(faker: Faker) -> NewBookInteractor:
    db_session = create_autospec(interfaces.DBSession)
    book_gateway = create_autospec(interfaces.BookSaver)
    uuid_generator = MagicMock(return_value=faker.uuid4())
    return NewBookInteractor(db_session, book_gateway, uuid_generator)


async def test_new_book_interactor(new_book_interactor: NewBookInteractor, faker: Faker) -> None:
    dto = NewBookDTO(
        title=faker.pystr(),
        pages=faker.pyint(),
        is_read=faker.pybool(),
    )
    result = await new_book_interactor(dto=dto)
    uuid = str(new_book_interactor._uuid_generator())
    new_book_interactor._book_gateway.save.assert_awaited_with(
        entities.BookDM(
            uuid=uuid,
            title=dto.title,
            pages=dto.pages,
            is_read=dto.is_read,
        )
    )
    new_book_interactor._db_session.commit.assert_awaited_once()
    assert result == uuid

Because I used dependency injection (DI) when writing the code, I can easily test the business logic without monkeypatching. All dependencies are replaced with mock objects, which fully isolates the tests.

TDD and Clean Architecture

Clean Architecture, combined with extensive use of DI, makes TDD (Test-Driven Development) much easier. When dependencies are clearly defined and injected from outside, writing tests before the implementation becomes a natural process. You can first describe the expected behavior through tests, create mock objects for all dependencies, and then implement the logic without waiting for external services or infrastructure to be ready. This approach leads to a more carefully considered API design and provides quick feedback on code quality.

Solitary vs. Sociable: Choosing Test Boundaries and the Cost of Mock Objects

When writing unit tests, it is important to define their boundaries:

• Solitary tests, which mock all neighboring components, are useful for checking logic in isolation, but make changes to the codebase painful.
• Sociable tests, which mock only external contracts, check how components interact through their public API. This gives you more freedom to change the internal implementation.

In our case, testing the interactors is a solitary approach: we isolate the business logic by mocking all dependencies, including database gateways and UUID generators. This lets us quickly check the interactor's logic, but comes with risks. Each mock fixes a specific interaction interface between components. During refactoring, for example, if we rename book_gateway.read_by_uuid(), all the tests will break even though the application itself still works. We become tied to internal APIs that should not be rigidly fixed.

For a complex domain model, a more resilient solution is the sociable approach, where several domain objects are tested together without mocks. These tests make it easy to check both the happy path and edge cases while fixing only the public API. This makes tests less fragile and gives you more freedom to change the codebase later. However, if the domain is trivial, service tests can cover it completely.

You can learn more about the types of unit tests and their use in a separate article.

In real projects with complex business logic, you should prefer sociable unit tests for testing the domain. The business logic in this educational example does not allow us to implement such tests. However, here is an example of what they could look like:

async def test_book_creation_with_validation(
    # Real dependencies, not mock objects
    session,
    book_gateway,
    uuid_generator,
    faker
):
    db_session = session
    interactor = NewBookInteractor(
        db_session=db_session,
        book_gateway=book_gateway,
        uuid_generator=uuid_generator
    )

    book_data = NewBookDTO(
        title=faker.pystr(min_chars=1, max_chars=200),
        pages=faker.pyint(min_value=1, max_value=1000),
        is_read=faker.pybool()
    )

    book_id = await interactor(book_data)

    created_book = await book_gateway.read_by_uuid(book_id)

    assert created_book is not None
    assert created_book.title == book_data.title
    assert created_book.pages == book_data.pages
    assert created_book.is_read == book_data.is_read
    assert created_book.uuid == book_id

async def test_book_creation_with_invalid_data(
    # Real dependencies, not mock objects
    session,
    book_gateway,
    uuid_generator
):
    interactor = NewBookInteractor(
        db_session=session,
        book_gateway=book_gateway,
        uuid_generator=uuid_generator
    )

    invalid_book_data = NewBookDTO(
        title="",
        pages=-5,
        is_read=False
    )

    with pytest.raises(ValueError) as exc_info:
        await interactor(invalid_book_data)

    assert "Book title cannot be empty" in str(exc_info.value)

Writing Integration Tests

Integration tests check how an application interacts with external systems. Our application has two external integrations:

• RabbitMQ
• Postgres

FastStream handles all communication with the queue, so we will not write tests for that: there is no point in testing code we do not write. If you have your own abstractions for working with the queue, you will, of course, need to cover them with integration tests as well.

The Postgres integration is encapsulated in BookGateway, so we will test it through this class's public API:

# tests/conftest.py
...

@pytest.fixture(scope="session")
async def session_maker(...) -> async_sessionmaker[AsyncSession]:
    engine = create_async_engine(...)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    return async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        autoflush=False,
        expire_on_commit=False
    )


@pytest.fixture
async def session(
    session_maker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[AsyncSession, Any]:
    async with session_maker() as session:
        session.commit = AsyncMock()
        yield session
        await session.rollback()


# tests/test_infrastructure.py
...

@pytest.fixture
async def book_gateway(session: AsyncSession) -> BookGateway:
    return BookGateway(session=session)


async def test_create_book(
    session: AsyncSession, book_gateway: BookGateway, faker: Faker
) -> None:
    uuid = faker.uuid4()
    title = faker.pystr()
    pages = faker.pyint()
    is_read = faker.pybool()
    await session.execute(
        insert(Book).values(
            uuid=uuid,
            title=title,
            pages=pages,
            is_read=is_read
        )
    )
    result = await book_gateway.read_by_uuid(uuid)
    assert result.title == title
    assert result.pages == pages
    assert result.is_read is is_read


async def test_save_book(
    session: AsyncSession,
    book_gateway: BookGateway,
    faker: Faker
) -> None:
    book_dm = BookDM(
        uuid=faker.uuid4(),
        title=faker.pystr(),
        pages=faker.pyint(),
        is_read=faker.pybool(),
    )
    await book_gateway.save(book_dm)

    result = await session.execute(
        select(Book).where(Book.uuid == book_dm.uuid)
    )
    rows = result.fetchall()
    assert len(rows) == 1
    book = rows[0][0]
    assert book.title == book_dm.title
    assert book.pages == book_dm.pages
    assert book.is_read == book_dm.is_read

The tests need a separate database. The simplest option is to create it in advance alongside your local development database. In a CI environment, you can start the test database as part of pipeline jobs. Here are examples for GitHub Actions and GitLab CI.

Writing Service Tests

Service tests sit between end-to-end tests and integration tests. They check the application as it runs in production, while allowing some external dependencies to be replaced.

These tests check end-to-end scenarios through external interfaces, involving all application components. For example:

async def test_create_user(client):
   response = await client.post(
       "/users",
       json={
           "email": "[e-mail]",
           "password": "password"
        },
   )
   assert response.status_code == 201
   assert "id" in response.json()

Our application has two external interfaces to test: HTTP and AMQP.

To test the application, we need to create an entry point and wrap it in a fixture. Since the application consists of many isolated components, we need to assemble them all.

Ideally, you should create a factory for generating application entry points and reuse it in fixtures. For simplicity, the demonstration example does not include such a factory.

HTTP Service Tests

Most HTTP frameworks provide built-in testing tools. FastAPI is no exception: its documentation has a section on testing. Setting up the test environment is straightforward:

from collections.abc import AsyncIterator

import pytest
from dishka import AsyncContainer
from faker import Faker
from fastapi import FastAPI
from dishka.integrations.fastapi import setup_dishka
from httpx import AsyncClient, ASGITransport
from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncSession

from book_club.infrastructure.models import Book
from book_club.controllers.http import book_router


@pytest.fixture
async def http_app(container: AsyncContainer) -> FastAPI:
    app = FastAPI()
    app.include_router(router=book_router)
    setup_dishka(container, app)
    return app


@pytest.fixture
async def http_client(http_app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=http_app),
        base_url="http://",
    ) as client:
        yield client

Since we have only one endpoint, we will use just one test scenario. We will check the successful retrieval of a book that we first save to the database:

...
async def test_get_book(
    session: AsyncSession,
    http_client: AsyncClient,
    faker: Faker,
) -> None:
    uuid = faker.uuid4()
    title = faker.pystr(min_chars=3, max_chars=120)
    pages = faker.pyint()
    is_read = faker.pybool()

    await session.execute(
        insert(Book).values(
            uuid=uuid,
            title=title,
            pages=pages,
            is_read=is_read
        ),
    )

    result = await http_client.get(f"/book/{uuid}")
    assert result.status_code == 200
    assert result.json()["title"] == title
    assert result.json()["pages"] == pages
    assert result.json()["is_read"] == is_read
...

AMQP Service Tests

FastStream also provides convenient tools for testing message handlers. You do not need to start a real queue: you can use an in-memory broker.

...
@pytest.fixture
async def broker() -> RabbitBroker:
    broker = RabbitBroker()
    broker.include_router(AMQPBookController)
    return broker


@pytest.fixture
async def amqp_app(
    broker: RabbitBroker,
    container: AsyncContainer,
) -> FastStream:
    app = FastStream(broker)
    faststream_integration.setup_dishka(container, app, auto_inject=True)
    return app


@pytest.fixture
async def amqp_client(amqp_app: FastStream) -> AsyncIterator[RabbitBroker]:
    async with TestRabbitBroker(amqp_app.broker) as br:
        yield br

...

The test itself is simple: I send a message in the required format to the in-memory queue and check that the handler has done its job by verifying that the book data appeared in the database:

@pytest.mark.asyncio
async def test_save_book(
    amqp_client: RabbitBroker,
    session: AsyncSession,
    faker: Faker,
) -> None:
    title = faker.name_nonbinary()
    pages = faker.pyint()
    is_read = faker.pybool()
    await amqp_client.publish(
        {
            "title": title,
            "pages": pages,
            "is_read": is_read
        },
        queue="create_book",
    )

    result = await session.execute(
        select(Book).where(
            Book.title == title,
            Book.pages == pages,
            Book.is_read == is_read
        )
    )
    rows = result.fetchall()
    assert len(rows) == 1
    book = rows[0][0]
    assert book.title == title
    assert book.pages == pages
    assert book.is_read == is_read
...

Conclusion

Throughout this article, we have looked at different approaches to testing, from isolated unit tests to end-to-end service tests. The main point I want to convey is this: in a real project, the best approach is to combine sociable tests for a complex domain with service and integration tests for everything else.

I encourage you to explore the prototype's source code yourself. There, you will find working versions of the examples discussed, including service tests for the HTTP and AMQP interfaces.

Also join our communities, where you can talk to contributors to the technologies mentioned in this article and ask questions:

• FastAPI / Litestar (ASGI Community)
• FastStream Community
• Reagento Community

Write tests: they make your code more reliable!

Where do you draw the boundary between unit and service tests? Which dependencies do you replace with mocks, and which do you keep real?

Comments

[DEV SUPPORTS]

You need to verify your account.

tr.ee/dev-to

[Sehat1137]

meh