如何在 Python MagicMock 中模拟 Azure ServiceBus 接收器

Question

我正在使用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()

Answer 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