Тестирование приложения 🧪#

Вы создали приложение FastStream с двумя независимыми сервисами (app.py и app2.py), которые обрабатывают сообщения с помощью Pydantic и строковые данные! 🎉 Теперь пора убедиться, что они работают правильно. В этом разделе мы научимся тестировать оба сервиса с помощью TestRabbitBroker, который эмулирует RabbitMQ в памяти, без необходимости запускать реальный брокер. Мы также проверим вызовы обработчиков, используя автоматическое мокирование. Это быстро, удобно и идеально для новичков! 😊 Готовы проверить ваш код? Погнали! 🚀

Зачем тестировать FastStream? 🤔#

Тестирование позволяет убедиться, что ваше приложение обрабатывает сообщения как ожидалось. С FastStream тестирование особенно удобно, потому что:

Эмуляция брокера 🔄: TestRabbitBroker имитирует RabbitMQ, не требуя реального подключения, что ускоряет тесты.
Проверка валидации ✅: Мы можем убедиться, что Pydantic правильно обрабатывает корректные и некорректные сообщения в app.py.
Мокирование 🤖: TestRabbitBroker автоматически мокирует обработчики, позволяя проверить, какие данные они получают.

Мы напишем тесты для сервисов из предыдущего раздела: app.py обрабатывает сообщения с полями username и message, передавая их из input-queue в output-queue, а app2.py обрабатывает строковые сообщения из final-queue.

Шаг 1: Установка pytest 📦#

Для тестирования будем использовать pytest и pytest-asyncio. Установите их в вашу виртуальную среду (если ещё не установлено):

pip install pytest pytest-asyncio

Проверьте установку:

pytest --version

Вы должны увидеть версию, например, pytest 8.x.x. Теперь мы готовы писать тесты! 💻

Шаг 2: Создание тестов ✍️#

Поскольку у нас два независимых сервиса, мы создадим два тестовых файла в папке faststream-tutorial:

test_app1.py для тестирования app.py.
test_app2.py для тестирования app2.py.

Добавьте код в test_app1.py:

test_app1.py
import os
from collections.abc import AsyncIterator

import pytest
import pytest_asyncio
from faststream.rabbit import RabbitBroker, TestRabbitBroker
from pydantic import ValidationError

from app import UserMessage, check_result, handle_message
from app import broker as broker_app1

WITH_REAL = os.getenv("TEST_BROKER_WITH_REAL", False) == "1"


@pytest_asyncio.fixture(scope="module")
async def client_broker1() -> AsyncIterator[RabbitBroker]:
    async with TestRabbitBroker(broker_app1, with_real=WITH_REAL) as br_client:
        yield br_client


@pytest.mark.asyncio
async def test_correct_message_app1(client_broker1: RabbitBroker) -> None:
    await client_broker1.publish(
        UserMessage(username="Alice", message="Hello"), queue="input-queue"
    )
    # Проверяем вызов handle_message (app.py)
    handle_message.mock.assert_called_once_with(
        {"username": "Alice", "message": "Hello"}
    )
    # Проверяем вызов check_result (app.py)
    check_result.mock.assert_called_once_with({"username": "Alice", "message": "HELLO"})


@pytest.mark.asyncio
async def test_invalid_message(client_broker1: RabbitBroker) -> None:
    with pytest.raises(ValidationError):
        await client_broker1.publish(
            {"username": "Alice", "wrong_field": "Hello"}, queue="input-queue"
        )

Добавьте код в test_app2.py:

test_app2.py
import os
from collections.abc import AsyncIterator

import pytest
import pytest_asyncio
from faststream.rabbit import RabbitBroker, TestRabbitBroker

from app2 import broker as broker_app2
from app2 import final_result

WITH_REAL = os.getenv("TEST_BROKER_WITH_REAL", False) == "1"


@pytest_asyncio.fixture(scope="module")
async def client_broker2() -> AsyncIterator[RabbitBroker]:
    async with TestRabbitBroker(broker_app2, with_real=WITH_REAL) as br_client:
        yield br_client


@pytest.mark.asyncio
async def test_correct_message_app2(client_broker2: RabbitBroker) -> None:
    await client_broker2.publish("Привет, FastStream!", queue="final-queue")
    # Проверяем вызов final_result (app2.py)
    final_result.mock.assert_called_once_with("Привет, FastStream!")

Что здесь происходит? 🔍

