diff --git a/tests/adapter/test_backpressure_handler.py b/tests/adapter/test_backpressure_handler.py index d2b75e0..b1987fa 100644 --- a/tests/adapter/test_backpressure_handler.py +++ b/tests/adapter/test_backpressure_handler.py @@ -601,6 +601,122 @@ class TestBackpressureHandler: assert "_wrap_rabbit_mq_api_init" in BackpressureHandler._callback_list assert "_wrap_rabbit_mq_api" in BackpressureHandler._callback_list + + @pytest.mark.asyncio + async def test_update_backpressure_value(self, mock_channel, mock_loop, mock_config): + """ + Test the update_backpressure_value method. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.update_parallel_executions = MagicMock() + handler.update_last_data_message_time = MagicMock() + handler.check_overload_condition = AsyncMock() + + # Test with current value and same maximum + await handler.update_backpressure_value(5, handler.parallel_workers) + handler.update_parallel_executions.assert_called_once_with(5) + handler.update_last_data_message_time.assert_called_once() + handler.check_overload_condition.assert_called_once() + + # Reset mocks + handler.update_parallel_executions.reset_mock() + handler.update_last_data_message_time.reset_mock() + handler.check_overload_condition.reset_mock() + + # Test with current value and different maximum + original_workers = handler.parallel_workers + await handler.update_backpressure_value(3, original_workers + 5) + handler.update_parallel_executions.assert_called_once_with(3) + handler.update_last_data_message_time.assert_called_once() + handler.check_overload_condition.assert_called_once() + assert handler.parallel_workers == original_workers + 5 + + @pytest.mark.asyncio + async def test_check_overload_condition_idle(self, mock_channel, mock_loop, mock_config): + """ + Test the check_overload_condition method when service is idle. + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler._handle_backpressure_idle_event = AsyncMock() + + # Set up initial state for IDLE condition + handler.current_parallel_executions = 0 + handler.last_backpressure_event = ScalingRequestAlert.UPDATE + + with patch("time.time") as mock_time: + # Set current time + current_time = 1000 + mock_time.return_value = current_time + + # Set last data message time to be older than idle_duration + handler.last_data_message_time = current_time - mock_config.backpressure.idle_duration - 10 + handler.last_backpressure_event_time = current_time - mock_config.backpressure.time_window - 10 + + # Call the method + await handler.check_overload_condition() + + # Verify IDLE event was triggered + handler._handle_backpressure_idle_event.assert_called_once() + assert handler.last_backpressure_event == ScalingRequestAlert.IDLE + assert handler.last_backpressure_event_time == current_time + + @pytest.mark.asyncio + async def test_check_overload_condition_update(self, mock_channel, mock_loop, mock_config): + """ + Test the check_overload_condition method when service is in normal state (UPDATE). + """ + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.publish_backpressure_request = AsyncMock() + + # Set up initial state for UPDATE condition + handler.current_parallel_executions = 2 + handler.parallel_workers = 10 + handler.last_backpressure_event = ScalingRequestAlert.IDLE # Coming from IDLE state + + with patch("time.time") as mock_time: + # Set current time + current_time = 1000 + mock_time.return_value = current_time + + # Set last event time to be older than time_window + handler.last_backpressure_event_time = current_time - mock_config.backpressure.time_window - 10 + + # Call the method + await handler.check_overload_condition() + + # Verify UPDATE event was triggered + handler.publish_backpressure_request.assert_called_once() + assert handler.last_backpressure_event == ScalingRequestAlert.UPDATE + assert handler.last_backpressure_event_time == current_time + + def test_start_backpressure_monitor_disabled(self, mock_channel, mock_loop, mock_config): + """ + Test the start_backpressure_monitor method when CPU monitoring is disabled. + """ + # Set CPU monitoring to disabled in config + mock_config.backpressure.cpu_monitoring_enabled = False + + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + thread = handler.start_backpressure_monitor() + + # Verify no thread was started + assert thread is None + assert not hasattr(handler, 'thread') or handler.thread is None + + def test_start_backpressure_monitor_enabled(self, mock_channel, mock_loop, mock_config): + """ + Test the start_backpressure_monitor method when CPU monitoring is enabled. + """ + # Set CPU monitoring to enabled in config + mock_config.backpressure.cpu_monitoring_enabled = True + + handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.do_loop = 1 # allow only 1 execution of the loop and then exit the thread + thread = handler.start_backpressure_monitor() + + # Verify thread was started + assert isinstance(thread, Thread) + assert handler.thread == thread # 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")