Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-unit-testing-python below:

Unit Testing Durable Functions in Python

Unit testing is an important part of modern software development practices. Unit tests verify business logic behavior and protect from introducing unnoticed breaking changes in the future. Durable Functions can easily grow in complexity so introducing unit tests helps avoid breaking changes. The following sections explain how to unit test the three function types - Orchestration client, orchestrator, and entity functions.

The examples in this article require knowledge of the following concepts and frameworks:

• Unit testing
• Durable Functions
• Python unittest
• unittest.mock

To test Durable Functions, it's crucial to set up a proper test environment. This includes creating a test directory and installing Python's unittest module into your Python environment. For more info, see the Azure Functions Python unit testing overview.

Trigger functions, often referred to as client functions, initiate orchestrations and external events. To test these functions:

• Mock the DurableOrchestrationClient to simulate orchestration execution and status management.
• Assign DurableOrchestrationClient methods such as start_new, get_status, or raise_event with mock functions that return expected values.
• Invoke the client function directly with a mocked client and other necessary inputs such as a req (HTTP request object) for HTTP trigger client functions.
• Use assertions and unittest.mock tools to verify expected orchestration start behavior, parameters, and HTTP responses.

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

Orchestrator functions manage the execution of multiple activity functions. To test an orchestrator:

• Mock the DurableOrchestrationContext to control function execution.
• Replace DurableOrchestrationContext methods needed for orchestrator execution like call_activity or create_timer with mock functions. These functions will typically return objects of type TaskBase with a result property.
• Call the orchestrator recursively, passing the result of the Task generated by the previous yield statement to the next.
• Verify the orchestrator result using the results returned from the orchestrator and 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!"])

Entity functions manage stateful objects with operations. To test an entity function:

• Mock the DurableEntityContext to simulate the entity's internal state and operation inputs.
• Replace DurableEntityContext methods like get_state, set_state, and operation_name with mocks that return controlled values.
• Invoke the entity function directly with the mocked context.
• Use assertions to verify state changes and returned values, along with unittest.mock utilities.

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)

Activity functions require no Durable-specific modifications to be tested. The guidance found in the Azure Functions Python unit testing overview is sufficient for testing these functions.

• Learn how to improve throughput performance of Python apps in Azure Functions
• Read the Azure Functions Python Developer Guide
• Learn about Durable Functions best practices and diagnostic tools

RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4