test: add unit tests for backpressure handler methods and conditions
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
This commit is contained in:
@@ -601,6 +601,122 @@ class TestBackpressureHandler:
|
|||||||
|
|
||||||
assert "_wrap_rabbit_mq_api_init" in BackpressureHandler._callback_list
|
assert "_wrap_rabbit_mq_api_init" in BackpressureHandler._callback_list
|
||||||
assert "_wrap_rabbit_mq_api" 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
|
# 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.
|
# not called because mockup overrides that call with preset return value.
|
||||||
# mock_channel.get_exchange.assert_called_once_with(name="cleverthis.clevermicro.management")
|
# mock_channel.get_exchange.assert_called_once_with(name="cleverthis.clevermicro.management")
|
||||||
|
|||||||
Reference in New Issue
Block a user