Stanislav Hejny
210 lines
10 KiB
Python
210 lines
10 KiB
Python
from unittest import IsolatedAsyncioTestCase
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from amqp.model.model import DataMessage
|
|
from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
|
|
|
|
|
|
class TestRabbitMQClient(IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.mock_config = MagicMock()
|
|
self.mock_config.dispatch.amq_host = "localhost"
|
|
self.mock_config.dispatch.amq_port = 5672
|
|
self.mock_config.dispatch.rabbit_mq_user = "user"
|
|
self.mock_config.dispatch.rabbit_mq_password = "pass"
|
|
self.mock_config.dispatch.use_dlq = False
|
|
patcher = patch("amqp.router.router_consumer.RouterConsumer")
|
|
self.addCleanup(patcher.stop)
|
|
self.mock_router = patcher.start()
|
|
self.client = RabbitMQClient(self.mock_config, self.mock_router)
|
|
|
|
async def test_init_sets_up_channel_and_reply_exchange(self):
|
|
"""Test that init sets up the channel, reply queue, and reply exchange."""
|
|
loop = MagicMock()
|
|
self.client.setup_amq_channel = AsyncMock()
|
|
self.client.router.init_routing_paths_consumer = AsyncMock()
|
|
self.client.router.get_reply_to_queue_name = MagicMock(return_value="reply_queue")
|
|
self.client.channel = MagicMock()
|
|
self.client.channel.declare_queue = AsyncMock()
|
|
self.client.channel.declare_exchange = AsyncMock()
|
|
queue_mock = MagicMock()
|
|
exchange_mock = MagicMock()
|
|
self.client.channel.declare_queue.return_value = queue_mock
|
|
self.client.channel.declare_exchange.return_value = exchange_mock
|
|
queue_mock.bind = AsyncMock()
|
|
|
|
result = await self.client.init(loop)
|
|
|
|
self.client.setup_amq_channel.assert_awaited_once_with(loop)
|
|
self.client.router.init_routing_paths_consumer.assert_awaited_once()
|
|
self.client.channel.declare_queue.assert_awaited_once()
|
|
self.client.channel.declare_exchange.assert_awaited_once()
|
|
queue_mock.bind.assert_awaited_once_with(exchange=exchange_mock, routing_key="reply_queue")
|
|
self.assertEqual(result, exchange_mock)
|
|
|
|
def test_cleanup_closes_channel_and_connection(self):
|
|
"""Test that cleanup closes the channel and connection if open."""
|
|
self.client.router.cleanup = MagicMock()
|
|
self.client.router.is_open = MagicMock(return_value=True)
|
|
self.client.channel = MagicMock()
|
|
self.client.connection = MagicMock()
|
|
self.client.connection.is_closed = False
|
|
self.client.channel.close = MagicMock()
|
|
self.client.connection.close = MagicMock()
|
|
|
|
self.client.cleanup()
|
|
|
|
self.client.router.cleanup.assert_called_once()
|
|
self.client.channel.close.assert_called_once()
|
|
self.client.connection.close.assert_called_once()
|
|
|
|
def test_cleanup_handles_exceptions(self):
|
|
"""Test that cleanup handles exceptions during close."""
|
|
self.client.router.cleanup = MagicMock()
|
|
self.client.router.is_open = MagicMock(return_value=True)
|
|
self.client.channel = MagicMock()
|
|
self.client.connection = MagicMock()
|
|
self.client.connection.is_closed = False
|
|
self.client.channel.close = MagicMock(side_effect=Exception("fail"))
|
|
self.client.connection.close = MagicMock()
|
|
|
|
# Should not raise
|
|
self.client.cleanup()
|
|
|
|
async def test_register_inbound_routes_registers_routes(self):
|
|
"""Test that register_inbound_routes declares queues, exchanges, binds, and registers routes."""
|
|
self.client.router.is_open = MagicMock(return_value=True)
|
|
self.client.amq_configuration = self.mock_config
|
|
self.client.router.unwrap_route_list = MagicMock(
|
|
return_value=[
|
|
MagicMock(
|
|
queue="q1",
|
|
exchange="ex1",
|
|
key="rk1",
|
|
component_name="comp",
|
|
timeout=0,
|
|
valid_until=0,
|
|
)
|
|
]
|
|
)
|
|
self.client.router.get_consume_queue_name = MagicMock(return_value="q1")
|
|
self.client.channel = MagicMock()
|
|
self.client.channel.declare_queue = AsyncMock()
|
|
self.client.channel.declare_exchange = AsyncMock()
|
|
queue_mock = MagicMock()
|
|
exchange_mock = MagicMock()
|
|
self.client.channel.declare_queue.return_value = queue_mock
|
|
self.client.channel.declare_exchange.return_value = exchange_mock
|
|
queue_mock.bind = AsyncMock()
|
|
queue_mock.consume = AsyncMock(return_value="ctag")
|
|
self.client.router.register_route = AsyncMock()
|
|
|
|
await self.client.register_inbound_routes("route_map", MagicMock())
|
|
|
|
self.client.channel.declare_queue.assert_awaited_once()
|
|
self.client.channel.declare_exchange.assert_awaited_once()
|
|
queue_mock.bind.assert_awaited_once_with(exchange_mock, "rk1")
|
|
queue_mock.consume.assert_awaited_once()
|
|
self.client.router.register_route.assert_awaited_once()
|
|
|
|
async def test_register_inbound_routes_channel_null(self):
|
|
"""Test that register_inbound_routes logs a warning if channel is null."""
|
|
self.client.router.is_open = MagicMock(return_value=False)
|
|
with patch("amqp.rabbitmq.rabbit_mq_client.logging_warning") as mock_log:
|
|
await self.client.register_inbound_routes("route_map", MagicMock())
|
|
mock_log.assert_called()
|
|
|
|
async def test_register_data_reply_callback(self):
|
|
"""Test that register_data_reply_callback consumes on reply_to_queue."""
|
|
queue_mock = MagicMock()
|
|
queue_mock.consume = AsyncMock(return_value="ctag")
|
|
queue_mock.name = "reply_queue"
|
|
self.client.reply_to_queue = queue_mock
|
|
with patch("amqp.rabbitmq.rabbit_mq_client.logging_info") as mock_log:
|
|
await self.client.register_data_reply_callback(MagicMock())
|
|
queue_mock.consume.assert_awaited_once()
|
|
mock_log.assert_called()
|
|
|
|
async def test_setup_amq_channel_sets_channel_and_qos(self):
|
|
"""Test that setup_amq_channel sets up the connection, channel, and QoS."""
|
|
loop = MagicMock()
|
|
with patch("aio_pika.connect_robust", new_callable=AsyncMock) as mock_connect:
|
|
mock_connection = MagicMock()
|
|
mock_connect.return_value = mock_connection
|
|
mock_connection.channel = AsyncMock()
|
|
mock_channel = MagicMock()
|
|
mock_connection.channel.return_value = mock_channel
|
|
mock_channel.set_qos = AsyncMock()
|
|
self.client.router.init_internal_routing_paths = AsyncMock()
|
|
self.client.router.set_route_added_notifier = MagicMock()
|
|
|
|
await self.client.setup_amq_channel(loop)
|
|
|
|
mock_connect.assert_awaited_once()
|
|
mock_connection.channel.assert_awaited_once()
|
|
mock_channel.set_qos.assert_awaited_once_with(
|
|
prefetch_count=RabbitMQClient.PREFETCH_COUNT
|
|
)
|
|
self.client.router.init_internal_routing_paths.assert_awaited_once_with(mock_channel)
|
|
self.client.router.set_route_added_notifier.assert_called_once()
|
|
|
|
async def test_send_message_no_route(self):
|
|
"""Test that send_message logs a warning if no route is found."""
|
|
self.client.router.find_route_by_message = MagicMock(return_value=None)
|
|
self.client.rpc_exchange = MagicMock()
|
|
message = MagicMock(spec=DataMessage)
|
|
with patch("amqp.rabbitmq.rabbit_mq_client.logging_warning") as mock_log:
|
|
await self.client.send_message(message)
|
|
mock_log.assert_called()
|
|
|
|
async def test_send_message_no_reply(self):
|
|
"""Test that send_message publishes a message with no reply expected."""
|
|
route = MagicMock(timeout=0)
|
|
self.client.router.find_route_by_message = MagicMock(return_value=route)
|
|
self.client.rpc_exchange = AsyncMock()
|
|
self.client.router.get_routing_key = MagicMock(return_value="rk")
|
|
message = MagicMock(spec=DataMessage)
|
|
with patch(
|
|
"amqp.rabbitmq.rabbit_mq_client.DataMessageFactory.serialize", return_value=b"bytes"
|
|
):
|
|
with patch("amqp.rabbitmq.rabbit_mq_client.logging_info") as mock_log:
|
|
await self.client.send_message(message)
|
|
self.client.rpc_exchange.publish.assert_awaited_once()
|
|
mock_log.assert_called()
|
|
|
|
async def test_send_message_rpc_call(self):
|
|
"""Test that send_message publishes a message with reply expected (RPC)."""
|
|
route = MagicMock(timeout=1)
|
|
self.client.router.find_route_by_message = MagicMock(return_value=route)
|
|
self.client.rpc_exchange = AsyncMock()
|
|
self.client.router.get_routing_key = MagicMock(return_value="rk")
|
|
self.client.router.get_reply_to_queue_name = MagicMock(return_value="reply_q")
|
|
message = MagicMock(spec=DataMessage)
|
|
message.id.return_value = "id"
|
|
with patch(
|
|
"amqp.rabbitmq.rabbit_mq_client.DataMessageFactory.serialize", return_value=b"bytes"
|
|
):
|
|
await self.client.send_message(message)
|
|
self.client.rpc_exchange.publish.assert_awaited_once()
|
|
|
|
async def test_send_message_exception(self):
|
|
"""Test that send_message logs an error if an exception occurs."""
|
|
self.client.router.find_route_by_message = MagicMock(side_effect=Exception("fail"))
|
|
message = MagicMock(spec=DataMessage)
|
|
with patch("amqp.rabbitmq.rabbit_mq_client.logging_error") as mock_log:
|
|
await self.client.send_message(message)
|
|
mock_log.assert_called()
|
|
|
|
async def test_request_routes_delegates(self):
|
|
"""Test that request_routes delegates to router."""
|
|
self.client.router.request_routes_from_adapters = AsyncMock()
|
|
await self.client.request_routes()
|
|
self.client.router.request_routes_from_adapters.assert_awaited_once()
|
|
|
|
def test_find_route_delegates(self):
|
|
"""Test that find_route delegates to router."""
|
|
self.client.router.find_route = MagicMock(return_value="route")
|
|
result = self.client.find_route("domain", "path")
|
|
self.client.router.find_route.assert_called_once_with("domain", "path")
|
|
self.assertEqual(result, "route")
|