اختبار الوحدة ل Durable Functions في Python

يعد اختبار الوحدة جزءا مهما من ممارسات تطوير البرامج الحديثة. تتحقق اختبارات الوحدة من سلوك منطق العمل وتحمي من إدخال تغييرات كسرية غير ملاحظتها في المستقبل. يمكن أن تنمو Durable Functions بسهولة في التعقيد، لذا فإن إدخال اختبارات الوحدة يساعد على تجنب كسر التغييرات. تشرح الأقسام التالية كيفية اختبار الوحدة أنواع الوظائف الثلاثة - عميل التزامن والمنسق ووظائف الكيان.

‏‫ملاحظة

ينطبق هذا الدليل فقط على تطبيقات Durable Functions المكتوبة في نموذج برمجة Python v2.

المتطلبات الأساسية

تتطلب الأمثلة الواردة في هذه المقالة معرفة المفاهيم والأطر التالية:

• اختبار الوحدة
• وظائف دائمة
• Python unittest
• unittest.mock

إعداد بيئة الاختبار

لاختبار Durable Functions، من الضروري إعداد بيئة اختبار مناسبة. يتضمن ذلك إنشاء دليل اختبار وتثبيت وحدة Python unittest في بيئة Python الخاصة بك. لمزيد من المعلومات، راجع نظرة عامة على اختبار وحدة 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 .
• استدعاء المنسق بشكل متكرر، وتمرير نتيجة المهمة التي تم إنشاؤها بواسطة عبارة العائد السابقة إلى التالي.
• تحقق من نتيجة المنسق باستخدام النتائج التي تم إرجاعها من المنسق و 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 كافية لاختبار هذه الدالات.

المحتوى ذو الصلة

• تعرف على كيفية تحسين أداء معدل النقل لتطبيقات Python في Azure Functions
• قراءة دليل مطور Azure Functions Python
• تعرف على أفضل ممارسات Durable Functions وأدوات التشخيص