amq-adapter-python/tests/router/test_router_producer.py

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)