在pytest中模拟一个连接class
Mock a connection class in pytest
我有一个继承自 kombu.ConsumerProducerMixin
的 class,我想在没有实际 rabbitmq 服务的情况下进行测试 运行ning.
class Aggregator(ConsumerProducerMixin):
def __init__(self, broker_url):
exchange_name = 'chargers'
self.status = 0
self.connection = Connection(broker_url)
...
在我的测试文件中,我做了以下操作:
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")
使用调试器进入 Aggregator.__init__
,我看到 connection
仍未修补为 Mock
实例:
(Pdb) self.connection
<Connection: amqp://guest:**@localhost:5672// at 0x7fc8b7f636d8>
(Pdb) Connection
<class 'kombu.connection.Connection'>
我的问题是如何正确修补连接,这样我就不需要 rabbitmq 来 运行 测试?
好的,docs声明如下:
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.
因此,解决方案:
@patch('aggregator.aggregator.Connection')
def test_on_request(mock_connect):
agg = Aggregator('localhost')
我有一个继承自 kombu.ConsumerProducerMixin
的 class,我想在没有实际 rabbitmq 服务的情况下进行测试 运行ning.
class Aggregator(ConsumerProducerMixin):
def __init__(self, broker_url):
exchange_name = 'chargers'
self.status = 0
self.connection = Connection(broker_url)
...
在我的测试文件中,我做了以下操作:
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")
使用调试器进入 Aggregator.__init__
,我看到 connection
仍未修补为 Mock
实例:
(Pdb) self.connection
<Connection: amqp://guest:**@localhost:5672// at 0x7fc8b7f636d8>
(Pdb) Connection
<class 'kombu.connection.Connection'>
我的问题是如何正确修补连接,这样我就不需要 rabbitmq 来 运行 测试?
好的,docs声明如下:
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.
因此,解决方案:
@patch('aggregator.aggregator.Connection')
def test_on_request(mock_connect):
agg = Aggregator('localhost')