다음을 통해 공유

Python에서 지속성 함수 단위 테스트

단위 테스트는 최신 소프트웨어 개발 사례의 중요한 부분입니다. 단위 테스트는 비즈니스 논리 동작을 확인하고 향후 눈에 띄지 않는 호환성이 손상되는 변경을 도입하지 않도록 보호합니다. 지속성 함수는 복잡성이 쉽게 증가할 수 있으므로 단위 테스트를 도입하면 호환성이 손상되는 변경을 방지할 수 있습니다. 다음 섹션에서는 오케스트레이션 클라이언트, 오케스트레이터 및 엔터티 함수의 세 가지 함수 형식을 단위 테스트하는 방법을 설명합니다.

비고

이 가이드는 Python v2 프로그래밍 모델로 작성된 Durable Functions 앱에만 적용됩니다.

필수 조건

이 문서의 예제에는 다음 개념 및 프레임워크에 대한 지식이 필요합니다.

테스트 환경 설정

지속성 함수를 테스트하려면 적절한 테스트 환경을 설정하는 것이 중요합니다. 여기에는 테스트 디렉터리를 만들고 Python 환경에 Python 모 unittest 듈을 설치하는 것이 포함됩니다. 자세한 내용은 Azure Functions Python 단위 테스트 개요를 참조하세요.

트리거 함수 단위 테스트

클라이언트 함수라고도 하는 트리거 함수는 오케스트레이션 및 외부 이벤트를 시작합니다. 이러한 함수를 테스트하려면 다음을 수행합니다.

  • 오케스트레이션 실행 및 상태 관리를 시뮬레이션하기 위해 DurableOrchestrationClient를 모의합니다.
  • DurableOrchestrationClient, start_new 또는 get_status 같은 raise_event 메서드를 예상값을 반환하는 모의함수로 할당합니다.
  • 클라이언트 함수를 직접 호출할 때, 가상 클라이언트와 req(HTTP 요청 객체) 같은 기타 필요한 입력을 사용하여 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와 같은 모의 함수로 대체합니다. 이러한 함수는 일반적으로 속성이 있는 TaskBase 형식의 개체를 result 반환합니다.
  • 오케스트레이터를 재귀적으로 호출하여 이전 yield 문에서 생성된 작업의 결과를 다음으로 전달합니다.
  • 오케스트레이터와 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 단위 테스트 개요에 있는 지침은 이러한 함수를 테스트하기에 충분합니다.

관련 콘텐츠