From bc98351493aae38edc9c2ad81df40c42ff727563 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Fri, 9 May 2025 22:15:11 +0100 Subject: [PATCH] feat \#16 - add unittest for backpressure_handler and logging_utils --- amqp/adapter/backpressure_handler.py | 194 ++++---- amqp/adapter/logging_utils.py | 2 +- tests/__init__.py | 0 tests/adapter/__init__.py | 0 tests/adapter/test_backpressure_handler.py | 533 +++++++++++++++++++++ tests/adapter/test_data_parser.py | 3 + tests/adapter/test_logging_utils.py | 138 ++++++ tests/adapter/test_serializer.py | 42 ++ tests/adapter/test_trace_info_adapter.py | 3 + tests/test_amq_configuration.py | 12 + 10 files changed, 844 insertions(+), 83 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/adapter/__init__.py create mode 100644 tests/adapter/test_backpressure_handler.py create mode 100644 tests/adapter/test_data_parser.py create mode 100644 tests/adapter/test_logging_utils.py create mode 100644 tests/adapter/test_serializer.py create mode 100644 tests/adapter/test_trace_info_adapter.py create mode 100644 tests/test_amq_configuration.py diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index 005f3c0..e5d1b7b 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/amqp/adapter/backpressure_handler.py @@ -48,8 +48,15 @@ class ScaleRequestV1: class RunningAverage: - def __init__(self, num_items): - self.buffer = [0] * num_items # Fixed-size array + """ + A class to maintain a running average of the last N values. + The intended use is to track CPU usage over time window, that is directly related to sample rate and window size. + This is the reason why the constructor takes the number of items to store, calculated as window size / sample rate + Initial_value is for unit test to emulate OVERLOAD or IDLE condition. + """ + + def __init__(self, num_items, initial_value=0): + self.buffer = [initial_value] * num_items # Fixed-size array self.pointer = 0 # Current position in array self.num_items = num_items # Maximum number of items to store self.is_filled = False # Track if buffer is fully filled @@ -89,6 +96,8 @@ class BackpressureHandler: # helps prevent flooding the system with backpressure events last_backpressure_event_time = 0 last_backpressure_event = ScalingRequestAlert.UPDATE + # _callback_list is used in unit tests to record the invoked callbacks + _callback_list = None def __init__( self, @@ -105,6 +114,18 @@ class BackpressureHandler: self.swarm_task_id = self.config.amq_adapter.swarm_task_id self.parallel_workers = self.config.backpressure.threshold_threads self.average_cpu_usage = None + self.do_loop = -1 + + def _do_loop(self) -> bool: + """ + Check if the loop should continue running. + Positive value of do_loop indicates the number of iterations left. + Negative value indicates the loop should run indefinitely. + """ + _val = self.do_loop != 0 + if self.do_loop > 0: + self.do_loop -= 1 + return _val def increase_parallel_executions(self): """Increase the number of parallel executions""" @@ -154,93 +175,96 @@ class BackpressureHandler: self.last_backpressure_event_time = time.time() self.last_backpressure_event = ScalingRequestAlert.OVERLOAD - def backpressure_monitor_loop(self): + async def _backpressure_monitor(self): + """Monitor the backpressure conditions and trigger events accordingly""" _time_window_millis = self.config.backpressure.time_window _idle_duration_millis = self.config.backpressure.idle_duration - _overload_duration_millis = self.config.backpressure.idle_duration - _loop = asyncio.new_event_loop() - _monitor_interval = 0.2 # Monitor every 200ms + _overload_duration_millis = self.config.backpressure.cpu_overload_duration + _monitor_interval = 0.5 # Monitor every 500ms CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2 IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90 - self.average_cpu_usage = RunningAverage( - int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval) - ) + if self.average_cpu_usage is None: + self.average_cpu_usage = RunningAverage( + int( + max(_overload_duration_millis, _idle_duration_millis) + // _monitor_interval + ) + ) + _cpu_usage_state = CPU_ACTIVE + _cpu_usage_changed = time.time() + while self._do_loop(): + _current_cpu_usage = psutil.cpu_percent(interval=None) + self.average_cpu_usage.add(_current_cpu_usage) + _average_cpu_usage = self.average_cpu_usage.get_average() + _now = time.time() + # logging_info( + # "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s", + # _current_cpu_usage, + # _average_cpu_usage, + # _cpu_usage_state, + # self.last_data_message_time, + # self.last_backpressure_event, + # self.last_backpressure_event_time, + # ) + if _average_cpu_usage < IDLE_THRESHOLD: + if _cpu_usage_state != CPU_IDLE: + _cpu_usage_changed = _now + _cpu_usage_state = CPU_IDLE + elif _average_cpu_usage > OVERLOAD_THRESHOLD: + if _cpu_usage_state != CPU_OVERLOAD: + _cpu_usage_changed = _now + _cpu_usage_state = CPU_OVERLOAD + else: + if _cpu_usage_state != CPU_ACTIVE: + _cpu_usage_changed = _now + _cpu_usage_state = CPU_ACTIVE - async def _backpressure_monitor(): - """Monitor the backpressure conditions and trigger events accordingly""" - _cpu_usage_state = CPU_ACTIVE - _cpu_usage_changed = time.time() - while True: - _current_cpu_usage = psutil.cpu_percent(interval=None) - self.average_cpu_usage.add(_current_cpu_usage) - _average_cpu_usage = self.average_cpu_usage.get_average() - _now = time.time() - # logging_info( - # "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s", - # _current_cpu_usage, - # _average_cpu_usage, - # _cpu_usage_state, - # self.last_data_message_time, - # self.last_backpressure_event, - # self.last_backpressure_event_time, - # ) - if _average_cpu_usage < IDLE_THRESHOLD: - if _cpu_usage_state != CPU_IDLE: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_IDLE - elif _average_cpu_usage > OVERLOAD_THRESHOLD: - if _cpu_usage_state != CPU_OVERLOAD: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_OVERLOAD - else: - if _cpu_usage_state != CPU_ACTIVE: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_ACTIVE - - # Check if the current time exceeds the last backpressure event time - if ( - _cpu_usage_state == CPU_OVERLOAD - and _now - _cpu_usage_changed > _overload_duration_millis - and ( - _now - self.last_backpressure_event_time > _time_window_millis - or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD + # Check if the current time exceeds the last backpressure event time + if ( + _cpu_usage_state == CPU_OVERLOAD + and _now - _cpu_usage_changed > _overload_duration_millis + and ( + _now - self.last_backpressure_event_time > _time_window_millis + or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD + ) + ): + # Trigger the overload event + await self.handle_backpressure_overload_event() + self.last_backpressure_event_time = _now + self.last_backpressure_event = ScalingRequestAlert.OVERLOAD + elif ( + _cpu_usage_state == CPU_IDLE + and _now - _cpu_usage_changed > _idle_duration_millis + and _now - self.last_data_message_time > _idle_duration_millis + and ( + _now - self.last_backpressure_event_time > _time_window_millis + or self.last_backpressure_event != ScalingRequestAlert.IDLE + ) + ): + # Trigger the idle event + await self._handle_backpressure_idle_event() + self.last_backpressure_event = ScalingRequestAlert.IDLE + self.last_backpressure_event_time = _now + else: + # Trigger update event + if _now - self.last_backpressure_event_time > _time_window_millis: + # Trigger the update event + _scaling_request: ScaleRequestV1 = ScaleRequestV1( + self.swarm_service_id, + self.swarm_task_id, + 100, + _average_cpu_usage, + ScalingRequestAlert.UPDATE, ) - ): - # Trigger the overload event - await self.handle_backpressure_overload_event() + # Address the message to any adapter capable of supporting BACKPRESSURE request + await self.publish_backpressure_request(_scaling_request) self.last_backpressure_event_time = _now - self.last_backpressure_event = ScalingRequestAlert.OVERLOAD - elif ( - _cpu_usage_state == CPU_IDLE - and _now - _cpu_usage_changed > _idle_duration_millis - and _now - self.last_data_message_time > _idle_duration_millis - and ( - _now - self.last_backpressure_event_time > _time_window_millis - or self.last_backpressure_event != ScalingRequestAlert.IDLE - ) - ): - # Trigger the idle event - await self._handle_backpressure_idle_event() - self.last_backpressure_event = ScalingRequestAlert.IDLE - self.last_backpressure_event_time = _now - else: - # Trigger update event - if _now - self.last_backpressure_event_time > _time_window_millis: - # Trigger the update event - _scaling_request: ScaleRequestV1 = ScaleRequestV1( - self.swarm_service_id, - self.swarm_task_id, - 100, - _average_cpu_usage, - ScalingRequestAlert.UPDATE, - ) - # Address the message to any adapter capable of supporting BACKPRESSURE request - await self.publish_backpressure_request(_scaling_request) - self.last_backpressure_event_time = _now - self.last_backpressure_event = ScalingRequestAlert.UPDATE - await asyncio.sleep(_monitor_interval) + self.last_backpressure_event = ScalingRequestAlert.UPDATE + await asyncio.sleep(_monitor_interval) - _loop.run_until_complete(_backpressure_monitor()) + def backpressure_monitor_loop(self): + _loop = asyncio.new_event_loop() + _loop.run_until_complete(self._backpressure_monitor()) async def handle_backpressure_overload_event(self): logging_warning("Backpressure: Capacity close to depleted!") @@ -283,6 +307,8 @@ class BackpressureHandler: _exchange = await channel.get_exchange(name="cleverthis.clevermicro.management") return _exchange + if BackpressureHandler._callback_list is not None: + BackpressureHandler._callback_list["_wrap_rabbit_mq_api_init"] = 1 self.exchange = await await_result( asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api_init(self.channel), self.loop) ) @@ -312,4 +338,8 @@ class BackpressureHandler: return True return False - await await_future(asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)) + if BackpressureHandler._callback_list is not None: + BackpressureHandler._callback_list["_wrap_rabbit_mq_api"] = 1 + await await_future( + asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop) + ) diff --git a/amqp/adapter/logging_utils.py b/amqp/adapter/logging_utils.py index 1046424..d8ff2d8 100644 --- a/amqp/adapter/logging_utils.py +++ b/amqp/adapter/logging_utils.py @@ -120,7 +120,7 @@ def logging_debug(msg: str, *args) -> None: _thread_id = threading.get_ident() module, line_number, function_name = get_context_info() if module and line_number and function_name: - logging.info( + logging.debug( f"{_thread_id}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'", *args, ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/adapter/__init__.py b/tests/adapter/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/adapter/test_backpressure_handler.py b/tests/adapter/test_backpressure_handler.py new file mode 100644 index 0000000..51c981f --- /dev/null +++ b/tests/adapter/test_backpressure_handler.py @@ -0,0 +1,533 @@ +# +# NOTE: need to install: pytest-asyncio +# + +import asyncio +import json +from asyncio import AbstractEventLoop +from threading import Thread +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from aio_pika import Exchange +from aio_pika.abc import AbstractRobustChannel + +from amqp.adapter.backpressure_handler import ( + BackpressureHandler, + RunningAverage, + ScaleRequestV1, +) +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import ScalingRequestAlert + + +class TestScaleRequestV1: + """ + A class containing pytest unit tests for the ScaleRequestV1 class. + """ + + def test_scale_request_v1_initialization(self): + """ + Test that the ScaleRequestV1 object is initialized correctly. + """ + service_id = "test-service" + task_id = "test-task" + max_availability = 10 + current_availability = 5 + request_type = ScalingRequestAlert.UPDATE + + scale_request = ScaleRequestV1( + service_id, + task_id, + max_availability, + current_availability, + request_type, + ) + + assert scale_request.version == 1 + assert scale_request.serviceId == service_id + assert scale_request.taskId == task_id + assert scale_request.maxAvailability == max_availability + assert scale_request.currentAvailability == current_availability + assert scale_request.requestType == request_type + assert scale_request.thread is None + + def test_scale_request_v1_to_json(self): + """ + Test the to_json method of ScaleRequestV1. + """ + service_id = "test-service" + task_id = "test-task" + max_availability = 10 + current_availability = 5 + request_type = ScalingRequestAlert.UPDATE + + scale_request = ScaleRequestV1( + service_id, + task_id, + max_availability, + current_availability, + request_type, + ) + + json_output = scale_request.to_json() + expected_json = json.dumps( + { + "version": 1, + "serviceId": service_id, + "taskId": task_id, + "maxAvailability": max_availability, + "currentAvailability": current_availability, + "requestType": request_type.value, + }, + indent=2, + ) + assert json_output == expected_json + + def test_scale_request_v1_to_json_different_alert_type(self): + """ + Test the to_json method with a different ScalingRequestAlert type. + """ + service_id = "test-service" + task_id = "test-task" + max_availability = 10 + current_availability = 5 + request_type = ScalingRequestAlert.OVERLOAD # Change the alert type + + scale_request = ScaleRequestV1( + service_id, + task_id, + max_availability, + current_availability, + request_type, + ) + + json_output = scale_request.to_json() + expected_json = json.dumps( + { + "version": 1, + "serviceId": service_id, + "taskId": task_id, + "maxAvailability": max_availability, + "currentAvailability": current_availability, + "requestType": request_type.value, # Ensure the value is correctly serialized + }, + indent=2, + ) + assert json_output == expected_json + + +class TestRunningAverage: + """ + A class containing pytest unit tests for the RunningAverage class. + """ + + def test_running_average_initialization(self): + """ + Test that the RunningAverage object is initialized correctly. + """ + num_items = 5 + ra = RunningAverage(num_items) + assert ra.buffer == [0] * num_items + assert ra.pointer == 0 + assert ra.num_items == num_items + assert ra.is_filled is False + + def test_running_average_add_one_value(self): + """ + Test adding a single value to the RunningAverage. + """ + ra = RunningAverage(3) + ra.add(10) + assert ra.buffer == [10, 0, 0] + assert ra.pointer == 1 + assert ra.is_filled is False + + def test_running_average_add_multiple_values_within_capacity(self): + """ + Test adding multiple values within the capacity of the RunningAverage buffer. + """ + ra = RunningAverage(3) + ra.add(5) + ra.add(10) + ra.add(15) + assert ra.buffer == [5, 10, 15] + assert ra.pointer == 0 + assert ra.is_filled is True + + def test_running_average_add_more_values_than_capacity(self): + """ + Test adding more values than the capacity of the RunningAverage buffer, checking wrap-around. + """ + ra = RunningAverage(3) + ra.add(1) + ra.add(2) + ra.add(3) + ra.add(4) # Overwrite the first value + ra.add(5) # Overwrite the second value + assert ra.buffer == [4, 5, 3] + assert ra.pointer == 2 + assert ra.is_filled is True + + def test_running_average_get_average_empty(self): + """ + Test get_average when no values have been added. + """ + ra = RunningAverage(3) + assert ra.get_average() == 0.0 + + def test_running_average_get_average_one_value(self): + """ + Test get_average after adding one value. + """ + ra = RunningAverage(3) + ra.add(10) + assert ra.get_average() == 10.0 + + def test_running_average_get_average_multiple_values_within_capacity(self): + """ + Test get_average with multiple values added within the buffer's capacity. + """ + ra = RunningAverage(3) + ra.add(1) + ra.add(2) + ra.add(3) + assert ra.get_average() == 2.0 + + def test_running_average_get_average_more_values_than_capacity(self): + """ + Test get_average after adding more values than the buffer's capacity (check wrap-around). + """ + ra = RunningAverage(3) + ra.add(1) + ra.add(2) + ra.add(3) + ra.add(4) + ra.add(5) + assert ra.get_average() == 4.0 + + def test_get_current_values_empty(self): + """Test get_current_values when no values have been added.""" + ra = RunningAverage(3) + assert ra.get_current_values() == [] + + def test_get_current_values_one_value(self): + """Test get_current_values after adding one value.""" + ra = RunningAverage(3) + ra.add(10) + assert ra.get_current_values() == [10] + + def test_get_current_values_multiple_values_within_capacity(self): + """Test get_current_values with multiple values within capacity.""" + ra = RunningAverage(3) + ra.add(1) + ra.add(2) + ra.add(3) + assert ra.get_current_values() == [1, 2, 3] + + def test_get_current_values_more_values_than_capacity(self): + """Test get_current_values after adding more values than capacity (wrap-around).""" + ra = RunningAverage(3) + ra.add(1) + ra.add(2) + ra.add(3) + ra.add(4) + ra.add(5) + assert ra.get_current_values() == [3, 4, 5] + + def test_get_current_values_wrap_around(self): + """Test get_current_values when the buffer has wrapped around.""" + ra = RunningAverage(4) + ra.add(1) + ra.add(2) + ra.add(3) + ra.add(4) + ra.add(5) # Overwrite 1 + ra.add(6) # Overwrite 2 + assert ra.get_current_values() == [3, 4, 5, 6] + + +# Test for BackpressureHandler +class TestBackpressureHandler: + """ + A class containing pytest unit tests for the BackpressureHandler class. + """ + + @pytest.fixture + def mock_channel(self): + """ + Pytest fixture to create a mock AbstractRobustChannel. + """ + return AsyncMock(spec=AbstractRobustChannel) + + @pytest.fixture + def mock_loop(self): + """ + Pytest fixture to create a mock AbstractEventLoop. + """ + return MagicMock(spec=AbstractEventLoop) + + @pytest.fixture + def mock_config(self): + """ + Pytest fixture to create a mock AMQConfiguration. + """ + return AMQConfiguration("") + + def test_backpressure_handler_initialization( + self, mock_channel, mock_loop, mock_config + ): + """ + Test that the BackpressureHandler object is initialized correctly. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + assert handler.channel == mock_channel + assert handler.loop == mock_loop + assert handler.config == mock_config + assert handler.swarm_service_id == "clever-amqp-service" + assert handler.swarm_task_id == "python-adapter" + assert handler.parallel_workers == 1 + assert handler.average_cpu_usage is None + + def test_increase_parallel_executions(self, mock_channel, mock_loop, mock_config): + """ + Test the increase_parallel_executions method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + initial_executions = handler.current_parallel_executions + handler.increase_parallel_executions() + assert handler.current_parallel_executions == initial_executions + 1 + + def test_decrease_parallel_executions(self, mock_channel, mock_loop, mock_config): + """ + Test the decrease_parallel_executions method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.current_parallel_executions = 5 + handler.decrease_parallel_executions() + assert handler.current_parallel_executions == 4 + + handler.current_parallel_executions = 0 + handler.decrease_parallel_executions() + assert handler.current_parallel_executions == 0 # Should not go below 0 + + def test_update_parallel_executions(self, mock_channel, mock_loop, mock_config): + """ + Test the update_parallel_executions method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.update_parallel_executions(100) + assert handler.current_parallel_executions == 100 + + def test_update_last_data_message_time(self, mock_channel, mock_loop, mock_config): + """ + Test the update_last_data_message_time method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + with patch("time.time") as mock_time: + mock_time.return_value = 12345 + handler.update_last_data_message_time() + assert handler.last_data_message_time == 12345 + + def test_start_backpressure_monitor(self, mock_channel, mock_loop, mock_config): + """ + Test the start_backpressure_monitor method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + thread = handler.start_backpressure_monitor() + assert isinstance(thread, Thread) + assert handler.thread == thread + + @pytest.mark.asyncio + async def test_check_overload_condition_no_overload( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the check_overload_condition method when there is no overload. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.current_parallel_executions = 1 + handler.parallel_workers = 5 + handler.last_backpressure_event = ScalingRequestAlert.UPDATE + with patch.object( + handler, "handle_backpressure_overload_event" + ) as mock_handle_overload: + await handler.check_overload_condition() + mock_handle_overload.assert_not_called() + + @pytest.mark.asyncio + async def test_check_overload_condition_overload( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the check_overload_condition method when there is an overload. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.current_parallel_executions = 9 # Assuming parallel_workers is 10 + handler.last_backpressure_event = ScalingRequestAlert.UPDATE + with patch.object( + handler, "handle_backpressure_overload_event" + ) as mock_handle_overload: + with patch("time.time") as mock_time: + mock_time.return_value = 12345 + await handler.check_overload_condition() + mock_handle_overload.assert_called_once() + assert handler.last_backpressure_event_time == 12345 + assert handler.last_backpressure_event == ScalingRequestAlert.OVERLOAD + + @pytest.mark.asyncio + async def test_check_overload_condition_already_overload( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the check_overload_condition method when the last event was already an overload. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.current_parallel_executions = 10 + handler.last_backpressure_event = ScalingRequestAlert.OVERLOAD + with patch.object( + handler, "handle_backpressure_overload_event" + ) as mock_handle_overload: + await handler.check_overload_condition() + mock_handle_overload.assert_not_called() + + # backpressure_monitor_loop is difficult to test directly without significant mocking + # and potentially long-running tests. The following is a basic outline of how + # you might approach testing it, focusing on verifying that the correct methods + # are called under specific conditions. This is NOT a complete, runnable test. + @pytest.mark.asyncio + async def test_backpressure_monitor_loop( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the backpressure_monitor_loop method (outline). + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.average_cpu_usage = RunningAverage( + 100, initial_value=99 + ) # Mock the RunningAverage in OVERLOAD state + handler.average_cpu_usage.is_filled = ( + True # Together with 0 values ensures OVERLOAD state + ) + # Setup mocks for dependent methods + handler.handle_backpressure_overload_event = AsyncMock() + handler._handle_backpressure_idle_event = AsyncMock() + handler.publish_backpressure_request = AsyncMock() + + # 1. Simulate an overload condition: + handler.last_backpressure_event_time = 0 # Force event trigger + handler.last_data_message_time = 0 + handler.do_loop = 1 + handler.config.backpressure.cpu_overload_duration = ( + -1 + ) # Force overload condition even for recent change + await handler._backpressure_monitor() + handler.handle_backpressure_overload_event.assert_called_once() + + # 2. Simulate an idle condition + handler.average_cpu_usage = RunningAverage( + 100, initial_value=0 + ) # Mock the RunningAverage in IDLE state + handler.average_cpu_usage.is_filled = ( + True # Together with 0 values ensures IDLE state + ) + handler.last_backpressure_event_time = 0 + handler.last_data_message_time = 0 + handler.do_loop = 1 + handler.config.backpressure.idle_duration = ( + -1 + ) # Force idle condition even for recent change + await handler._backpressure_monitor() + handler._handle_backpressure_idle_event.assert_called_once() + + # 3. Simulate neither idle nor overload, but time_window passed + handler.average_cpu_usage = RunningAverage( + 100 + ) # Mock the RunningAverage in pritine state + handler.last_backpressure_event_time = 0 + handler.last_data_message_time = 0 + handler.do_loop = 1 + await handler._backpressure_monitor() + handler.publish_backpressure_request.assert_called_once() + + @pytest.mark.asyncio + async def test_handle_backpressure_overload_event( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the handle_backpressure_overload_event method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.publish_backpressure_request = AsyncMock() + with patch("time.time") as mock_time: + mock_time.return_value = 12345 + await handler.handle_backpressure_overload_event() + assert handler.publish_backpressure_request.call_count == 1 + assert handler.last_data_message_time == 12345 + + @pytest.mark.asyncio + async def test_handle_backpressure_idle_event( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the _handle_backpressure_idle_event method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.publish_backpressure_request = AsyncMock() + with patch("time.time") as mock_time: + mock_time.return_value = 12345 + await handler._handle_backpressure_idle_event() + assert handler.publish_backpressure_request.call_count == 1 + assert handler.last_data_message_time == 12345 + + @pytest.mark.asyncio + async def test_publish_backpressure_request_exchange_exists( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the publish_backpressure_request method when the exchange already exists. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.exchange = AsyncMock(spec=Exchange) + with patch("asyncio.run_coroutine_threadsafe") as mock_async_run: + mock_async_run.return_value = asyncio.Future() + mock_async_run.return_value.set_result(None) + await handler.publish_backpressure_request( + ScaleRequestV1( + "test_service", "test_task", 100, 50, ScalingRequestAlert.UPDATE + ) + ) + assert mock_async_run.call_count == 1 + # handler.exchange.publish.assert_called_once() + + @pytest.mark.asyncio + async def test_publish_backpressure_request_exchange_does_not_exist( + self, mock_channel, mock_loop, mock_config + ): + """ + Test the publish_backpressure_request method when the exchange does not exist and needs to be created. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.exchange = None # Simulate exchange not existing + mock_exchange = AsyncMock(spec=Exchange) + mock_channel.get_exchange.return_value = mock_exchange + BackpressureHandler._callback_list = {} + + with patch("asyncio.run_coroutine_threadsafe") as mock_async_run: + mock_async_run.return_value = asyncio.Future() + mock_async_run.return_value.set_result( + mock_exchange + ) # Make it return the mock exchange + await handler.publish_backpressure_request( + ScaleRequestV1( + "test_service", "test_task", 100, 50, ScalingRequestAlert.UPDATE + ) + ) + assert mock_async_run.call_count == 2 + + assert "_wrap_rabbit_mq_api_init" in BackpressureHandler._callback_list + assert "_wrap_rabbit_mq_api" in BackpressureHandler._callback_list + # Since I need also to mockup the asyncio.run_coroutine_threadsafe, the following mockups are + # not called because mockup overrides that call with preset return value. + # mock_channel.get_exchange.assert_called_once_with(name="cleverthis.clevermicro.management") + # handler.exchange.publish.assert_called_once() diff --git a/tests/adapter/test_data_parser.py b/tests/adapter/test_data_parser.py new file mode 100644 index 0000000..be709f5 --- /dev/null +++ b/tests/adapter/test_data_parser.py @@ -0,0 +1,3 @@ +class DataParserTest: + def test_multipart_form_data_parser(self): + assert False diff --git a/tests/adapter/test_logging_utils.py b/tests/adapter/test_logging_utils.py new file mode 100644 index 0000000..9ca5ea3 --- /dev/null +++ b/tests/adapter/test_logging_utils.py @@ -0,0 +1,138 @@ +import os +from importlib.metadata import version +from unittest.mock import patch + +from amqp.adapter.logging_utils import ( + get_context_info, + logging_debug, + logging_error, + logging_info, + logging_warning, +) + +__version__ = version("amq_adapter") # Name must match pyproject.toml [project] name +__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1" + + +class TestLoggingFunctions: + """ + A class containing pytest unit tests for the logging functions. + """ + + @patch("logging.error") + def test_logging_error_with_exception(self, mock_logger): + """ + Test logging_error function when called within an exception handler. + """ + try: + raise ValueError("Test error") + except ValueError as ve: + logging_error("An error occurred: %s", ve) + mock_logger.assert_called_once() + # Assert that the traceback information is included in the log message + assert len(mock_logger.call_args_list) == 1 + _val = mock_logger.call_args_list[0][0][0] + assert "[E] 'An error occurred: %s'" in _val + assert "test_logging_error_with_exception" in _val + + @patch("logging.error") + def test_logging_error_without_exception(self, mock_logger): + """ + Test logging_error function when called without an active exception. + """ + logging_error("A regular error") + mock_logger.assert_called_once() + assert "A regular error" in mock_logger.call_args_list[0][0][0] + + @patch("logging.warning") + def test_logging_warning_verbose_on(self, mock_logger): + """ + Test logging_warning function when verbose logging is enabled. + """ + with patch( + "amqp.adapter.logging_utils.__verbose__", True + ): # Simulate verbose logging + logging_warning("A warning message") + mock_logger.assert_called_once() + assert "A warning message" in mock_logger.call_args_list[0][0][0] + + @patch("logging.warning") + def test_logging_warning_verbose_off(self, mock_logger): + """ + Test logging_warning function when verbose logging is disabled. + """ + with patch( + "amqp.adapter.logging_utils.__verbose__", False + ): # Simulate non-verbose logging + logging_warning("A warning message") + mock_logger.assert_called_once() + assert "A warning message" in mock_logger.call_args_list[0][0][0] + + @patch("logging.info") + def test_logging_info_verbose_on(self, mock_logger): + """ + Test logging_info function with verbose logging enabled. + """ + with patch("amqp.adapter.logging_utils.__verbose__", True): + logging_info("An info message") + mock_logger.assert_called_once() + assert "An info message" in mock_logger.call_args_list[0][0][0] + + @patch("logging.info") + def test_logging_info_verbose_off(self, mock_logger): + """ + Test logging_info function with verbose logging disabled. + """ + with patch("amqp.adapter.logging_utils.__verbose__", False): + logging_info("An info message") + mock_logger.assert_called_once() + assert "An info message" in mock_logger.call_args_list[0][0][0] + + @patch("logging.debug") + def test_logging_debug(self, mock_logger): + """ + Test logging_debug function. + """ + logging_debug("A debug message") + mock_logger.assert_called_once() + assert "A debug message" in mock_logger.call_args_list[0][0][0] + + def test_get_context_info_normal(self): + """ + Test get_context_info function in a normal function call. + """ + + def dummy_function(): + return get_context_info() + + module_name, line_number, function_name = dummy_function() + assert ( + "test_get_context_info_normal" in function_name + ) # changed from co_name to function name + assert ( + "test_logging_utils.py" in module_name + ) # changed from None to test_unit.py + assert isinstance(line_number, int) + + def test_get_context_info_no_frame(self): + """ + Test get_context_info function when no frame is available. + """ + # Simulate a situation where inspect.currentframe() returns None (which is hard to do directly) + with patch("inspect.currentframe") as mock_frame: + mock_frame.return_value = None + module_name, line_number, function_name = get_context_info() + assert module_name is None + assert line_number is None + assert function_name is None + + def test_get_context_info_exception(self): + """ + Test get_context_info function when an exception occurs. + """ + with patch("inspect.currentframe") as mock_frame: + mock_frame.side_effect = Exception("Simulated exception") + module_name, line_number, function_name = get_context_info() + assert module_name is None + assert line_number is None + assert function_name is None diff --git a/tests/adapter/test_serializer.py b/tests/adapter/test_serializer.py new file mode 100644 index 0000000..d15a91f --- /dev/null +++ b/tests/adapter/test_serializer.py @@ -0,0 +1,42 @@ +def test_long_to_bytes(): + assert False + + +def test_bytes_to_long(): + assert False + + +def test_read_long(): + assert False + + +def test_object_or_list_as_string(): + assert False + + +def test_map_as_string(): + assert False + + +def test_map_with_list_as_string(): + assert False + + +def test_parse_map_string(): + assert False + + +def test_csv_as_list(): + assert False + + +def test_add_to_map(): + assert False + + +def test_parse_map_list_string(): + assert False + + +def test_str_to_bool(): + assert False diff --git a/tests/adapter/test_trace_info_adapter.py b/tests/adapter/test_trace_info_adapter.py new file mode 100644 index 0000000..8299acf --- /dev/null +++ b/tests/adapter/test_trace_info_adapter.py @@ -0,0 +1,3 @@ +class TraceInfoAdapterTest: + def test_create_produce_span(self): + assert False diff --git a/tests/test_amq_configuration.py b/tests/test_amq_configuration.py new file mode 100644 index 0000000..afd5422 --- /dev/null +++ b/tests/test_amq_configuration.py @@ -0,0 +1,12 @@ +class AMQConfigurationTest: + def test_amqadapter(self): + assert False + + def test_dispatch(self): + assert False + + def test_backpressure(self): + assert False + + def test_amqconfiguration(self): + assert False