Bagikan melalui

Pengujian unit pada Durable Functions dengan Python

Pengujian unit adalah bagian penting dari praktik pengembangan perangkat lunak modern. Pengujian unit memverifikasi perilaku logika bisnis dan melindungi dari perubahan yang merusak yang tidak terdeteksi di masa mendatang. Durable Functions dapat dengan mudah menjadi lebih kompleks sehingga memperkenalkan pengujian unit dapat membantu menghindari perubahan yang melanggar. Bagian berikut menjelaskan cara menguji tiga jenis fungsi - Fungsi klien orkestrasi, orkestrator, dan entitas.

Nota

Panduan ini hanya berlaku untuk aplikasi Durable Functions yang ditulis dalam model pemrograman Python v2.

Prasyarat

Contoh dalam artikel ini memerlukan pengetahuan tentang konsep dan kerangka kerja berikut:

Menyiapkan lingkungan pengujian

Untuk menguji Durable Functions, sangat penting untuk menyiapkan lingkungan pengujian yang tepat. Ini termasuk membuat direktori pengujian dan menginstal modul Python unittest ke lingkungan Python Anda. Untuk informasi selengkapnya, lihat ringkasan pengujian unit Azure Functions Python.

Fungsi pemicu pengujian unit

Fungsi pemicu, sering disebut sebagai fungsi klien , memulai orkestrasi dan peristiwa eksternal. Untuk menguji fungsi-fungsi ini:

  • Membuat tiruan dari DurableOrchestrationClient guna mensimulasikan orkestrasi eksekusi dan manajemen status.
  • Tetapkan DurableOrchestrationClient metode seperti start_new, , get_statusatau raise_event dengan fungsi tiruan yang mengembalikan nilai yang diharapkan.
  • Panggil fungsi klien secara langsung dengan klien yang ditiru serta input lain yang diperlukan, seperti req (objek permintaan HTTP) untuk fungsi klien dengan pemicu HTTP.
  • Gunakan pernyataan dan unittest.mock alat untuk memverifikasi perilaku mulai orkestrasi, parameter, dan respons HTTP yang diharapkan.
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")

Fungsi pengatur pengujian unit

Fungsi orkestrator mengelola eksekusi beberapa fungsi aktivitas. Untuk menguji orkestrator:

  • Membuat tiruan DurableOrchestrationContext untuk mengontrol eksekusi fungsi.
  • Ganti DurableOrchestrationContext metode yang diperlukan untuk eksekusi orkestrator seperti call_activity atau create_timer dengan fungsi tiruan. Fungsi-fungsi ini biasanya akan mengembalikan objek jenis TaskBase dengan result properti .
  • Panggil orkestrator secara rekursif, meneruskan hasil Tugas yang dihasilkan oleh pernyataan hasil sebelumnya ke pernyataan berikutnya.
  • Verifikasi hasil orkestrator menggunakan hasil yang dikembalikan dari orkestrator dan 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!"])

Fungsi entitas pengujian unit

Fungsi entitas mengelola objek stateful dengan operasi. Untuk menguji fungsi entitas:

  • Mock DurableEntityContext untuk mensimulasikan status internal entitas dan input operasi.
  • Ganti DurableEntityContext metode seperti get_state, set_state, dan operation_name dengan tiruan yang mengembalikan nilai terkontrol.
  • Panggil fungsi entitas secara langsung dengan konteks yang ditidakan.
  • Gunakan pernyataan untuk memverifikasi perubahan status dan nilai yang dikembalikan, bersama dengan unittest.mock utilitas.
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)

Fungsi aktivitas pengujian unit

Fungsi aktivitas tidak memerlukan modifikasi khusus tahan lama untuk diuji. Panduan yang ditemukan dalam ringkasan pengujian unit Azure Functions Python cukup untuk menguji fungsi-fungsi ini.

Konten terkait