Поделиться через

Модульное тестирование устойчивых функций в Python

Модульное тестирование является важной частью современных методик разработки программного обеспечения. Модульные тесты проверяют поведение бизнес-логики и защищают от внедрения незамеченных критических изменений в будущем. Устойчивые функции могут легко нарастать в сложности, поэтому внедрение модульных тестов помогает избежать нарушающих изменений. В следующих разделах объясняется, как выполнять модульное тестирование трех типов функций: клиента оркестрации, оркестратора и функций сущности.

Замечание

Это руководство относится только к приложениям устойчивых функций, написанным в модели программирования Python версии 2.

Предпосылки

В примерах этой статьи требуются знания о следующих понятиях и платформах:

Настройка тестовой среды

Чтобы протестировать устойчивые функции, важно настроить правильную тестовую среду. Это включает создание тестового каталога и установку модуля Python unittest в среду Python. Дополнительные сведения см. в обзоре модульного тестирования функций 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, макетами, возвращающими управляемые значения.
  • Вызовите функцию сущности непосредственно с помощью макетированного контекста.
  • Используйте утверждения (assertions) для проверки изменений состояния и возвращаемых значений совместно с 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 модификаций для тестирования. Рекомендации, приведенные в обзоре модульного тестирования функций Python, достаточно для тестирования этих функций.

Связанный контент