Testing RabbitMQConsumerBase class

Testing the RabbitMQConsumerBase consume Method

In our previous blog post, we discussed how to create a RabbitMQ consumer using the RabbitMQConsumerBase class. In this blog post, we will focus on testing the consume method of the RabbitMQConsumerBase class to ensure that it performs the necessary steps for consuming messages from RabbitMQ.

Setting Up the Test

To begin with, we need to set up the necessary test environment. We will be using the pytest framework for our tests and the unittest.mock module to create mock objects. Additionally, we will use the pytest-mock plugin to simplify the creation of mock objects.

import pytest
from unittest.mock import MagicMock, ANY
from consumers.base import RabbitMQConsumerBase
from config import Config

We will also define a sample subclass of RabbitMQConsumerBase called SampleRabbitMQConsumer for testing purposes.

class SampleRabbitMQConsumer(RabbitMQConsumerBase):
    def process_message(self, channel, method, properties, body):
        pass

Next, we will create a fixture called consumer_mock using the pytest.fixture decorator. This fixture will set up a SampleRabbitMQConsumer instance and a mock connection object. The mocker parameter is used to create the mock objects.

@pytest.fixture
def consumer_mock(mocker):
    consumer = SampleRabbitMQConsumer("name", "sample", "sample", 0, "sample", "sample")
    connection_mock = MagicMock()

    connection_context_mock = mocker.patch("connection.RabbitMQConnection.__enter__")
    connection_context_mock.return_value = connection_mock

    return consumer, connection_mock

Now, we are ready to write individual tests for the consume method.

Testing Exchange Declaration

The first step in the consume method is to declare the exchange. We can test this by asserting that the exchange_declare method is called on the mock channel object with the correct arguments.

def test_consume_exchange_declared(consumer_mock):
    config = Config()
    consumer, connection_mock = consumer_mock
    consumer.consume()

    assert connection_mock.get_channel.return_value.exchange_declare.call_count == 1
    connection_mock.get_channel.return_value.exchange_declare.assert_called_with(
        exchange_type="topic",
        exchange=config.EXCHANGE_NAME
    )

Testing Queue Declaration

The next step is to declare the queue. We can test this by asserting that the queue_declare method is called on the mock channel object with the correct arguments.

def test_consume_queue_declared(consumer_mock):
    consumer, connection_mock = consumer_mock
    consumer.consume()

    assert connection_mock.get_channel.return_value.queue_declare.call_count == 1
    connection_mock.get_channel.return_value.queue_declare.assert_called_with(
        queue="name"
    )

Testing Queue Binding

After declaring the queue, the next step is to bind it to the exchange. We can test this by asserting that the queue_bind method is called on the mock channel object with the correct arguments.

def test_consume_queue_bind(consumer_mock):
    config = Config()
    consumer, connection_mock = consumer_mock
    consumer.consume()

    assert connection_mock.get_channel.return_value.queue_bind.call_count == 1
    connection_mock.get_channel.return_value.queue_bind.assert_called_with(
        queue="name", exchange=config.EXCHANGE_NAME, routing_key="sample"
    )

Testing Basic Consume

Once the queue is bound to the exchange, we can

start consuming messages. We can test this by asserting that the basic_consume method is called on the mock channel object with the correct arguments.

def test_consume_basic_consume(consumer_mock):
    consumer, connection_mock = consumer_mock
    consumer.consume()

    assert connection_mock.get_channel.return_value.basic_consume.call_count == 1
    connection_mock.get_channel.return_value.basic_consume.assert_called_with(
        queue="name",
        auto_ack=False,
        on_message_callback=ANY,
    )

Testing Start Consuming

The consume method should start consuming messages by calling the start_consuming method on the mock channel object. We can test this by asserting that the start_consuming method is called.

def test_consume_uses_start_consuming(consumer_mock):
    consumer, connection_mock = consumer_mock
    consumer.consume()

    assert connection_mock.get_channel.return_value.start_consuming.call_count == 1

Testing Interrupted Consumption

We should also test the scenario where the consumption is interrupted, such as when a KeyboardInterrupt occurs. In this case, the stop_consuming method should be called on the mock channel object.

def test_consume_uses_start_consuming_interrupted(consumer_mock):
    consumer, connection_mock = consumer_mock
    connection_mock.get_channel.return_value.start_consuming.side_effect = KeyboardInterrupt
    consumer.consume()

    assert connection_mock.get_channel.return_value.stop_consuming.call_count == 1

Conclusion

In this blog post, we have covered the testing of the consume method of the RabbitMQConsumerBase class. By testing each step of the consumption process, we can ensure that messages are consumed correctly from RabbitMQ. Properly testing our consumer helps identify and address any issues or bugs, ensuring the reliability and stability of our RabbitMQ messaging system.