Testes unitários de funções duráveis em Python

O teste de unidade é uma parte importante das práticas modernas de desenvolvimento de software. Testes de unidade verificam o comportamento de lógica de negócios e protegem contra a introdução de alterações com falhas despercebidas no futuro. As Funções Duráveis podem aumentar facilmente em complexidade, portanto, a introdução de testes de unidade ajuda a evitar alterações significativas. As seções a seguir explicam como testar os três tipos de função – cliente de orquestração, orquestrador, e funções de entidade.

Pré-requisitos

Os exemplos neste artigo exigem conhecimento dos seguintes conceitos e estruturas:

• Teste de unidade
• Funções duráveis
• Teste de unidade do Python
• unittest.mock

Configurando o ambiente de teste

Para testar o Durable Functions, é crucial configurar um ambiente de teste adequado. Isso inclui a criação de um diretório de teste e a instalação do módulo unittest do Python no seu ambiente Python. Para obter mais informações, consulte a visão geral do teste de unidade python do Azure Functions.

Testes de unidade das funções de gatilho

As funções de gatilho, geralmente chamadas de funções de cliente , iniciam orquestrações e eventos externos. Para testar essas funções:

• Simule o DurableOrchestrationClient para simular a execução e o gerenciamento de status da orquestração.
• Atribua DurableOrchestrationClient métodos como start_new, get_statusou raise_event com funções simuladas que retornam valores esperados.
• Invoque a função cliente diretamente com um cliente simulado e outras entradas necessárias, como um req (objeto de solicitação HTTP) para funções de cliente de gatilho HTTP.
• Use asserções e ferramentas unittest.mock para verificar o comportamento de início de orquestração esperado, parâmetros e respostas 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")

Testes de unidade das funções de orquestrador

As funções de orquestrador gerenciam a execução de várias funções de atividade. Para testar um orquestrador:

• Simule o DurableOrchestrationContext para controlar a execução da função.
• Substitua os DurableOrchestrationContext métodos necessários para a execução do orquestrador, como call_activity ou create_timer com funções simuladas. Essas funções normalmente retornarão objetos do tipo TaskBase com uma result propriedade.
• Chame o orquestrador recursivamente, passando o resultado da Tarefa gerada pela instrução de rendimento anterior para a próxima.
• Verifique o resultado do orquestrador usando os resultados retornados do orquestrador e 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!"])

Funções de entidade de teste de unidade

As funções de entidade gerenciam objetos com estado com operações. Para testar uma função de entidade:

• Simule o DurableEntityContext para simular o estado interno da entidade e os parâmetros de operação.
• Substitua DurableEntityContext métodos como get_state, set_statee operation_name por simulações que retornam valores controlados.
• Invoque a função de entidade diretamente com o contexto simulado.
• Use asserções para verificar alterações de estado e valores retornados, juntamente com utilitários unittest.mock.

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)

Testes de unidade das funções de atividade

As funções de atividade não exigem modificações específicas do Durable para serem testadas. As diretrizes encontradas na visão geral do teste de unidade python do Azure Functions são suficientes para testar essas funções.

Conteúdo relacionado

• Saiba como melhorar o desempenho da taxa de transferência de aplicativos Python no Azure Functions
• Ler o Guia do Desenvolvedor do Python do Azure Functions
• Saiba mais sobre as melhores práticas e ferramentas de diagnóstico do Durable Functions