Эта статья — продолжение материала про универсальный прототип бэкенд-приложений. В ней мы поделимся практическим опытом написания тестов и рассмотрим, как выбранная архитектура упрощает этот процесс.
Все тесты в приложении условно разделили на три категории:
- Модульные тесты (unit tests)
- Интеграционные тесты (integration tests)
- Сервисные тесты (service tests) Для написания тестов мы будем использовать pytest и Faker. Это ключевые библиотеки для тестирования в Python, и мы настоятельно рекомендуем ознакомиться с ними, если вы еще не работали с ними ранее.
Введение в 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": " test@test.com",
"password": "password"
},
)
assert response.status_code == 201
assert "id" in
response.json()
В нашем приложении есть два внешних интерфейса для тестирования: HTTP и AMQP.
Для тестирования приложения нужно создать точку входа и обернуть её в фикстуру. Поскольку приложение состоит из множества изолированных компонентов, необходимо собрать их все вместе.
В идеале стоит создать фабрику для генерации точек входа в приложение — такую фабрику можно переиспользовать в фикстурах. В демонстрационном примере для простоты эта фабрика отсутствует.
Сервисные HTTP тесты
Большинство HTTP-фреймворков предоставляют встроенные инструменты для тестирования. Litestar — не исключение, в документации есть раздел про тестирование. Создание тестового окружения выглядит просто и понятно:
...
@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 интерфейсов.
Также присоединяйтесь к нашим комьюнити, где вы можете пообщаться с контрибьюторами технологий, упомянутых в статье, и задать интересующие вопросы:
Пишите тесты — это делает ваш код надёжнее!
