Partager via

Tests unitaires de fonctions durables dans Python

Les tests unitaires constituent une partie importante des pratiques de développement de logiciels modernes. Les tests unitaires vérifient le comportement de la logique métier et protègent contre l’introduction de changements cassants inaperçus à l’avenir. Les fonctions durables peuvent facilement devenir complexes, donc introduire des tests unitaires aide à éviter les changements perturbateurs. Les sections suivantes expliquent comment tester unitairement les trois types de fonctions : client d’orchestration, orchestrateur et fonctions d’entité.

Remarque

Ce guide s’applique uniquement aux applications Durable Functions écrites dans le modèle de programmation Python v2.

Conditions préalables

Les exemples de cet article nécessitent une connaissance des concepts et infrastructures suivants :

Configuration de l’environnement de test

Pour tester Durable Functions, il est essentiel de configurer un environnement de test approprié. Cela inclut la création d’un répertoire de test et l’installation du unittest module Python dans votre environnement Python. Pour plus d’informations, consultez la vue d’ensemble des tests unitaires Python d’Azure Functions.

Fonctions de déclencheur des tests unitaires

Les fonctions de déclencheur, souvent appelées fonctions clientes , initient des orchestrations et des événements externes. Pour tester ces fonctions :

  • Simulez le DurableOrchestrationClient pour simuler l’exécution de l’orchestration et la gestion de l’état.
  • Affectez des DurableOrchestrationClient méthodes telles que start_new, get_statusou raise_event avec des fonctions fictives qui retournent des valeurs attendues.
  • Appelez directement la fonction cliente avec un client fictif et d’autres entrées nécessaires telles qu’un req (objet de requête HTTP) pour les fonctions clientes du déclencheur HTTP.
  • Utilisez les assertions et les outils unittest.mock pour vérifier le comportement de démarrage d’orchestration prévu, ainsi que les paramètres et les réponses 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")

Fonctions d’orchestrateur des tests unitaires

Les fonctions Orchestrator gèrent l’exécution de plusieurs fonctions d’activité. Pour tester un orchestrateur :

  • Simuler le DurableOrchestrationContext pour contrôler l'exécution de la fonction.
  • Remplacez DurableOrchestrationContext les méthodes nécessaires à l’exécution de l’orchestrateur comme call_activity ou create_timer par des fonctions fictives. Ces fonctions retournent généralement des objets de type TaskBase avec une result propriété.
  • Appelez l’orchestrateur de manière récursive, en passant le résultat de la tâche générée par l’instruction de rendement précédente à la suivante.
  • Vérifiez le résultat de l’orchestrateur à l’aide des résultats retournés par l’orchestrateur et 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!"])

Fonctions d’entité de test unitaire

Les fonctions d’entité gèrent des objets avec état avec des opérations. Pour tester une fonction d’entité :

  • Simulez le DurableEntityContext pour simuler l’état interne et les entrées d’opération de l’entité.
  • Remplacez DurableEntityContext, get_state, set_state et operation_name par des mocks qui renvoient des valeurs contrôlées.
  • Appelez la fonction d’entité directement avec le contexte simulé.
  • Utilisez des assertions pour vérifier les modifications d’état et les valeurs retournées, ainsi que les unittest.mock utilitaires.
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)

Fonctions d’activité des tests unitaires

Les fonctions d’activité n’ont besoin d’aucune modification spécifique à Durable Functions pour les tester. Les conseils trouvés dans la vue d’ensemble des tests unitaires Python d’Azure Functions sont suffisants pour tester ces fonctions.

Contenu connexe