es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Cómo simular un receptor de Azure ServiceBus en Python usando MagicMock

14 Jun, 2023 Programación

Estoy utilizando el paquete Python Azure ServiceBus y tengo el siguiente fragmento de código:


<h1>receiver.py</h1>

import logging
logger = logging.getLogger("test")

def receive_message(
        connection_str,
        queue_name
) -> None:
    """
    Llama a la API de Azure ServiceBus para obtener el mensaje de una cola.
    """
    with ServiceBusClient.from_connection_string(
            connection_str, logging_enable=True
    ) as servicebus_client:

<pre><code>    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}")
</code></pre>

Intento escribir pruebas unitarias para esta función y quiero poder hacer una simulación de “mock” del receiver. Este es mi intento de escribir la prueba unitaria, que falla porque no puedo hacer que la prueba entre en el bloque for message in receiver.


<h1>test_receiver.py</h1>

@patch("receiver.ServiceBusClient")
@patch("receiver.logger")
def test_receive_message(mock_logger, mock_svc_bus_client):

<pre><code># Intento hacer una simulación de "mock":
mock_svc_bus_client.from_connection_string.return_value.get_queue_receiver.return_value = iter(["message"])


receive_message("mock_connection_str", "mock_q_name")

# La aserción falla
mock_logger.return_value.debug.assert_called_once()
</code></pre>
Tags:  , , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    Para crear un objeto de prueba (‘mock’) que se comporte como el objeto receiver en la función receive_message, puede utilizar la función unittest.mock.MagicMock(). A continuación se muestra una versión actualizada del código de prueba que usa el objeto de prueba:

    # test_receiver.py

from unittest.mock import patch, MagicMock

@patch("receiver.ServiceBusClient")
@patch("receiver.logger")
def test_receive_message(mock_logger, mock_svcbus_client):

    queue_receiver_mock = MagicMock()
    queue_receiver_mock.__iter__.return_value = ["mensaje"]

    mock_svcbus_client.from_connection_string.return_value.get_queue_receiver.return_value = queue_receiver_mock

    receive_message("cadena_de_conexión_simulada", "nombre_de_cola_simulado")

    mock_logger.return_value.debug.assert_called_once_with("Received message mensaje")

    En este código actualizado, creamos un objeto MagicMock llamado queue_receiver_mock que se comporta como el objeto receiver. A continuación, establecemos el método __iter__ del objeto queue_receiver_mock para que devuelva una lista que contiene un mensaje ("mensaje"). Finalmente, establecemos que el objeto queue_receiver_mock simulado (“mock”) sea devuelto por el método get_queue_receiver del objeto ServiceBusClient.

    Con este objeto simulado en su lugar, el bucle for message in receiver dentro de la función receive_message iterará sobre la lista que contiene un mensaje, y la llamada logger.debug dentro del bucle se llamará una vez con el mensaje esperado. La prueba pasará si se cumple esta afirmación.

Comments are closed.