Практическое тестирование приложений на python

Всем привет! Эта статья — продолжение материала про универсальный прототип бэкенд-приложений. В ней я поделюсь практическим опытом написания тестов и покажу, как выбранная архитектура упрощает этот процесс.

Все тесты в приложении я условно разделил на три категории:

• Модульные тесты (unit tests)

• Интеграционные тесты (integration tests)

• Сервисные тесты (service tests) Для написания тестов мы будем использовать pytest и Faker. Это ключевые библиотеки для тестирования в Python, и я настоятельно рекомендую ознакомиться с ними, если вы еще не работали с ними ранее.

Для быстрого перехода к разделам

• Введение в mock-объекты

• Пишем юнит-тесты

• Пишем интеграционные тесты

• Пишем сервисные тесты

• Заключение

Введение в mock-объекты

Для начинающих разработчиков тема mock-объектов может показаться сложной. Во многих статьях смешивают концепции mock-объектов и monkeypatch, но при использовании Dependency Injection (DI) нам не нужен monkeypatch в тестах.

Более того, если в ваших тестах требуется monkeypatch — это плохой знак. Скорее всего, в коде используются глобальные переменные, о недостатках которых можно почитать здесь.

Что такое mock-объект?

Mock-объект — это специальный объект, который имитирует поведение реального объекта во время тестирования.

Представьте, что мы пишем функцию, которая получает прогноз погоды по Кельвину и преобразует его в градусы по Цельсию:

from typing import Any

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

Без mock-объектов для тестирования этой функции потребовались бы:

• Работающий сервис погоды (если он недоступен — тесты не пройдут)

• Город с постоянной темпера��урой +15°C (такого не существует)

• Город с постоянной температурой -15°C (такого тоже не существует)

Если же мы воспользуемся mock-объектом, то мы решим эти проблемы и сможем протестировать эту функцию как есть:

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

Типы mock-объектов в Python

В библиотеке unittest есть несколько полезных классов:

• Mock — базовый класс

• MagicMock — расширение Mock с реализованными магическими методами

• AsyncMock — расширение MagicMock с поддержкой async/await

Возможности mock-объектов

Mock-объекты могут:

• Записывать историю вызовов и параметры

• Считать количество вызовов

• Генерировать исключения через 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

Проверка вызовов

Mock-объекты предоставляют удобные методы для проверки сценариев использования:

from unittest.mock import Mock

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

x.assert_called()  # Проверяет, что объект вызывали
x.assert_called_with(foo="bar")  # Проверяет параметры вызова
x.assert_called_once()  # Проверяет, что вызвали ровно один раз

Функция create_autospec

Функция create_autospec создаёт mock-объект, который проверяет сигнатуру методов:

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'

Пишем юнит-тесты

Юнит-тесты проверяют работу отдельных частей программы (функций, методов, классов) в изоляции от остального кода. Обычно это самые многочисленные и хрупкие тесты в проекте. В нашем примере мы напишем юнит-тесты для бизнес-логики, описанной в интеракторах.

Вот пример интерактора для получения книги и его теста:

# 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

В фикстуре я создаю интерактор, но вместо реального BookGateway (который обращается к базе данных) подставляю mock-объект, соответствующий интерфейсу BookReader. Это позволяет проверить, что интерактор работает корректно.

Теперь рассмотрим интерактор для сохранения книги:

# 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

Благодаря тому, что при написании кода я использовал принцип внедрения зависимостей (DI), я могу легко тестировать бизнес-логику без применения monkeypatch. Все зависимости заменяются на mock-объекты, что обеспечивает полную изоляцию тестов.

TDD и чистая архитектура

Использование чистой архитектуры с активным применением DI существенно упрощает практику TDD (Test-Driven Development). Когда зависимости чётко определены и внедряются извне, писать тесты перед реализацией становится естественным процессом. Вы можете сначала описать ожидаемое поведение через тесты, создавать mock-объекты для всех зависимостей, а затем реализовывать саму логику, не дожидаясь готовности внешних сервисов или инфраструктуры. Такой подход приводит к более продуманному дизайну API и позволяет быстро получать обратную связь о качестве кода.

Solitary vs Sociable: выбор границ тестирования и цена mock-объектов

При написании юнит-тестов важно определить их границы:

• Solitary-тесты (с моками всех соседей) полезны для изолированной проверки логики, но делают любое изменение в кодовой базе болезненным.

• Sociable-тесты (где мокаются только внешние контракты) проверяют взаимодействие компонентов через их публичное API, что даёт больше свободы для изменений внутренней реализации.

В нашем случае тестирование интеракторов — это solitary-подход: мы изолируем бизнес-логику, мокая все зависимости (шлюзы к БД, генераторы UUID). Такой подход позволяет быстро проверить логику интерактора, но несёт риски. Каждый mock фиксирует конкретный интерфейс взаимодействия между компонентами. При рефакторинге — например, если мы решим переименовать метод book_gateway.read_by_uuid() — все тесты сломаются, хотя само приложение будет работать. Мы становимся заложниками внутренних API, которые не должны быть жёстко зафиксированы.

