Как выполнить простое модульное тестирование с RabbitMQ в python? - PullRequest
Новая
       86

Как выполнить простое модульное тестирование с RabbitMQ в python?

0 голосов
/ 25 апреля 2020

В моем модульном тесте я хочу просто начать потреблять, публиковать sh сообщение, получать ответ и утверждать, является ли ответ тем, чем я его ожидаю. Тем не менее, я пытался делать это часами и не нашел решения.

Проблема в том, что я не могу определить метод в классе, который останавливает потребление. Я попытался определить метод, подобный этому: 

def stop(self):
    self.channel.basic_cancel()

def stop(self):
    self.channel.stop_consuming()

def stop(self):
    self.connection.close()

Но, похоже, ничего не работает. Я читал, что это потому, что после выполнения start_consuming() единственный способ остановить потребление - отменить его после отправки сообщения. Но если я сделаю это, то я бы изменил исходный on_request, и это не было бы полезно для моего приложения, потому что соединение закроется после первого сообщения. Я нашел pytest-rabbitmq , но документация мне не ясна, и поэтому я не знаю, смогу ли я использовать этот плагин для достижения того, чего я хочу.

Кстати, что чем разница между basic_cancel, stop_consuming и close?

1 Ответ

0 голосов
/ 28 апреля 2020

Я не понимаю вашего сценария !. Насколько я понимаю, вы можете создать соединение и канал тем же способом, чтобы вы могли публиковать sh, потребителя, утверждать и останавливать потребление при необходимости

Надеюсь, это поможет! 

def test_rabbitmq():
    from pika import BlockingConnection, ConnectionParameters, PlainCredentials

    conn = BlockingConnection(ConnectionParameters(host='host', virtual_host='vhost', credentials=PlainCredentials('username', 'password')))
    channel = conn.channel()

    # define your consumer
    def on_message(channel, method_frame, header_frame, body):
        message = body.decode()
        # assert your message here
        # asset message == 'value'
        channel.basic_cancel('test-consumer')  # stops the consumer

    # define your publisher
    def publish_message(message):
        channel.basic_publish(exchange='', routing_key='', body=message')

    publish('your message')
    tag = channel.basic_consume(queue='queue', on_message_callback=on_message, consumer_tag='test-consumer')

stop_consuming - Отменяет всех потребителей, сигнализируя start_consuming l oop о выходе.

basic_cancel - Этот метод отменяет потребителя. В качестве входных данных будет принят потребительский тег.

close - закрывает соединение / канал

Ссылка

...