如何在 Python MagicMock 中模拟 Azure ServiceBus 接收器
Posted
技术标签:
【中文标题】如何在 Python MagicMock 中模拟 Azure ServiceBus 接收器【英文标题】:How to mock Azure ServiceBus Receiver in Python MagicMock 【发布时间】:2022-01-07 06:51:58 【问题描述】:我正在使用Python Azure ServiceBus package 并拥有以下 sn-p 代码:
# receiver.py
import logging
logger = logging.getLogger("test")
def receive_message(
connection_str,
queue_name
) -> None:
"""
Call Azure ServiceBus API to retrieve message from a queue
"""
with ServiceBusClient.from_connection_string(
connection_str, logging_enable=True
) as servicebus_client:
with servicebus_client.get_queue_receiver(
queue_name=queue_name,
max_wait_time=20,
receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
) as receiver:
for message in receiver:
logger.debug(f"Received message message")
我正在尝试为此函数编写单元测试,并希望能够模拟出recevier
。这是我编写单元测试的尝试,但失败了,因为我不知道如何让测试进入for message in receiver
块。
# test_receiver.py
@patch("receiver.ServiceBusClient")
@patch("receiver.logger")
def test_receive_message(mock_logger, mock_svcbus_client):
# Figure out how to mock
mock_svcbus_client.from_connection_string.return_value.get_queue_receiver.return_value = iter(["message"])
receive_message("mock_connection_str", "mock_q_name")
# Assertion fails
mock_logger.return_value.debug.assert_called_once()
【问题讨论】:
【参考方案1】:您可以尝试from mocks import MockReceivedMessage, MockReceiver
模拟receiver
示例 1:
class MockReceivedMessage(ServiceBusReceivedMessage):
def __init__(self, prevent_renew_lock=False, exception_on_renew_lock=False, **kwargs):
self._lock_duration = kwargs.get("lock_duration", 2)
self._raw_amqp_message = None
self._received_timestamp_utc = utc_now()
self.locked_until_utc = self._received_timestamp_utc + timedelta(seconds=self._lock_duration)
self._settled = False
self._receiver = MockReceiver()
self._prevent_renew_lock = prevent_renew_lock
self._exception_on_renew_lock = exception_on_renew_lock
示例 2:
def test_queue_message_receive_and_delete(self, servicebus_namespace_connection_string, servicebus_queue, **kwargs):
with ServiceBusClient.from_connection_string(
servicebus_namespace_connection_string, logging_enable=False) as sb_client:
with sb_client.get_queue_sender(servicebus_queue.name) as sender:
message = ServiceBusMessage("Receive and delete test")
sender.send_messages(message)
with sb_client.get_queue_receiver(servicebus_queue.name,
receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE) as receiver:
messages = receiver.receive_messages(max_wait_time=10)
assert len(messages) == 1
message = messages[0]
print_message(_logger, message)
with pytest.raises(ValueError):
receiver.complete_message(message)
with pytest.raises(ValueError):
receiver.abandon_message(message)
with pytest.raises(ValueError):
receiver.defer_message(message)
with pytest.raises(ValueError):
receiver.dead_letter_message(message)
with pytest.raises(ValueError):
receiver.renew_message_lock(message)
time.sleep(10)
with sb_client.get_queue_receiver(servicebus_queue.name) as receiver:
messages = receiver.receive_messages(max_wait_time=10)
for m in messages:
print_message(_logger, m)
assert len(messages) == 0
可以参考mocks.py 和test_queues.py
如果你还有疑问,你可以在 GitHub 上打开一个问题:azure-sdk-for-python
【讨论】:
以上是关于如何在 Python MagicMock 中模拟 Azure ServiceBus 接收器的主要内容,如果未能解决你的问题,请参考以下文章
Python 3:unittest.mock如何为特定输入指定不同的返回值?