Python での Durable Functions の単体テスト

単体テストは、最新のソフトウェア開発プラクティスの重要な部分です。 単体テストでは、ビジネス ロジックの動作を検証し、将来目に見えない破壊的変更が発生しないように保護します。 Durable Functions は複雑さが増しやすいため、単体テストを導入すると、破壊的変更を回避できます。 次のセクションでは、オーケストレーション クライアント、オーケストレーター、エンティティ関数の 3 つの関数の種類を単体テストする方法について説明します。

[前提条件]

この記事の例では、次の概念とフレームワークに関する知識が必要です。

• 単体テスト
• デュラブル ファンクションズ (Durable Functions)
• Python unittest
• unittest.mock

テスト環境の設定

Durable Functions をテストするには、適切なテスト環境を設定することが重要です。 これには、テスト ディレクトリの作成と Python 環境への Python の unittest モジュールのインストールが含まれます。 詳細については、 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 単体テストの概要に記載されているガイダンスで十分です。

関連コンテンツ

• Azure Functions で Python アプリのスループット パフォーマンスを向上させる方法について説明します
• Azure Functions Python 開発者ガイドを読む
• Durable Functions のベスト プラクティスと診断ツールについて説明します