Testing the RabbitMQProducer publish_message Method

In our previous blog post, we discussed how to publish messages to RabbitMQ using the RabbitMQProducer class. In this blog post, we will focus on testing the publish_message method of the RabbitMQProducer class to ensure that it functions as expected.

Setting Up the Test

To begin with, we need to set up the necessary test environment. We will be using the unittest.mock module to create mock objects for testing. Additionally, we will capture the output using the capsys fixture provided by pytest.

from pika.exceptions import ConnectionClosedByBroker
from producers.base import RabbitMQProducer
from unittest.mock import MagicMock

Testing the Successful Message Publishing

Let's start by testing the case where the message publishing is successful. We will create a mock connection object and a RabbitMQProducer instance with the mock connection. Then, we will mock the channel object and its exchange_declare and basic_publish methods. Finally, we will call the publish_message method with appropriate arguments.

def test_publish_message(capsys):
    connection = MagicMock()
    producer = RabbitMQProducer(connection)

    channel = MagicMock()
    channel.exchange_declare.return_value = None
    channel.basic_publish.return_value = None

    connection.get_channel.return_value = channel

    exchange = "test_exchange"
    routing_key = "test_routing_key"
    data = {"key": "value"}

    producer.publish_message(exchange=exchange, routing_key=routing_key, data=data)

    channel.basic_publish.assert_called_once()
    channel.exchange_declare.assert_called_once()

    captured = capsys.readouterr()
    assert captured.out.startswith("Message sent to exchange: ")

In the test function, we create a mock connection object and a RabbitMQProducer instance with the mock connection. Then, we create a mock channel object and specify the return values for its exchange_declare and basic_publish methods. Next, we assign the mock channel object to the connection's get_channel method. After that, we define the exchange, routing key, and data for the message. Finally, we call the publish_message method of the producer and assert that the basic_publish and exchange_declare methods were called once.

Testing the Failed Message Publishing

Now, let's test the case where the message publishing fails. We will use the side_effect property of the basic_publish method to raise a ConnectionClosedByBroker exception. This will simulate a scenario where the connection is closed by the broker.

def test_publish_message_failed(capsys):
    connection = MagicMock()
    producer = RabbitMQProducer(connection)

    channel = MagicMock()
    channel.exchange_declare.return_value = None
    channel.basic_publish.return_value = None

    def side_effect(exchange, routing_key, body, properties):
        raise ConnectionClosedByBroker(0, "sample")

    channel.basic_publish.side_effect = side_effect
    connection.get_channel.return_value = channel

    exchange = "test_exchange"
    routing_key = "test_routing_key"
    data = {"key": "value"}

    producer.publish_message(exchange=exchange, routing_key=routing_key, data=data)

    captured = capsys.readouterr()
    assert captured.out.startswith("Connection closed by broker. Failed to publish the message")

In this test function, we set up the same mock connection and producer objects. We also create a mock channel object and specify its return values. However, this time

, we define a side_effect function that raises a ConnectionClosedByBroker exception when called. We assign this side_effect function to the basic_publish method of the channel object. Finally, we call the publish_message method of the producer and assert that the expected error message is printed.

Running the Tests

To run the tests, we can use a test runner such as pytest. With pytest, simply execute the following command in your terminal:

pytest -v

The -v flag enables verbose mode, which provides more detailed output during test execution.

Conclusion

In this blog post, we have covered how to test the publish_message method of the RabbitMQProducer class. By testing both the successful and failed scenarios, we can ensure that our message publishing functionality works correctly and handles errors gracefully. Properly testing our code helps identify and address any issues or bugs, ensuring the reliability and stability of our RabbitMQ messaging system.