Как ускорить ручное тестирование интеграций с RabbitMQ с помощью библиотеки pika

Чаще всего для ручного тестирования интеграций с RabbitMQ используют UI-клиент. Но не всегда это удобно и позволяет тестировать быстро, особенно когда очередей к функционалу привязано несколько и в каждой нужно отдельно просматривать данные об отправленных сообщениях.

В этой статье я покажу как с помощью python-скрипта можно получать все нужные сообщения из одного/нескольких exchange в отдельную тестовую очередь и просматривать данные из нее в удобном формате в run-консоли.

Для получения сообщений RabbitMQ выбран подход "подписка как шпион", при котором сообщения дублируются в отдельную тестовую очередь (Queue 2):

Плюсы использования такого скрипта во время тестов:

• нет вмешательства в основную цепочку интеграции с RabbitMQ и не нужно выключать на время тестов консьюмеры

• можно тестировать без UI-клиента RabbitMQ

• есть возможность работать сразу с несколькими обменниками/очередями

• есть возможность выводить данные сообщений в каком угодно формате

• есть возможность расширять скрипт или сохранять под каждую фичу свой код скрипта с набором нужных параметров

Для скрипта была использована python-библиотека pika - это популярная библиотека, предназначенная для взаимодействия с брокерами сообщений по протоколу AMQP 0-9-1 (в первую очередь с RabbitMQ).

Далее будет приведено несколько примеров кода скрипта в зависимости от количества обменников или очередей, в которые отправляются сообщения для тестируемой фичи.

Один обменник и один очередь в интеграции

Если для тестируемой фичи сообщения отправляются в один обменник и очередь, то можно использовать код скрипта ниже:

#!/usr/bin/env python3

import textwrap
from functools import partial

import pika


"""
USER_NAME, USER_PASSWORD - логин и пароль пользователя, 
который имеет соответствующие права доступа к указанному виртуальному хосту
RABBITMQ_HOST - IP-адрес или доменное имя сервера (или докер-контейнера), 
где запущен RabbitMQ (например, localhost или 192.XXX.X.XX)
VIRT_HOST - виртуальный хост (vhost) - логическое пространство внутри брокера, 
изолирующее очереди и обменники
RABBITMQ_PORT - cетевой порт для подключения к брокеру 
(по умолчанию для AMQP используется протокольный порт 5672; если брокер развернут
на сервере в docker-контейнере, то это порт слева из строки проброса портов для 
контейнера "0.0.0.0:5672->5672/tcp")
EXСHANGE_NAME - наименование обменника
DEBUG_QUEUE_NAME - наименование тестовой очереди, в которую хотим получать сообщения
ROUTING_KEY - ключ маршрутизации, который указан для очереди, в которую 
идет отправка сообщений в тестируемой фиче
"""

# Параметры подключения к RabbitMQ
USER_NAME = "guest"
USER_PASSWORD = "guest"
RABBITMQ_HOST = "localhost"
RABBITMQ_PORT = 5672
VIRT_HOST = ""

# Параметры для работы с сообщениями из RabbitMQ
EXCHANGE_NAME = "base_exchange"
DEBUG_QUEUE_NAME = "debug_queue"
ROUTING_KEY = "queue.r_key"


def callback(ch, method, properties, body, queue_name):
    # Вывод строки из прочерков с наименованием тестовой очереди
    print("\n" + "-" * 65 + f"Сообщение из очереди \"{queue_name}\"" + "-" * 65)
    # Вывод строки с данными по exchange и routing_key
    print(f"exchange: {method.exchange}, routing_key: {method.routing_key}")
    # Вывод значения заголовка x-message-uid сообщения, если он был указан, или None
    headers = properties.headers or {}
    print(f"x-message-uid: {headers.get("x-message-uid")}")
    # Вывод данных из тела сообщения, по 150 символов на строку
    print(f"\nBody_message: \n{textwrap.fill(body.decode('utf-8'), width=150)}")

    ch.basic_ack(delivery_tag=method.delivery_tag)


def create_exchange(ch, exchange_name):
    ch.exchange_declare(
        exchange=exchange_name,
        exchange_type='topic',
        durable=True,
        passive=True
    )
    return exchange_name


