Stanislav Hejny
ef5a0363cd
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
100 lines
4.2 KiB
Python
100 lines
4.2 KiB
Python
import asyncio
|
|
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from amqp.service.amq_service import AMQService
|
|
|
|
|
|
class TestAMQService(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
# Mock the configuration
|
|
self.config = MagicMock()
|
|
self.config.amq_adapter.service_name = "test-service"
|
|
self.config.amq_adapter.swarm_task_id = "test-task"
|
|
self.config.amq_adapter.swarm_service_id = "test-service-id"
|
|
self.config.backpressure.threshold_load = 100
|
|
|
|
# Mock the service adapter
|
|
self.service_adapter = MagicMock()
|
|
|
|
# Patch the get_unique_instance_id function
|
|
with patch("amqp.service.amq_service.get_unique_instance_id", return_value=12345):
|
|
# Create the AMQService instance
|
|
self.service = AMQService(self.config, self.service_adapter)
|
|
|
|
# Mock the backpressure handler with AsyncMock for the coroutine method
|
|
self.service.backpressure_handler = MagicMock()
|
|
self.service.backpressure_handler.update_backpressure_value = AsyncMock()
|
|
self.service.backpressure_handler.loop = asyncio.get_event_loop()
|
|
|
|
async def test_backpressure_with_maximum(self):
|
|
"""Test backpressure method with a specified maximum value."""
|
|
# Call the backpressure method with a maximum value
|
|
self.service.backpressure(current_availability=50, maximum=200)
|
|
|
|
# Check that run_coroutine_threadsafe was called with the correct arguments
|
|
# We need to patch asyncio.run_coroutine_threadsafe to capture its arguments
|
|
with patch('asyncio.run_coroutine_threadsafe') as mock_run:
|
|
self.service.backpressure(current_availability=50, maximum=200)
|
|
mock_run.assert_called_once()
|
|
|
|
# The first argument should be the coroutine
|
|
coro_arg = mock_run.call_args[0][0]
|
|
# The second argument should be the loop
|
|
loop_arg = mock_run.call_args[0][1]
|
|
|
|
self.assertEqual(loop_arg, self.service.backpressure_handler.loop)
|
|
|
|
# For the coroutine, we can check its __qualname__ to verify it's the right method
|
|
self.assertEqual(
|
|
coro_arg.__qualname__,
|
|
'BackpressureHandler.update_backpressure_value'
|
|
)
|
|
|
|
# We can also verify the arguments passed to update_backpressure_value
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_with(
|
|
50, 200
|
|
)
|
|
|
|
async def test_backpressure_without_maximum(self):
|
|
"""Test backpressure method without a maximum value."""
|
|
# Set a maximum_availability value
|
|
self.service.maximum_availability = 100
|
|
|
|
with patch('asyncio.run_coroutine_threadsafe') as mock_run:
|
|
# Call the backpressure method without a maximum value
|
|
self.service.backpressure(current_availability=50)
|
|
|
|
# Verify the handler was called with the right values
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_with(
|
|
50, 100
|
|
)
|
|
|
|
async def test_backpressure_with_negative_maximum(self):
|
|
"""Test backpressure method with a negative maximum value."""
|
|
# Set a maximum_availability value
|
|
self.service.maximum_availability = 100
|
|
|
|
with patch('asyncio.run_coroutine_threadsafe') as mock_run:
|
|
# Call the backpressure method with a negative maximum value
|
|
self.service.backpressure(current_availability=50, maximum=-1)
|
|
|
|
# Verify the handler was called with the right values
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_with(
|
|
50, 100
|
|
)
|
|
|
|
async def test_backpressure_no_handler(self):
|
|
"""Test backpressure method when no handler is available."""
|
|
# Set the backpressure handler to None
|
|
self.service.backpressure_handler = None
|
|
|
|
# Call the backpressure method
|
|
self.service.backpressure(current_availability=50)
|
|
|
|
# Nothing should happen, no exception should be raised
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|