Чаще всего для ручного тестирования интеграций с 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 в докере (для UI-клиента пароль/логин по умолчанию "guest"):
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 можно добавить проверки данных в определенных полях тела сообщения или выводить в терминал только данные из отдельного поля тела сообщения, чтобы не просматривать целиком все сообщение.