From aec53f5b34c483422a184798c6b9fc6e8accf39a Mon Sep 17 00:00:00 2001 From: Rui Hu Date: Thu, 15 May 2025 17:06:05 +0800 Subject: [PATCH] test(General): implement unit test for data parser, rabbitmq client and other python files This commit bump the coverage rate to 77% --- amqp/adapter/cleverthis_service_adapter.py | 47 ++-- amqp/adapter/data_parser.py | 7 +- .../test_cleverthis_service_adapter.py | 71 +++++- tests/adapter/test_data_parser.py | 168 +++++++++++++- tests/rabbitmq/test_rabbit_mq_client.py | 209 ++++++++++++++++++ tests/router/test_router_producer.py | 164 ++++++++++++++ 6 files changed, 635 insertions(+), 31 deletions(-) create mode 100644 tests/rabbitmq/test_rabbit_mq_client.py create mode 100644 tests/router/test_router_producer.py diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index a1ae77e..f09e843 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -36,6 +36,30 @@ from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from amqp.router.utils import await_result +async def _create_reply_exchange_and_queue( + connection: AbstractRobustConnection, +) -> tuple[AbstractRobustChannel, AbstractExchange, str]: + """ + Create objects for reply. + Return (channel for reply queue, exchange for reply, reply queue name) + """ + + # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with + # correct event loop. This is needed because the RabbitMQ API is not thread-safe and needs + # to be called in the context of event loop the first RabbitMQ connection was created with. + _channel = await connection.channel() + _exchange_name: str = AMQ_REPLY_TO_EXCHANGE + _exchange = await _channel.get_exchange(name=_exchange_name, ensure=True) + _queue = await _channel.declare_queue(exclusive=False) # Declare a unique, non-exclusive queue + _queue_name = _queue.name + await _queue.bind( + exchange=_exchange_name, + routing_key=_queue_name, # Use queue name as routing key + ) + logging_debug(f"Publishing to exchange: {_exchange_name}, q: {_queue_name}") + return _channel, _exchange, _queue_name + + async def _convert_file_response( obj: FileResponse, connection: AbstractRobustConnection, @@ -47,33 +71,12 @@ async def _convert_file_response( The file content itself is transmitted through the named stream (which is implemented as temporary dedicated queue. """ - - async def _wrap_amq_api_calls( - connection: AbstractRobustConnection, - ) -> tuple[AbstractRobustChannel, AbstractExchange, str]: - # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with - # correct event loop. This is needed because the RabbitMQ API is not thread-safe and needs - # to be called in the context of event loop the first RabbitMQ connection was created with. - _channel = await connection.channel() - _exchange_name: str = AMQ_REPLY_TO_EXCHANGE - _exchange = await _channel.get_exchange(name=_exchange_name, ensure=True) - _queue = await _channel.declare_queue( - exclusive=False - ) # Declare a unique, non-exclusive queue - _queue_name = _queue.name - await _queue.bind( - exchange=_exchange_name, - routing_key=_queue_name, # Use queue name as routing key - ) - logging_debug(f"Publishing to exchange: {_exchange_name}, q: {_queue_name}") - return _channel, _exchange, _queue_name - # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached # to parent event loop, and need to execute at that event loop, not the current one # allow coroutine to execute. calling _future.result() will block the entire thread # and the coroutine will never execute _channel, _exchange, _routing_key = await await_result( - asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop) + asyncio.run_coroutine_threadsafe(_create_reply_exchange_and_queue(connection), loop) ) # publish_file also needs RabbitMQ API, ensure the loop is passed as argument (_stream_name, _file_size) = await file_handler.publish_file( diff --git a/amqp/adapter/data_parser.py b/amqp/adapter/data_parser.py index 47f23fa..96d718f 100644 --- a/amqp/adapter/data_parser.py +++ b/amqp/adapter/data_parser.py @@ -40,7 +40,7 @@ class MultipartFormDataParser: try: self.form_data = json.loads(message.body()) self.is_json = True - if self.form_data["files"] and self.form_data["form"]: + if self.form_data["files"] is not None and self.form_data["form"] is not None: self.form_data = AMQDataParser.process_form_data(self.form_data) except Exception as jsonerror: logging_error(f"Parsing JSON: {jsonerror}") @@ -55,7 +55,7 @@ class MultipartFormDataParser: try: self.form_data = json.loads(message.body()) self.is_json = True - if self.form_data["files"] and self.form_data["form"]: + if self.form_data["files"] is not None and self.form_data["form"] is not None: self.form_data = AMQDataParser.process_form_data(self.form_data) except Exception as jsonerror: logging_error(f"Parsing JSON: {jsonerror}") @@ -109,7 +109,6 @@ class AMQDataParser: for key, value in data["files"].items(): if value: # Only override if files data is non-empty result[key] = value[0] - return result @staticmethod @@ -162,7 +161,7 @@ class AMQDataParser: } if len(_form_data) == 0 and len(message.body()) > 0: _form_data = json.loads(message.body()) - if _form_data["files"] and _form_data["form"]: + if _form_data["files"] is not None and _form_data["form"] is not None: _form_data = AMQDataParser.process_form_data(_form_data) return _form_data except Exception as e: diff --git a/tests/adapter/test_cleverthis_service_adapter.py b/tests/adapter/test_cleverthis_service_adapter.py index d11ba6b..e3688c4 100644 --- a/tests/adapter/test_cleverthis_service_adapter.py +++ b/tests/adapter/test_cleverthis_service_adapter.py @@ -1,14 +1,82 @@ +import asyncio import unittest from unittest.mock import AsyncMock, MagicMock, patch +from aio_pika.abc import AbstractChannel, AbstractRobustConnection from aiohttp import ClientResponse +from aiohttp.web_fileresponse import FileResponse -from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter +from amqp.adapter.cleverthis_service_adapter import ( + AMQMessage, + CleverThisServiceAdapter, + _convert_file_response, + _create_reply_exchange_and_queue, + is_likely_json, +) +from amqp.adapter.file_handler import FileHandler from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import DataResponse +from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase): + + async def test_create_reply_exchange_and_queue_success(self): + connection_mock = AsyncMock() + channel_mock = AsyncMock() + exchange_mock = AsyncMock() + queue_mock = AsyncMock() + + connection_mock.channel.return_value = channel_mock + channel_mock.get_exchange.return_value = exchange_mock + channel_mock.declare_queue.return_value = queue_mock + queue_mock.name = "test_queue_name" + + channel, exchange, queue_name = await _create_reply_exchange_and_queue(connection_mock) + + assert channel == channel_mock + assert exchange == exchange_mock + assert queue_name == "test_queue_name" + channel_mock.get_exchange.assert_called_once_with(name=AMQ_REPLY_TO_EXCHANGE, ensure=True) + channel_mock.declare_queue.assert_called_once_with(exclusive=False) + queue_mock.bind.assert_called_once_with( + exchange=AMQ_REPLY_TO_EXCHANGE, routing_key="test_queue_name" + ) + + async def test_convert_file_response_successful(self): + connection_mock = AsyncMock(spec=AbstractRobustConnection) + channel_mock: AbstractChannel = AsyncMock(spec=AbstractChannel) + exchange_mock = AsyncMock() + queue_mock = AsyncMock() + + connection_mock.channel = AsyncMock(return_value=channel_mock) + channel_mock.get_exchange.return_value = exchange_mock + channel_mock.declare_queue.return_value = queue_mock + channel_mock.close = AsyncMock() + queue_mock.name = "test_queue_name" + + mock_file_handler = MagicMock(spec=FileHandler) + mock_file_handler.publish_file = AsyncMock(return_value=("test_stream", 12345)) + mock_file_response = MagicMock(spec=FileResponse) + mock_file_response.path = "/test/path" + mock_file_response.filename = "test_file.txt" + mock_loop = asyncio.get_event_loop() + + json_response = await _convert_file_response( + mock_file_response, connection_mock, mock_file_handler, mock_loop + ) + + assert '"filename": "test_file.txt"' in json_response + assert '"streamName": "test_stream"' in json_response + assert '"fileSize": 12345' in json_response + + def test_is_likely_json(self): + assert is_likely_json("{ }") is True + assert is_likely_json("[ ]") is True + assert is_likely_json("{]") is False + assert is_likely_json("[}") is False + assert is_likely_json("something else") is False + def setUp(self): # Mock AMQConfiguration self.mock_amq_config = AMQConfiguration("./application.properties.local") @@ -204,7 +272,6 @@ class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase): wraps=self.adapter._authorize_forward_request, ) as mock_handler: response = await self.adapter.post(mock_message, None) - print(type(response.response_code())) self.assertEqual(response.response_code(), "201") mock_handler.assert_called_once() diff --git a/tests/adapter/test_data_parser.py b/tests/adapter/test_data_parser.py index be709f5..29c00fc 100644 --- a/tests/adapter/test_data_parser.py +++ b/tests/adapter/test_data_parser.py @@ -1,3 +1,165 @@ -class DataParserTest: - def test_multipart_form_data_parser(self): - assert False +import json +import unittest +from unittest.mock import MagicMock, patch + +from amqp.adapter.data_parser import AMQDataParser, MultipartFormDataParser + + +class DummyAMQMessage: + def __init__(self, headers=None, body=b"", extra_data=None): + self._headers = headers or {"Content-Type": ["application/json"]} + self._body = body + self.extra_data = extra_data if extra_data is not None else {} + + def headers(self): + return self._headers + + def body(self): + return self._body + + +class TestAMQDataParser(unittest.TestCase): + def test_process_form_data(self): + cases = [ + ( + '{"form": {"foo": ["bar"]}, "files": {"file1": ["data"]}}', + {"foo": "bar", "file1": "data"}, + ), + ({"form": {"a": ["1"]}, "files": {"b": ["2"]}}, {"a": "1", "b": "2"}), + ({"form": {}, "files": {}}, {}), + ] + for json_input, expected in cases: + result = AMQDataParser.process_form_data(json_input) + self.assertEqual(result, expected) + + def test_parse_get_url(self): + url = "http://example.com/path?foo=1&bar=2&foo=3" + path, params = AMQDataParser.parse_get_url(url) + self.assertEqual(path, "/path") + self.assertTrue(params["foo"] == ["1", "3"] or params["foo"] == "1" or params["foo"] == "3") + self.assertEqual(params["bar"], "2") + + def test_parse_request_body_json(self): + data = {"foo": "bar"} + msg = DummyAMQMessage( + headers={"Content-Type": ["application/json"]}, body=json.dumps(data).encode() + ) + result = AMQDataParser.parse_request_body(msg) + self.assertEqual(result, data) + + def test_parse_request_body_invalid_json(self): + msg = DummyAMQMessage(headers={"Content-Type": ["application/json"]}, body=b"notjson") + with patch("amqp.adapter.data_parser.logging_error") as log_err: + with self.assertRaises(ValueError): + AMQDataParser.parse_request_body(msg) + self.assertTrue(log_err.called) + + def test_parse_request_body_urlencoded(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["application/x-www-form-urlencoded"]}, body=b"foo=bar&baz=qux" + ) + result = AMQDataParser.parse_request_body(msg) + self.assertEqual(result, {b"foo": b"bar", b"baz": b"qux"}) + + def test_parse_request_body_urlencoded_json(self): + data = {"form": {"foo": ["bar"]}, "files": {}} + msg = DummyAMQMessage( + headers={"Content-Type": ["application/x-www-form-urlencoded"]}, + body=json.dumps(data).encode(), + ) + result = AMQDataParser.parse_request_body(msg) + self.assertEqual(result, {"foo": "bar"}) + + def test_parse_request_body_invalid_urlencoded(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["application/x-www-form-urlencoded"]}, body=b"not a json" + ) + with patch("amqp.adapter.data_parser.logging_error") as log_err: + with self.assertRaises(ValueError): + AMQDataParser.parse_request_body(msg) + self.assertTrue(log_err.called) + + def test_parse_request_body_text(self): + msg = DummyAMQMessage(headers={"Content-Type": ["text/plain"]}, body=b"hello world") + result = AMQDataParser.parse_request_body(msg) + self.assertEqual(result, {"text": b"hello world"}) + + def test_parse_request_body_xml(self): + xml = b"barqux" + msg = DummyAMQMessage(headers={"Content-Type": ["application/xml"]}, body=xml) + result = AMQDataParser.parse_request_body(msg) + self.assertEqual(result, {"foo": "bar", "baz": "qux"}) + + def test_parse_request_body_invalid_xml(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["application/xml"]}, body=b"" + ) + with self.assertRaises(ValueError): + AMQDataParser.parse_request_body(msg) + + def test_parse_request_body_unsupported(self): + msg = DummyAMQMessage(headers={"Content-Type": ["application/unknown"]}, body=b"") + with self.assertRaises(ValueError): + AMQDataParser.parse_request_body(msg) + + +class TestMultipartFormDataParser(unittest.TestCase): + def test_multipart_form_data_parser_json(self): + data = {"form": {"foo": ["bar"]}, "files": {}} + msg = DummyAMQMessage( + headers={"Content-Type": ["multipart/form-data"]}, + body=json.dumps(data).encode(), + extra_data={"some": "data"}, + ) + parser = MultipartFormDataParser(msg) + self.assertTrue(parser.is_json) + self.assertEqual(parser.form_data, {"foo": "bar"}) + self.assertTrue(parser.is_valid) + + def test_multipart_form_data_parser_invalid_json(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["multipart/form-data"]}, + body=b"notjson", + extra_data={"some": "data"}, + ) + with patch("amqp.adapter.data_parser.logging_error") as log_err: + parser = MultipartFormDataParser(msg) + self.assertFalse(parser.is_valid) + self.assertTrue(log_err.called) + + def test_multipart_form_data_parser_fallback_to_json(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["multipart/form-data"]}, + body=json.dumps({"form": {"foo": ["bar"]}, "files": {}}).encode(), + ) + with patch( + "amqp.adapter.data_parser.python_multipart.parse_form", side_effect=Exception("fail") + ): + parser = MultipartFormDataParser(msg) + self.assertTrue(parser.is_json) + self.assertEqual(parser.form_data, {"foo": "bar"}) + self.assertTrue(parser.is_valid) + + def test_multipart_form_data_parser_on_field_and_file(self): + msg = DummyAMQMessage( + headers={"Content-Type": ["multipart/form-data"]}, body=b"", extra_data={} + ) + parser = MultipartFormDataParser(msg) + + class DummyField: + field_name = b"foo" + value = b"bar" + + class DummyFile: + field_name = b"file" + file_object = MagicMock() + file_object.seek = MagicMock() + size = 123 + file_name = b"filename.txt" + + parser.on_field(DummyField()) + self.assertEqual(parser.form_data["foo"], b"bar") + parser.on_file(DummyFile()) + self.assertIn("file", parser.form_data) + self.assertTrue(hasattr(parser.form_data["file"], "filename")) + self.assertEqual(parser.form_data["file"].filename, "filename.txt") diff --git a/tests/rabbitmq/test_rabbit_mq_client.py b/tests/rabbitmq/test_rabbit_mq_client.py new file mode 100644 index 0000000..65da09e --- /dev/null +++ b/tests/rabbitmq/test_rabbit_mq_client.py @@ -0,0 +1,209 @@ +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.rabbitmq.rabbit_mq_client.RouterConsumer") + self.addCleanup(patcher.stop) + self.mock_router = patcher.start() + self.client = RabbitMQClient(self.mock_config) + + 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") diff --git a/tests/router/test_router_producer.py b/tests/router/test_router_producer.py new file mode 100644 index 0000000..da1af66 --- /dev/null +++ b/tests/router/test_router_producer.py @@ -0,0 +1,164 @@ +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)