Compartir a través de

Pruebas unitarias de Durable Functions en Python

Las pruebas unitarias son una parte importante de las prácticas modernas de desarrollo de software. Las pruebas unitarias comprueban el comportamiento de la lógica empresarial y protegen frente a la introducción de cambios importantes desapercibidos en el futuro. Durable Functions puede crecer fácilmente en complejidad, por lo que la introducción de pruebas unitarias ayuda a evitar cambios importantes. En las secciones siguientes se explica cómo probar unitariamente los tres tipos de función: cliente de orquestación, orquestador y funciones de entidad.

Nota:

Esta guía solo se aplica a las aplicaciones de Durable Functions escritas en el modelo de programación de Python v2.

Prerrequisitos

Los ejemplos de este artículo requieren conocimientos de los siguientes conceptos y marcos:

Configuración del entorno de prueba

Para probar Durable Functions, es fundamental configurar un entorno de prueba adecuado. Esto incluye la creación de un directorio de prueba y la instalación del módulo de unittest Python en el entorno de Python. Para más información, consulte introducción a las pruebas unitarias de Python de Azure Functions.

Funciones de desencadenador de prueba unitaria

Las funciones desencadenantes, a menudo denominadas funciones cliente, inician las orquestaciones y eventos externos. Para probar estas funciones:

  • Simulación de la DurableOrchestrationClient ejecución de orquestación y administración de estado.
  • Asigne DurableOrchestrationClient métodos como start_new, get_statuso raise_event con funciones simuladas que devuelven valores esperados.
  • Invoque la función de cliente directamente con un cliente ficticio y otras entradas necesarias, como un req (objeto de solicitud HTTP) para las funciones de cliente de desencadenador HTTP.
  • Utiliza aserciones y herramientas unittest.mock para verificar el comportamiento esperado de inicio de orquestación, los parámetros y las respuestas HTTP.
import asyncio
import unittest
import azure.functions as func
from unittest.mock import AsyncMock, Mock, patch

from function_app import start_orchestrator

class TestFunction(unittest.TestCase):
  @patch('azure.durable_functions.DurableOrchestrationClient')
  def test_HttpStart(self, client):
    # Get the original method definition as seen in the function_app.py file
    func_call = http_start.build().get_user_function().client_function

    req = func.HttpRequest(method='GET',
                           body=b'{}',
                           url='/api/my_second_function',
                           route_params={"functionName": "my_orchestrator"})

    client.start_new = AsyncMock(return_value="instance_id")
    client.create_check_status_response = Mock(return_value="check_status_response")

    # Execute the function code
    result = asyncio.run(func_call(req, client))

    client.start_new.assert_called_once_with("my_orchestrator")
    client.create_check_status_response.assert_called_once_with(req, "instance_id")
    self.assertEqual(result, "check_status_response")

Funciones de orquestador de pruebas unitarias

Las funciones de orquestador gestionan la ejecución de varias funciones de actividad. Para probar un orquestador:

  • Simule DurableOrchestrationContext para controlar la ejecución de la función.
  • Reemplace DurableOrchestrationContext los métodos necesarios para la ejecución del orquestador como call_activity o create_timer por funciones simuladas. Estas funciones normalmente devolverán objetos de tipo TaskBase con una result propiedad .
  • Llame al orquestador de forma recursiva y pase el resultado de la tarea generada por la instrucción yield anterior a la siguiente.
  • Compruebe el resultado del orquestador mediante los resultados devueltos por el orquestador y unittest.mock.
import unittest
from unittest.mock import Mock, patch, call
from datetime import timedelta
from azure.durable_functions.testing import orchestrator_generator_wrapper

from function_app import my_orchestrator


class TestFunction(unittest.TestCase):
  @patch('azure.durable_functions.DurableOrchestrationContext')
  def test_chaining_orchestrator(self, context):
    # Get the original method definition as seen in the function_app.py file
    func_call = my_orchestrator.build().get_user_function().orchestrator_function

    # The mock_activity method is defined above with behavior specific to your app.
    # It returns a TaskBase object with the result expected from the activity call.
    context.call_activity = Mock(side_effect=mock_activity)

    # Create a generator using the method and mocked context
    user_orchestrator = func_call(context)

    # Use orchestrator_generator_wrapper to get the values from the generator.
    # Processes the orchestrator in a way that is equivalent to the Durable replay logic
    values = [val for val in orchestrator_generator_wrapper(user_orchestrator)]

    expected_activity_calls = [call('say_hello', 'Tokyo'),
                               call('say_hello', 'Seattle'),
                               call('say_hello', 'London')]
    
    self.assertEqual(context.call_activity.call_count, 3)
    self.assertEqual(context.call_activity.call_args_list, expected_activity_calls)
    self.assertEqual(values[3], ["Hello Tokyo!", "Hello Seattle!", "Hello London!"])

Funciones de entidad de pruebas unitarias

Las funciones de entidad gestionan objetos con estado mediante operaciones. Para probar una función de entidad:

  • Simulación de para DurableEntityContext simular el estado interno y las entradas de operación de la entidad.
  • Reemplaza DurableEntityContext métodos como get_state, set_state, y operation_name por simulaciones que devuelven valores controlados.
  • Invoque la función de entidad directamente con el contexto ficticio.
  • Use aserciones para comprobar los cambios de estado y los valores devueltos, junto con unittest.mock las utilidades.
import unittest
from unittest.mock import Mock, patch

from function_app import Counter

class TestEntityFunction(unittest.TestCase):
  @patch('azure.durable_functions.DurableEntityContext')
  def test_entity_add_operation(self, context_mock):
    # Get the original method definition as seen in function_app.py
    func_call = Counter.build().get_user_function().entity_function
    
    # Setup mock context behavior
    state = 0
    result = None

    def set_state(new_state):
        nonlocal state
        state = new_state

    def set_result(new_result):
        nonlocal result
        result = new_result

    context_mock.get_state = Mock(return_value=state)
    context_mock.set_state = Mock(side_effect=set_state)

    context_mock.operation_name = "add"
    context_mock.get_input = Mock(return_value=5)

    context_mock.set_result = Mock(side_effect=lambda x: set_result)

    # Call the entity function with the mocked context
    func_call(context_mock)

    # Verify the state was updated correctly
    context_mock.set_state.assert_called_once_with(5)
    self.assertEqual(state, 5)
    self.assertEqual(result, None)

Funciones de actividad de pruebas unitarias

Las funciones de actividad no requieren modificaciones específicas de Durable para ser probadas. Las instrucciones que se encuentran en la introducción a las pruebas unitarias de Python de Azure Functions son suficientes para probar estas funciones.

Contenido relacionado