TestRabbitBroker 🔄: Контекстный менеджер, эмулирующий RabbitMQ в памяти, без подключения к реальному брокеру. async with TestRabbitBroker(broker) вызывает broker.connect() и broker.close(), что идеально для тестирования без реальных подписчиков. Параметр with_real позволяет переключаться на реальный брокер, если установлена переменная окружения TEST_BROKER_WITH_REAL=1.
Мокирование обработчиков 🤖: TestRabbitBroker автоматически мокирует обработчики (handle_message, check_result, final_result), позволяя проверить, какие данные они получают через .mock.assert_called_once_with().
test_correct_message_app1 ✅: Отправляет корректное сообщение UserMessage в input-queue, проверяет, что:
- handle_message вызван с входным сообщением в виде словаря ({"username": "Alice", "message": "Hello"}), так как FastStream сериализует UserMessage.
- check_result вызван с обработанным сообщением ({"username": "Alice", "message": "HELLO"}) из output-queue.
test_correct_message_app2 ✅: Отправляет строковое сообщение в final-queue, проверяет, что final_result вызван с "Привет, FastStream!".
test_invalid_message 🚫: Отправляет некорректное сообщение (с wrong_field) в input-queue и проверяет, что Pydantic выбросил ошибку валидации.
pytest.mark.asyncio ⚡: Позволяет использовать async/await в тестах.

Напоминание 📝: Контекстный менеджер TestRabbitBroker вызывает только broker.connect() и broker.close(), что идеально для тестирования. Для реального приложения с подписчиками используется broker.start() (как в faststream run app:app).

Шаг 3: Запуск тестов ▶️#

Убедитесь, что файлы app.py, app2.py, test_app1.py и test_app2.py находятся в папке faststream-tutorial. Запустите тесты:

pytest test_app1.py test_app2.py -v

Вы увидите вывод, например:

============================= test session starts ==============================
test_app1.py::test_correct_message_app1 PASSED                           [ 33%]
test_app1.py::test_invalid_message PASSED                               [ 66%]
test_app2.py::test_correct_message_app2 PASSED                          [100%]
=========================== 3 passed in 0.XXs ==============================

Все тесты прошли успешно! 🎉 Это подтверждает, что:

В app.py корректное сообщение обрабатывается, передается в output-queue с измененным message, и валидация Pydantic работает.
В app2.py строковое сообщение из final-queue обрабатывается корректно.

Шаг 4: Разбираемся с `TestRabbitBroker` и мокированием 🛠️#

TestRabbitBroker делает тестирование мощным и гибким:

Эмуляция очередей 🗄️: TestRabbitBroker обрабатывает сообщения в памяти, не требуя реального RabbitMQ.
Автоматическое мокирование 🔎: Подписчики автоматически мокируются, позволяя проверить вызовы обработчиков через .mock.assert_called_once_with().
Поддержка валидации 🛡️: Pydantic работает так же, как в реальном приложении, обеспечивая проверку данных в app.py.
Тестирование с реальным брокером 🔗: Переменная WITH_REAL (устанавливается через окружение TEST_BROKER_WITH_REAL=1) позволяет переключить TestRabbitBroker на использование реального брокера вместо эмуляции. Это полезно для интеграционного тестирования, когда нужно проверить взаимодействие с настоящим RabbitMQ, например, для подтверждения корректной работы с очередями или сетевыми настройками.
Тестирование независимых сервисов 🤝: Разделение тестов на test_app1.py и test_app2.py отражает независимость сервисов.

Этот подход, вдохновленный примерами FastStream, позволяет тестировать сложные сценарии с минимальным кодом.

Шаг 5: Практическое задание 📚#

Закрепите знания с помощью заданий:

В test_app1.py добавьте тест, который проверяет, что message в output-queue всегда в верхнем регистре (например, HELLO для входного Hello).
В test_app1.py создайте тест для проверки обработки сообщения с пустой строкой в поле message, если вы добавили min_length=1 в модель UserMessage в app.py.
(Дополнительно) В test_app2.py напишите тест, который проверяет, что final_result вызывается только один раз для одного сообщения в final-queue.

Что дальше? 🗺️#

Вы научились тестировать FastStream-приложения с использованием TestRabbitBroker и автоматического мокирования! 🎉 Это важный навык для создания надежных микросервисов. В следующем разделе мы узнаем, как генерировать красивую документацию в формате AsyncAPI, чтобы делиться спецификацией вашего приложения. Перейдите к Генерация документации, чтобы освоить этот процесс.

Если у вас есть идеи, вопросы или нужна помощь, загляните в официальную документацию FastStream, пишите в Telegram или Discord. Продолжайте тестировать и кодить! 🚀