clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

test: add AsyncMock to test backpressure handler conditions

Browse Source 
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
This commit is contained in:
Stanislav Hejny
2025-07-12 23:02:19 +01:00
parent 0e294b223d
commit 7d4e358ce5
1 changed files with 15 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapter/test_backpressure_handler.py
+15 -3
View File
@@ -351,9 +351,18 @@ class TestBackpressureHandler:
handler.current_parallel_executions = 1 handler.current_parallel_executions = 1
handler.parallel_workers = 5 handler.parallel_workers = 5
handler.last_backpressure_event = ScalingRequestAlert.UPDATE handler.last_backpressure_event = ScalingRequestAlert.UPDATE
with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: handler.publish_backpressure_request = AsyncMock()
await handler.check_overload_condition() with patch("time.time") as mock_time:
mock_handle_overload.assert_not_called() mock_time.return_value = 12345
with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload:
await handler.check_overload_condition()
mock_handle_overload.assert_not_called()
if 12345 - handler.last_backpressure_event_time > mock_config.backpressure.time_window:
handler.publish_backpressure_request.assert_called_once()
assert handler.last_backpressure_event == ScalingRequestAlert.UPDATE
assert handler.last_backpressure_event_time == 12345
else:
handler.publish_backpressure_request.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_check_overload_condition_overload(self, mock_channel, mock_loop, mock_config): async def test_check_overload_condition_overload(self, mock_channel, mock_loop, mock_config):
@@ -363,7 +372,9 @@ class TestBackpressureHandler:
handler = BackpressureHandler(mock_channel, mock_loop, mock_config) handler = BackpressureHandler(mock_channel, mock_loop, mock_config)
handler.current_parallel_executions = 9 # Assuming parallel_workers is 10 handler.current_parallel_executions = 9 # Assuming parallel_workers is 10
handler.last_backpressure_event = ScalingRequestAlert.UPDATE handler.last_backpressure_event = ScalingRequestAlert.UPDATE
handler.publish_backpressure_request = AsyncMock()
with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload:
mock_handle_overload.return_value = None
with patch("time.time") as mock_time: with patch("time.time") as mock_time:
mock_time.return_value = 12345 mock_time.return_value = 12345
await handler.check_overload_condition() await handler.check_overload_condition()
@@ -381,6 +392,7 @@ class TestBackpressureHandler:
handler = BackpressureHandler(mock_channel, mock_loop, mock_config) handler = BackpressureHandler(mock_channel, mock_loop, mock_config)
handler.current_parallel_executions = 10 handler.current_parallel_executions = 10
handler.last_backpressure_event = ScalingRequestAlert.OVERLOAD handler.last_backpressure_event = ScalingRequestAlert.OVERLOAD
handler.publish_backpressure_request = AsyncMock()
with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload:
await handler.check_overload_condition() await handler.check_overload_condition()
mock_handle_overload.assert_not_called() mock_handle_overload.assert_not_called()