Stanislav Hejny
75 lines
3.4 KiB
Python
75 lines
3.4 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
|
|
self.service.backpressure_handler = MagicMock()
|
|
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 update_backpressure_value was called with the correct values
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_once()
|
|
# Get the coroutine that was passed to run_coroutine_threadsafe
|
|
coro = self.service.backpressure_handler.update_backpressure_value.call_args[0]
|
|
self.assertEqual(coro, (50, 200))
|
|
|
|
async def test_backpressure_without_maximum(self):
|
|
"""Test backpressure method without a maximum value."""
|
|
# Call the backpressure method without a maximum value
|
|
self.service.backpressure(current_availability=50)
|
|
|
|
# Check that update_backpressure_value was called with the correct values
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_once()
|
|
# Get the coroutine that was passed to run_coroutine_threadsafe
|
|
coro = self.service.backpressure_handler.update_backpressure_value.call_args[0]
|
|
self.assertEqual(coro, (50, 100)) # Should use threshold_load from config
|
|
|
|
async def test_backpressure_with_negative_maximum(self):
|
|
"""Test backpressure method with a negative maximum value."""
|
|
# Call the backpressure method with a negative maximum value
|
|
self.service.backpressure(current_availability=50, maximum=-1)
|
|
|
|
# Check that update_backpressure_value was called with the correct values
|
|
self.service.backpressure_handler.update_backpressure_value.assert_called_once()
|
|
# Get the coroutine that was passed to run_coroutine_threadsafe
|
|
coro = self.service.backpressure_handler.update_backpressure_value.call_args[0]
|
|
self.assertEqual(coro, (50, 100)) # Should use threshold_load from config
|
|
|
|
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()
|