Aracılığıyla paylaş

Python'da Dayanıklı Fonksiyonlar için Birim Testi

Birim testi, modern yazılım geliştirme uygulamalarının önemli bir parçasıdır. Birim testleri iş mantığı davranışını doğrular ve gelecekte fark edilmeden bozukluklara neden olan değişikliklerden korur. Dayanıklı İşlevler karmaşıklığı kolayca artabilir, bu nedenle birim testlerinin getirilmesi hataya neden olan değişiklikleri önlemeye yardımcı olur. Aşağıdaki bölümlerde orchestration istemcisi, orchestrator ve varlık işlevleri olmak üzere üç işlev türünün nasıl birim testi yapılacağı açıklanmaktadır.

Uyarı

Bu kılavuz yalnızca Python v2 programlama modelinde yazılan Dayanıklı İşlevler uygulamaları için geçerlidir.

Önkoşullar

Bu makaledeki örnekler aşağıdaki kavramlar ve çerçeveler hakkında bilgi gerektirir:

Test ortamını ayarlama

Dayanıklı İşlevleri test etmek için uygun bir test ortamı ayarlamak çok önemlidir. Bu, bir test dizini oluşturmayı ve Python'ın unittest modülünü Python ortamınıza yüklemeyi içerir. Daha fazla bilgi için bkz. Azure İşlevleri Python birim testlerine genel bakış.

Birim testi tetikleyici işlevleri

Genellikle istemci işlevleri olarak adlandırılan tetikleyici işlevleri, düzenlemeleri ve dış olayları başlatır. Bu işlevleri test etmek için:

  • Orkestrasyon yürütmeyi ve durum yönetimini simüle etmek için DurableOrchestrationClient'yi taklit edin.
  • DurableOrchestrationClient gibi yöntemlere, start_new, get_status veya raise_event gibi beklenen değerleri döndüren sahte işlevler atayın.
  • İstemci işlevini doğrudan taklit edilmiş bir istemci ve HTTP tetikleyicisi istemci işlevleri için gerekli diğer girdilerle, örneğin bir req (HTTP istek nesnesi), çağırın.
  • Beklenen orkestrasyon başlatma davranışını, parametrelerini ve HTTP yanıtlarını doğrulamak için doğrulamalar ve unittest.mock araçlarını kullanın.
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")

Birim testi düzenleyici işlevleri

Orchestrator işlevleri birden çok etkinlik işlevinin yürütülmesini yönetir. Düzenleyiciyi test etmek için:

  • İşlev yürütmesini kontrol etmek için DurableOrchestrationContext'yi taklit edin.
  • DurableOrchestrationContext veya call_activity gibi düzenleyici yürütüm için gereken yöntemleri sahte işlevlerle değiştirin. Bu işlevler genellikle bir result özelliğe sahip TaskBase türünde nesneler döndürür.
  • Önceki yield ifadesi tarafından oluşturulan Görevin sonucunu sonrakine geçirerek yönlendiriciyi özyinelemeli olarak çağırın.
  • orchestrator ve unittest.mock'den döndürülen sonuçları kullanarak orchestrator sonucunu doğrulayın.
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!"])

Birim testi nesne işlevleri

Varlık işlevleri durum bilgisi olan nesneleri işlemlerle yönetir. Bir varlık işlevini test etmek için:

  • Varlığın DurableEntityContext iç durumunun ve işlem girişlerinin benzetimini yapmak için mock edin.
  • DurableEntityContext yöntemlerinin yerine, get_state, set_state ve operation_name gibi denetimli değerler döndüren sahte yöntemler kullanın.
  • Varlık işlevini simüle edilmiş bağlamla doğrudan çağırın.
  • Durum değişikliklerini ve döndürülen değerleri, unittest.mock yardımcı araçlarla birlikte doğrulamak için doğrulamalar kullanın.
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)

Birim testi etkinlik işlevleri

Etkinlik işlevlerinin test edilmesi için Dayanıklıya özgü değişiklikler gerekmez. Azure İşlevleri Python birim testlerine genel bakış bölümünde bulunan yönergeler bu işlevleri test etmek için yeterlidir.

İlgili içerik

  • Last updated on