import unittest from unittest.mock import AsyncMock, MagicMock, patch from amqp.model.service_message_type import ServiceMessageType from amqp.router.router_producer import RouterProducer class DummyAMQAdapter: def __init__(self): self.service_name = "test-service" self.route_mapping = "route-map" def adapter_prefix(self): return "prefix-" class DummyConfig: def __init__(self): self.amq_adapter = DummyAMQAdapter() class TestRouterProducer(unittest.IsolatedAsyncioTestCase): def setUp(self): self.config = DummyConfig() self.router = RouterProducer(self.config) def test_get_reply_to_queue_name(self): self.assertEqual(self.router.get_reply_to_queue_name(), "prefix-P-REPLY-TO") def test_get_response_queue_name(self): self.assertEqual(self.router.get_response_queue_name(), "prefix-P-RESP") def test_get_consume_queue_name(self): self.assertEqual(self.router.get_consume_queue_name(), "prefix-C-AMQ") def test_is_open_true(self): channel = MagicMock() channel.is_closed = False self.assertTrue(self.router.is_open(channel)) def test_is_open_false(self): channel = MagicMock() channel.is_closed = True self.assertFalse(self.router.is_open(channel)) self.assertFalse(self.router.is_open(None)) @patch("amqp.router.router_producer.logging_info") @patch("amqp.router.router_producer.logging_error") async def test_init_internal_routing_paths_channel_open(self, mock_error, mock_info): channel = AsyncMock() channel.is_closed = False channel.declare_exchange = AsyncMock() channel.declare_queue = AsyncMock() queue = AsyncMock() channel.declare_queue.return_value = queue queue.bind = AsyncMock() queue.consume = AsyncMock(return_value="ctag") await self.router.init_internal_routing_paths(channel) self.assertIsNotNone(self.router.cm_request_exchange) self.assertIsNotNone(self.router.cm_response_exchange) queue.bind.assert_awaited() queue.consume.assert_awaited() mock_error.assert_not_called() @patch("amqp.router.router_producer.logging_error") @patch("amqp.router.router_producer.logging_info") async def test_init_internal_routing_paths_channel_closed(self, mock_info, mock_error): channel = MagicMock() channel.is_closed = True await self.router.init_internal_routing_paths(channel) mock_error.assert_called() @patch("amqp.router.router_producer.logging_info") @patch("amqp.router.router_producer.logging_debug") async def test_handle_routing_data_message_valid(self, mock_debug, mock_info): message = AsyncMock() message.delivery_tag = 1 message.body = b"body" message.message_id = "mid" message.ack = AsyncMock() # Patch service_message_factory service_message = MagicMock() service_message.payload_type = 0 service_message.is_valid.return_value = True service_message.message_type = ServiceMessageType.ROUTING_DATA service_message.reply_to = "other-service" service_message.message = "route-string" self.router.service_message_factory.from_bytes = MagicMock(return_value=service_message) self.router.amq_configuration.amq_adapter.service_name = "test-service" self.router.add_routes = MagicMock() self.router.remove_routes = MagicMock() await self.router.handle_routing_data_message(message) message.ack.assert_awaited() self.router.add_routes.assert_called_with("route-string") self.router.remove_routes.assert_not_called() @patch("amqp.router.router_producer.logging_info") @patch("amqp.router.router_producer.logging_debug") async def test_handle_routing_data_message_remove(self, mock_debug, mock_info): message = AsyncMock() message.delivery_tag = 1 message.body = b"body" message.message_id = "mid" message.ack = AsyncMock() service_message = MagicMock() service_message.is_valid.return_value = True service_message.message_type = ServiceMessageType.ROUTING_DATA_REMOVE service_message.payload_type = 0 service_message.reply_to = "other-service" service_message.message = "route-string" self.router.service_message_factory.from_bytes = MagicMock(return_value=service_message) self.router.amq_configuration.amq_adapter.service_name = "test-service" self.router.add_routes = MagicMock() self.router.remove_routes = MagicMock() await self.router.handle_routing_data_message(message) message.ack.assert_awaited() self.router.remove_routes.assert_called_with("route-string") self.router.add_routes.assert_not_called() @patch("amqp.router.router_producer.logging_info") @patch("amqp.router.router_producer.logging_warning") @patch("amqp.router.router_producer.logging_error") async def test_request_routes_from_adapters_success(self, mock_error, mock_warning, mock_info): self.router.cm_request_exchange = AsyncMock() sm = MagicMock() self.router.service_message_factory.of = MagicMock(return_value=sm) self.router.publish_service_message = AsyncMock(return_value=True) await self.router.request_routes_from_adapters() mock_warning.assert_not_called() mock_error.assert_not_called() @patch("amqp.router.router_producer.logging_info") @patch("amqp.router.router_producer.logging_warning") @patch("amqp.router.router_producer.logging_error") async def test_request_routes_from_adapters_fail(self, mock_error, mock_warning, mock_info): self.router.cm_request_exchange = AsyncMock() sm = MagicMock() self.router.service_message_factory.of = MagicMock(return_value=sm) self.router.publish_service_message = AsyncMock(return_value=False) await self.router.request_routes_from_adapters() mock_warning.assert_called() mock_error.assert_not_called() @patch("amqp.router.router_producer.logging_info") async def test_publish_service_message_open(self, mock_info): self.router.channel = MagicMock() self.router.channel.is_closed = False self.router.service_message_factory.serialize = MagicMock(return_value=b"bin") exchange = AsyncMock() exchange.publish = AsyncMock() message = MagicMock() with patch("amqp.router.router_producer.Message") as MockMessage: MockMessage.return_value = MagicMock() result = await self.router.publish_service_message(message, exchange) exchange.publish.assert_awaited() self.assertTrue(result) async def test_publish_service_message_closed(self): self.router.channel = MagicMock() self.router.channel.is_closed = True exchange = AsyncMock() message = MagicMock() result = await self.router.publish_service_message(message, exchange) self.assertFalse(result)