Для сложной доменной модели более устойчивым решением являются sociable-подход, где несколько доменных объектов тестируются вместе без моков. Они позволяют легко проверять не только happy path, но и граничные случаи, при этом фиксируя только публичное API. Это делает тесты менее хрупкими и даёт больше свободы при дальнейшем изменении кодовой базы. Однако если домен тривиален, его тестирование можно полностью покрыть сервисными тестами.

В реальных проектах, для сложной бизнес-логики необходимо отдавать предпочтение sociable юнит-тестам для тестирования домена, в данном учебном примере из-за особенностей бизнес-логики такие тесты не реализовать. Но вот пример того, как бы они выглядели:

async def test_book_creation_with_validation(
    # Реальные зависимости, не mock-объекты
	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(
    # Реальные зависимости, не mock-объекты
	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 "Название книги не может быть пустым" in str(exc_info.value)

Пишем интеграционные тесты

Интеграционные тесты проверяют взаимодействие приложения с внешними системами. Наше приложение имеет две внешние интеграции:

• RabbitMQ

• Postgres

Всю работу с очередью на себя берёт FastStream, поэтому тесты мы на это писать не будем, т.к. нет никакого смысла тестировать код, который мы не пишем. Если у вас есть какие-то собственные абстракции для работы с очередью, то в таком случае, конечно, потребуется их тоже покрывать интеграционными тестами.

Интеграция с Postgres инкапсулирована в классе BookGateway, поэтому мы будем тестировать её через публичный 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

Для тестов потребуется отдельная база данных. Проще всего создать её заранее рядом с базой для локальной разработки. В CI-окружении тестовую базу можно разворачивать в рамках pipeline jobs. Примеры для GitHub Actions и GitLab CI.

Пишем сервисные тесты

Сервисные тесты занимают промежуточное положение между сквозными и интеграционными тестами. Они проверяют работу приложения в боевом режиме, но позволяют подменять некоторые внешние зависимости.

Эти тесты проверяют сквозные сценарии через внешние интерфейсы с участием всех компонентов приложения. Пример:

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()
    

В нашем приложении есть два внешних интерфейса для тестирования: HTTP и AMQP.

Для тестирования приложения нужно создать точку входа и обернуть её в фикстуру. Поскольку приложение состоит из множества изолированных компонентов, необходимо собрать их все вместе.

В идеале стоит создать фабрику для генерации точек входа в приложение — такую фабрику можно переиспользовать в фикстурах. В демонстрационном примере для простоты эта фабрика отсутствует.

Сервисные HTTP тесты

Большинство HTTP-фреймворков предоставляют встроенные инструменты для тестирования. FastAPI — не исключение, в документации есть раздел про тестирование. Создание тестового окружения выглядит просто и понятно:

...
@pytest.fixture
async def http_app(container: AsyncContainer) -> Litestar:
    app = Litestar(
        route_handlers=[HTTPBookController],
    )
    litestar_integration.setup_dishka(container, app)
    return app


@pytest.fixture
async def http_client(
	http_app: Litestar,
) -> AsyncIterator[AsyncTestClient]:
    async with AsyncTestClient(app=http_app) as client:
        yield client
...

��оскольку у нас всего один эндпоинт, ограничимся одним тестовым сценарием. Мы проверим положительный сценарий получения книги, которую предварительно сохраним в базу данных:

...
async def test_get_book(
    session: AsyncSession,
    http_client: AsyncTestClient,
    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 тесты

FastStream также предоставляет удобные инструменты для тестирования обработчиков сообщений. Для этого не требуется поднимать реальную очередь — можно использовать in-memory брокер:

...
@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

...

Сам тест выглядит просто: я отправляю сообщение в нужном формате в in-memory очередь и проверяю, что обработчик выполнил свою работу — данные о книге появились в базе:

@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
...

Заключение

На протяжении статьи мы рассмотрели разные подходы к тестированию — от изолированных юнит-тестов до сквозных сервисных тестов. Главный вывод, который я хочу донести: в реальном проекте оптимально использовать комбинацию sociable-тестов для сложного домена и сервисных и интеграционных тестов для всего остального.

Призываю заглянуть в исходный код прототипа и изучить его самостоятельно — там вы найдёте все рассмотренные примеры в рабочем состоянии, включая сервисные тесты для HTTP и AMQP интерфейсов.

Также присоединяйтесь к нашим комьюнити, где вы можете пообщаться с контрибьюторами технологий, упомянутых в статье, и задать интересующие вопросы:

• FastAPI / Litestar (ASGI Community)

• FastStream Community

• Reagento Community

Пишите тесты — это делает ваш код надёжнее!