I have a class which inherits from kombu.ConsumerProducerMixin
which I would like to test without an actual rabbitmq service running.
class Aggregator(ConsumerProducerMixin):
def __init__(self, broker_url):
exchange_name = 'chargers'
self.status = 0
self.connection = Connection(broker_url)
...
In my test file I did the following:
from unittest.mock import Mock, patch
from aggregator import Aggregator
@patch('kombu.connection.Connection')
def test_on_request(conn_mock):
agg = Aggregator('localhost')
m = Message("", {"action": "start"}, content_type="application/json")
Stepping into the Aggregator.__init__
with the debugger, I see that connection
is still not patched to be a Mock
instance:
(Pdb) self.connection
<Connection: amqp://guest:**@localhost:5672// at 0x7fc8b7f636d8>
(Pdb) Connection
<class 'kombu.connection.Connection'>
My question is how do I properly patch connection such that I don't need rabbitmq to run the tests?
OK, The docs state the following:
patch() works by (temporarily) changing the object that a name points to with another one. There can be many names pointing to any individual object, so for patching to work you must ensure that you patch the name used by the system under test.
The basic principle is that you patch where an object is looked up, which is not necessarily the same place as where it is defined. A couple of examples will help to clarify this.
Hence, the solution:
@patch('aggregator.aggregator.Connection')
def test_on_request(mock_connect):
agg = Aggregator('localhost')
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