Consider this code under test:
import pika
class MQ_Client():
connection = None
channel = None
exchange_name="my_exchange"
def connect(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")
def publish(self, message):
properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)
if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):
log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)
else:
log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")
raise Exception("Failed to publish message to the queue.")
What I'd like to to is to patch Pika's basic_publish
method to raise an exception, and could use some help figuring out how to do this. This is from my last attempt:
@mock.patch.object(pika.BlockingConnection, 'channel')
def test_that_mq_publish_problems_return_error(self, mocked_channel):
with self.client as client:
mocked_channel.basic_publish.side_effect = Exception()
response = client.put("/api/users/bob", json="somedata")
self.assertEqual(response.status_code, 500)
Any advice on how to get this working will be appreciated!
I got it working. Mocking instance methods is quite different from mocking class methods. After reading this excellent article I finally understood how to do this, and it turns out all I needed to patch the basic_publish
method was to slightly modify my test like this:
@mock.patch('mq_client.pika.BlockingConnection', spec=pika.BlockingConnection)
def test_that_mq_publish_problems_return_error(self, mocked_connection):
with self.client as client:
mocked_connection.return_value.channel.return_value.basic_publish.return_value = False
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With