共用方式為

針對 Python 中的 Durable Functions 進行單元測試

單元測試是新式軟體開發實務的重要組成部分。 單元測試會驗證業務邏輯的行為，並防止未來引入未被注意的破壞性變更。 Durable Functions 可以輕鬆地以複雜度成長，因此引進單元測試有助於避免重大變更。 下列各節說明如何單元測試三種函式類型 - Orchestration 用戶端、協調器和實體函式。

備註

本指南僅適用於以 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)

針對活動函式進行單元測試

活動函式不需要測試耐久性特定的修改。 Azure Functions Python 單元測試概觀中找到的指引足以測試這些函式。

相關內容

  • Last updated on