通过

你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

在 Python 中对 Durable Functions 进行单元测试

单元测试是现代软件开发实践的重要组成部分。 单元测试验证业务逻辑行为，并防止将来引入未注意到的中断性变更。 Durable Functions 可以轻松增加复杂性，因此引入单元测试有助于避免重大更改。 以下部分介绍如何对三种函数类型（业务流程客户端、业务流程协调程序和实体函数）进行单元测试。

注释

本指南仅适用于以 Python v2 编程模型编写的 Durable Functions 应用。

先决条件

本文中的示例需要了解以下概念和框架：

设置测试环境

若要测试 Durable Functions，设置适当的测试环境至关重要。 这包括创建测试目录并将 Python 模块 unittest 安装到 Python 环境中。 有关详细信息，请参阅 Azure Functions Python 单元测试概述。

对触发器函数进行单元测试

触发器函数（通常称为 客户端 函数）启动业务流程和外部事件。 若要测试以下函数，请执行以下步骤：

  • 模拟 DurableOrchestrationClient 以仿真业务流程执行和状态管理。
  • 将DurableOrchestrationClient等方法与start_new、get_status或raise_event的模拟函数结合，使其返回预期值。
  • 通过模拟客户端和其他必要输入（如 HTTP 触发器客户端函数的 req（HTTP 请求对象））直接调用客户端函数。
  • 使用断言和 unittest.mock 工具验证预期的业务流程启动行为、参数和 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")

对业务流程协调程序函数进行单元测试

协调器函数管理多个活动函数的执行。 测试业务流程协调程序：

  • 模拟 DurableOrchestrationContext 以控制函数执行。
  • 将业务流程协调程序执行所需的 DurableOrchestrationContext 方法（如 call_activity 或 create_timer）替换为模拟函数。 这些函数通常返回具有 result 属性的 TaskBase 类型的对象。
  • 以递归方式调用业务流程协调程序，将上一个 yield 语句生成的 Task 结果传递给下一个。
  • 使用业务流程协调程序返回的结果和 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!"])

对实体函数进行单元测试

实体函数通过操作管理有状态对象。 测试一个实体函数：

  • 模拟 DurableEntityContext 以仿真实体的内部状态和操作输入。
  • 将 DurableEntityContext、get_state 和 set_state 等方法替换为返回受控值的 operation_name 模拟。
  • 借助模拟上下文直接调用实体函数。
  • 使用断言来验证状态更改、返回值和 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)

对活动函数进行单元测试

活动函数无需进行任何特定于 Durable 的修改即可测试。 Azure Functions Python 单元测试概述中找到的指南足以测试这些函数。

相关内容