Eenheid testen van Durable Functions in Python

2025-06-19

Eenheidstests zijn een belangrijk onderdeel van moderne softwareontwikkelingsprocedures. Eenheidstests controleren het gedrag van bedrijfslogica en beschermen tegen het introduceren van onopgemerkte wijzigingen die in de toekomst fouten veroorzaken. Durable Functions kan eenvoudig in complexiteit toenemen, zodat het introduceren van eenheidstests helpt voorkomen dat wijzigingen fouten veroorzaken. In de volgende secties wordt uitgelegd hoe u de drie functietypen kunt testen: Orchestration-client, orchestrator en entiteitsfuncties.

Opmerking

Deze handleiding is alleen van toepassing op Durable Functions-apps die zijn geschreven in het Python v2-programmeermodel.

Vereiste voorwaarden

De voorbeelden in dit artikel vereisen kennis van de volgende concepten en frameworks:

Het testen van modules
Duurzame Functies
Python-eenheidstest
unittest.mock

De testomgeving instellen

Als u Durable Functions wilt testen, is het van cruciaal belang om een juiste testomgeving in te stellen. Dit omvat het maken van een testmap en het installeren van de module van unittest Python in uw Python-omgeving. Voor meer informatie, zie het overzicht van unittesten van Azure Functions in Python.

Functies voor testtriggers voor eenheden

Triggerfuncties, ook wel clientfuncties genoemd, initiëren indelingen en externe gebeurtenissen. Ga als volgt te werk om deze functies te testen:

Simuleer de DurableOrchestrationClient om de uitvoeringen en het statusbeheer van de orkestratie te testen.
Wijs DurableOrchestrationClient methoden toe, zoals start_new, get_statusof raise_event met mock-functies die verwachte waarden retourneren.
Roep de clientfunctie rechtstreeks aan met een gesimuleerde client en andere benodigde invoer, zoals een req (HTTP-aanvraagobject) voor HTTP-triggerclientfuncties.
Gebruik asserties en unittest.mock tools om het verwachte startgedrag van de orkestratie, parameters en HTTP-antwoorden te controleren.

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")

Orchestratorfuncties voor eenheidstests

Orchestrator-functies beheren de uitvoering van meerdere activiteitsfuncties. Een orchestrator testen:

Simuleer de DurableOrchestrationContext om de uitvoering van de functie te controleren.
Vervang DurableOrchestrationContext methoden die nodig zijn voor orchestratoruitvoering zoals call_activity of create_timer met mock-functies. Deze functies retourneren doorgaans objecten van het type TaskBase met een result eigenschap.
Roep de orchestrator recursief aan, waarbij het resultaat van de taak die door de vorige yield-instructie is gegenereerd, door aan de volgende wordt gegeven.
Controleer het orchestratorresultaat met behulp van de resultaten die zijn geretourneerd door de orchestrator en 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!"])

Eenheidstests voor entiteitsfuncties

Entiteitsfuncties beheren toestand-afhankelijke objecten door middel van operaties. Een entiteitsfunctie testen:

Maak een mock van de DurableEntityContext om de interne toestand en operatieve invoer van de entiteit te simuleren.
Vervang DurableEntityContext methoden zoals get_state, set_stateen operation_name door mocks die gecontroleerde waarden retourneren.
Roep de entiteitsfunctie rechtstreeks aan met de gesimuleerde context.
Gebruik asserties om statuswijzigingen en geretourneerde waarden te controleren, samen met unittest.mock hulpprogramma's.

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)

Activiteitsfuncties voor unittesting

Voor activiteitsfuncties hoeven geen duurzame, specifieke wijzigingen te worden getest. De richtlijnen in het overzicht van het testen van Python-eenheden in Azure Functions zijn voldoende voor het testen van deze functies.

Delen via