Testare le unità delle Durable Functions in Python

Gli unit test sono una parte importante delle procedure di sviluppo software moderne. I test unitari verificano il comportamento della logica di business e proteggono dall'introduzione di cambiamenti importanti che potrebbero passare inosservati in futuro. La complessità di Funzioni durevoli può facilmente aumentare. Pertanto, l'introduzione di unit test consentirà di evitare le modifiche che causano un'interruzione. Le sezioni seguenti illustrano come eseguire unit test dei tre tipi di funzione: client di orchestrazione, agente di orchestrazione e funzioni di entità.

Prerequisiti

Gli esempi in questo articolo richiedono la conoscenza dei concetti e dei framework seguenti:

• Test unitario
• Funzioni durevoli
• Unittest python
• unittest.mock

Configurazione dell'ambiente di test

Per testare Durable Functions, è fondamentale configurare un ambiente di test appropriato. Ciò include la creazione di una directory di test e l'installazione del modulo di unittest Python nell'ambiente Python. Per ulteriori informazioni, vedere la panoramica dei test unitari per le Funzioni di Azure tramite Python.

Test delle unità delle funzioni di trigger

Le funzioni trigger, spesso definite funzioni client , avviano orchestrazioni ed eventi esterni. Per testare queste funzioni:

• Imitare il DurableOrchestrationClient per simulare l'esecuzione e la gestione dello stato dell'orchestrazione.
• Assegnare DurableOrchestrationClient metodi come start_new, get_statuso raise_event con funzioni fittizie che restituiscono valori previsti.
• Richiamare la funzione client direttamente con un client fittizio e gli altri input necessari, ad esempio un req (oggetto richiesta HTTP) per le funzioni client del trigger HTTP.
• Usare asserzioni e strumenti unittest.mock per verificare il comportamento di avvio previsto dell'orchestrazione, i parametri e le risposte 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")

Test delle unità delle funzioni dell'agente di orchestrazione

Le funzioni dell'agente di orchestrazione gestiscono l'esecuzione di più funzioni di attività. Per testare un orchestratore:

• Imitare DurableOrchestrationContext per controllare l'esecuzione della funzione.
• Sostituire i metodi DurableOrchestrationContext necessari per l'esecuzione dell'agente di orchestrazione ad esempio call_activity o create_timer, con funzioni fittizie. Queste funzioni restituiscono in genere oggetti di tipo TaskBase con una result proprietà .
• Chiamare l'agente di orchestrazione in modo ricorsivo, passando il risultato dell'attività generata dall'istruzione yield precedente a quella successiva.
• Verificare il risultato dell'agente di orchestrazione utilizzando i risultati restituiti da tale agente 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!"])

Test unitari delle funzioni dell'entità

Le funzioni di entità gestiscono oggetti con stato con operazioni. Per testare una funzione di entità:

• Imitare DurableEntityContext per simulare lo stato interno e gli input dell'operazione dell'entità.
• Sostituire DurableEntityContext metodi come get_state, set_statee operation_name con simulazioni che restituiscono valori controllati.
• Richiamare la funzione di entità direttamente con il contesto fittizio.
• Utilizzare le asserzioni per verificare le modifiche dello stato e i valori restituiti, insieme alle utilità 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)

Test delle unità delle funzioni dell'attività

Le funzioni dell'attività non richiedono modifiche specifiche della funzione durevole da testare. Le indicazioni disponibili nella panoramica dei test unitari Python di Azure Functions sono sufficienti per testare queste funzioni.

Contenuti correlati

• Informazioni su come migliorare le prestazioni della velocità effettiva delle app Python in Funzioni di Azure
• Leggere la Guida per sviluppatori Python per Funzioni di Azure
• Informazioni sulle procedure consigliate e sugli strumenti di diagnostica di Durable Functions