Stanislav Hejny
88 lines
3.8 KiB
Python
88 lines
3.8 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 second argument should be the loop
|
|
loop_arg = mock_run.call_args[1]["loop"]
|
|
|
|
self.assertEqual(loop_arg, self.service.backpressure_handler.loop)
|
|
|
|
# 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)
|
|
mock_run.assert_called_once()
|
|
|
|
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)
|
|
mock_run.assert_called_once()
|
|
|
|
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()
|