Автотесты приложений через AMQP

В статье разбираем протокол AMQP, его частную реализацию — RabbitMQ и пишем автотест с помощью PyTest для тестирования очередей сообщений.

Тестирование — одна из самых горячих тем в разработке программного обеспечения. Все согласны с необходимостью качественных проверок и определённого покрытия кода всевозможными тестами. Но как тестировать приложения, работающие не по привычному HTTP протоколу? В статье мы рассмотрим протокол AMQP, его частную реализацию RabbitMQ и протестируем наше простое приложение, разработав автотесты для него.

Андрей Мальчук

Бэкенд разработчик группы частных облаков КРОК.

Введение

AMQP (Advanced Message Queuing Protocol) — открытый протокол прикладного уровня для передачи сообщений между компонентами системы. Идея в том, что отдельные подсистемы или независимые приложения могут обмениваться произвольным образом сообщениями через AMQP-брокер, который осуществляет маршрутизацию, возможно гарантирует доставку, распределение потоков данных, подписку на нужные типы сообщений.

AMQP основан на трёх понятиях:

Сообщение (message) — единица передаваемых данных, основная его часть (содержание) никак не интерпретируется сервером, к сообщению могут быть присоединены структурированные заголовки.

Точка обмена (exchange) — в неё отправляются сообщения. Она распределяет сообщения в одну или несколько очередей. При этом в точке обмена сообщения не хранятся.

Очередь (queue) — здесь сообщения хранятся до тех пор, пока не будут забраны клиентом. Клиент всегда забирает сообщения из одной или нескольких очередей.

Producer — клиентское приложение, которое публикует сообщения в exchange.

Consumer — клиентское приложение, которое получает сообщения из очереди.

По сравнению с HTTP, у систем, построенных на очередях сообщений AMQP, есть ряд преимуществ и недостатков.

Плюсы HTTP

Отладка HTTP-запросов проще, чем в AMQP. Подключаться к очереди сообщений в AMQP придётся только через сторонние утилиты, тогда как HTTP можно отлаживать прямо в браузере.
Это популярный протокол — его используют практически всегда и везде, а значит и понимает его гораздо больше людей.

Плюсы AMQP

Надёжность доставки сообщений реализуется «из коробки» — значит не нужно об этом волноваться. Сообщение, которое будет отправлено в AMQP-брокер, будет доставлено и обработано одним из обработчиков очередей AMQP.
Broadcast — функционал, который позволяет уведомить разные компоненты системы в рамках одного сообщения. Таким образом, можно сократить количество отправляемых сообщений в единицу времени.

Коротко о RabbitMQ и RPC

RabbitMQ — это брокер сообщений с открытым исходным кодом. Он маршрутизирует сообщения по всем принципам протокола AMQP. Отправитель передаёт сообщение брокеру, а тот доставляет его получателю. RabbitMQ реализует и дополняет протокол AMQP.

RPC (Remote Procedure Call) — один из шаблонов взаимодействия в распределённых приложениях. Этот протокол позволяет программам вызывать функции и процедуры удалённо таким образом, как будто они представлены локально.

Совмещая RPC и RabbitMQ, мы в итоге получаем отказоустойчивую, распределённую систему для простого вызова функций и последующего агрегирования результатов.

Тестирование AMQP

Существуют разные способы тестирования приложений, основанных на протоколе AMQP. Вот несколько их них:

Функциональное тестирование;
Ручное тестирование;
Автоматизированное тестирование;
Интеграционное тестирование.

Под функциональным тестированием чаще всего подразумевается непосредственная проверка каждого элемента в тестируемой среде.

Ручное тестирование — проверка всех компонентов или отдельной, в частности, непосредственно человеком, вручную.

Но для автоматизации рутинных действий, проводимых человеком, существует автоматизированное тестирование — когда все действия, проводимые для тестирования системы человеком, описаны процедурно и могут исполняться автоматически.

При тестировании систем важно, чтобы система работала корректно полностью, целиком, в условиях продакшена. Для этого мало провести функциональное тестирование каждой компоненты, необходимо получить результат при тестировании всей системы. Поэтому существует интеграционное тестирование — когда программные модули объединяются и тестируются в группе.

Автотесты на Python

На мой взгляд, самый удобный способ тестировать API и MQ-сервисы — с помощью языка программирования Python, а также нескольких библиотек, о которых расскажу подробнее далее.

Основным инструментом при разработке автотестов будет pytest — библиотека с простым интерфейсом для написания тестов.

Для подключения к RabbitMQ есть множество библиотек, самая распространённая и хорошо документированная — pika.

Устанавливается всё с помощью утилиты pip:

			$ pip install pytest pika

Далее, представим, что у нас есть сервис, который работает по протоколу AMQP и непосредственно через RabbitMQ, в интерфейсе которого реализована простая функция, возвращающая последнее число из последовательности Фибоначчи:

			import pika

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