def main():
    url = f'amqp://{USER_NAME}:{USER_PASSWORD}@{RABBITMQ_HOST}:{RABBITMQ_PORT}/{VIRT_HOST}'
    params = pika.URLParameters(url)
    with pika.BlockingConnection(params) as conn:
        with conn.channel() as ch:

            # Создание тестового обменника
            # EXCHANGE_NAME = create_exchange(ch, 'test_exchange')

            # Создание тестовой очереди
            ch.queue_declare(queue=DEBUG_QUEUE_NAME)

            # Привязка тестовой очереди к обменнику через ключ маршрутизации
            ch.queue_bind(exchange=EXCHANGE_NAME, queue=DEBUG_QUEUE_NAME, routing_key=ROUTING_KEY)

            # Регистрация обработчика (callback) сообщений из тестовой очереди
            ch.basic_consume(queue=DEBUG_QUEUE_NAME, on_message_callback=partial(callback, queue_name=DEBUG_QUEUE_NAME))

            try:
                print('Отслеживание сообщений...')
                ch.start_consuming()
            except KeyboardInterrupt:
                pass
            finally:
                # Удаление тестовых обменника/очереди при Ctrl+C, ошибке или успешном завершении
                print('\nУдаление тестовых обменника/очереди перед закрытием соединения...')
                if ch.is_open:
                    # ch.exchange_delete(exchange=EXCHANGE_NAME)
                    ch.queue_delete(queue=DEBUG_QUEUE_NAME)


if __name__ == "__main__":
    main()

Запущенный скрипт будет дублировать сообщения в тестовую очередь по указанным ключам маршрутизации, а затем выводить данные этих сообщений в run-консоли:

Несколько обменников и несколько очередей в интеграции

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

В параметрах указать данные этих обменников и ключей маршрутизации:

# Параметры для работы с сообщениями из RabbitMQ
EXCHANGE_NAME_1 = "base_exchange_1"
EXCHANGE_NAME_2 = "base_exchange_2"
ROUTING_KEY_1 = "queue.r_key_1"
ROUTING_KEY_2 = "queue.r_key_2"
ROUTING_KEY_3 = "*.r_key_3"
DEBUG_QUEUE_NAME = "debug_queue"

В функции main добавить/изменить код:

# Создание тестовых обменников (если нужно для теста самого скрипта)
# EXCHANGE_NAME_1 = create_exchange(ch, 'test_exchange_1')
# EXCHANGE_NAME_2 = create_exchange(ch, 'test_exchange_2')

...

# Привязка тестовой очереди к обменнику через ключ маршрутизации
ch.queue_bind(exchange=EXCHANGE_NAME_1, queue=DEBUG_QUEUE_NAME, routing_key=ROUTING_KEY_1)
ch.queue_bind(exchange=EXCHANGE_NAME_1, queue=DEBUG_QUEUE_NAME, routing_key=ROUTING_KEY_2)
ch.queue_bind(exchange=EXCHANGE_NAME_2, queue=DEBUG_QUEUE_NAME, routing_key=ROUTING_KEY_3)

...

# Удаление тестовых обменников (если были созданы для теста самого скрипта)
if ch.is_open:
  # ch.exchange_delete(exchange=EXCHANGE_NAME_1)
  # ch.exchange_delete(exchange=EXCHANGE_NAME_2)

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

Также можно расширить скрипт так, чтобы дублировать отдельные сообщения в разные тестовые очереди, но выводить данные сообщений также в одно окно run-консоли. Но в реальной практике пока это ни разу не понадобилось.

Проверка скрипта на локально запущенном контейнере c RabbitMQ

Создание и удаление тестового обменника закомментировано, так как при тестировании реальной фичи обменник уже будет создан и его наименование должно быть прописано в параметрах в начале скрипта.

Если на реальном проекте нет пока такой интеграции c RabbitMQ, а в учебных целях хочется посмотреть работу скрипта, то можно поднять контейнер с RabbitMQ в докере локально, раскомментировать в скрипте строки создания/удаления обменника/ов, запустить основной скрипт и с помощью дополнительного скрипта поотправлять сообщения в RabbitMQ.

Поднять контейнер с RabbitMQ в докере:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

Пример скрипта для отправки тестового соообщения в RabbitMQ (один обменник и один ключ маршрутизации, под несколько - нужно самостоятельно расширить код):

import random

import pika


# Параметры для отправки тестовых сообщений в RabbitMQ
EXCHANGE_NAME = "test_exchange"
ROUTING_KEY = "queue.r_key"

# Подключение к серверу RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = connection.channel()

# Отправка тестового сообщения в RabbitMQ
test_message = ('{"event_id": "req-12345","event_type": "user.registered","timestamp": "2026-10-11T23:01:00Z",'
                '"user_data": {"id": 1221,"email": "test@example.com"}}')
ch.basic_publish(
    exchange=EXCHANGE_NAME,
    routing_key=ROUTING_KEY,
    properties=pika.BasicProperties(
        headers={
            'x-message-uid': str(random.randint(1000, 2000))
        },
        delivery_mode=pika.DeliveryMode.Persistent
    ),
    body=test_message
)
connection.close()

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