def on_request(ch, method, props, body):
    n = int(body)
    response = fib(n)

    ch.basic_publish(
        exchange="",
        routing_key=props.reply_to,
        properties=pika.BasicProperties(
            correlation_id=props.correlation_id
        ),
        body=str(response)
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host="localhost")
)
channel = connection.channel()
channel.queue_declare(queue="rpc_queue")

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="rpc_queue", on_message_callback=on_request)

channel.start_consuming()

Данный сервис подключается к RabbitMQ, создаёт очередь rpc_queue, из которой принимает сообщения, где тело запроса — это число N, от которого нужно вычислить последнее число из последовательности чисел Фибоначчи.

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

Напишем клиент, который взаимодействует с нашим сервисом с помощью очереди сообщений через RabbitMQ:

			import pika
import uuid

class FibonacciRpcClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host="localhost")
        )
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(queue="", exclusive=True)
        self.callback_queue = result.method.queue
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )
        self.response = None
        self.corr_id = None

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(
            exchange="",
            routing_key="rpc_queue",
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,
            ),
            body=str(n)
        )
        self.connection.process_data_events(time_limit=None)
        return int(self.response)

В данном примере:

Устанавливаем соединение к RabbitMQ и подключаемся к очереди rpc_queue;
Подписываемся на очередь, в которую возвращаются ответы от сервиса при вызове RPC-методов;
Реализуем функцию on_response, которая проверяет каждый ответ и сверяет correlation_id с тем, который мы отправили. Если идентификатор ответа совпадает — это ответ конкретного запроса, и функция сохраняет ответ в self.response.

Далее мы реализуем автотесты к нашему сервису. Создадим папку tests и в ней файлы conftest.py и test_fibonacci_rpc.py:

			.
└── tests
    ├── conftest.py
    └── test_fibonacci_rpc.py

Файл conftest.py содержит в себе всевозможные надстройки для pytest. В частности, добавим наш клиент туда для того, чтобы его можно было переиспользовать во всех подтестах, не импортируя вручную. Наш клиент будет иметь свойство Test Fixture — это объект, который можно рассматривать как набор условий, необходимых тесту для выполнения. Например, зачастую фикстуры создаются, чтобы генерировать какие-то данные ещё до теста и возвращать их для использования в тесте.

			import pytest

# Класс FibonacciRpcClient из примера выше:
class FibonacciRpcClient:
    ...

@pytest.fixture(scope="module")
def fibonacci_rpc():
    return FibonacciRpcClient()

Таким образом, наша фикстура fibonacci_rpc может переиспользоваться во всех тестах как аргумент к каждой тестовой функции.

Файлы с префиксом test_ — это файлы, в которых непосредственно реализованы сами тесты. Напишем несколько тестов, которые проверяют реализацию RPC-метода fib с нашими пользовательскими значениями.

Содержимое файла test_fibonacci_rpc.py:

			def test_fibonacci(fibonacci_rpc):
    # Последний элемент последовательности чисел Фибоначчи от N=0 это 0:
    assert fibonacci_rpc.call(0) == 0
    # От N=10 это 55:
    assert fibonacci_rpc.call(10) == 55
    # От N=25 это 75025:
    assert fibonacci_rpc.call(25) == 75025

def test_non_number(fibonacci_rpc):
    try:
        fibonacci_rpc.call("non-number")
    except Exception:
        pass
    else:
        assert False, "fib must consume only ints"

Функции с префиксом test_ означают, что это тестируемые методы, которые запускаются с помощью pytest.

Наш пример в test_fibonacci реализует проверку метода fib на стороне сервиса с положительными аргументами. Директивная assert проверяет условие, выполняющееся в её блоке на True или False. Если условие не выполнилось, тест упадёт с ошибкой, уведомив нас о некорректной работе сервиса.

Пример в test_non_number реализует проверку того же метода fib, только с заведомо некорректными входными параметрами. При вызове такого метода должна произойти ошибка. Если её не произошло, то мы закончим проверку метода с текстом fib must consume only ints.

Таким образом, мы полностью покрыли тестами функцию в нашем сервисе.

Заключение

В статье мы рассмотрели протокол AMQP и его частную реализацию RabbitMQ, реализовали тестовый сервис, реализующий простой удалённый вызов процедур с одним методом, а также протестировали работоспособность нашего сервиса с помощью автотестов, написанных на Python с помощью pytest.

Особенность тестирования сервисов, которые работают на протоколе AMQP, это корректная реализация объекта, который будет эмулировать запросы клиента. Для этого нужно знать некоторые особенности работы самого протокола AMQP, а также нюансы, которые использовали при разработке тестируемого сервиса (название очередей, публичные RPC-методы и прочее).