diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index 3494880..0e161ee 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -17,7 +17,7 @@ from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import DataMessage from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient from amqp.service.amq_message_handler import DataMessageHandler -from amqp.service.tracing import initialize_trace +from amqp.service.tracing import TracingConfig, initialize_telemetry logging.basicConfig(level=logging.DEBUG) logging.info(f"CWD = {os.getcwd()}") @@ -46,7 +46,12 @@ class AMQService: self.amq_configuration.amq_adapter.generator_id ) self.rabbit_mq_client = RabbitMQClient(self.amq_configuration) - self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name) + self.tracer = initialize_telemetry( + tracing_config=TracingConfig( + service_name=amq_configuration.amq_adapter.service_name, + service_version=os.getenv("SERVICE_VERSION", "unknown"), + ) + ) self.amq_data_message_handler = None self.service_message_factory = ServiceMessageFactory( self.amq_configuration.amq_adapter.service_name diff --git a/tests/router/test_router_consumer.py b/tests/router/test_router_consumer.py new file mode 100644 index 0000000..00290fd --- /dev/null +++ b/tests/router/test_router_consumer.py @@ -0,0 +1,210 @@ +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from aio_pika.abc import ( + AbstractIncomingMessage, + AbstractRobustChannel, + AbstractRobustQueue, +) + +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import AMQRoute, ServiceMessage +from amqp.model.service_message_type import ServiceMessageType +from amqp.router.router_base import ( + CM_C_REQ_QUEUE, + CM_REQUEST_EXCHANGE, +) +from amqp.router.router_consumer import RouterConsumer + + +class TestRouterConsumer(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # Create a mock for amq_adapter + self.amq_adapter = MagicMock() + self.amq_adapter.adapter_prefix.return_value = "test_" + self.amq_adapter.service_name = "test_service" + + # Create the config mock with amq_adapter + self.config = MagicMock(spec=AMQConfiguration) + self.config.amq_adapter = self.amq_adapter + self.consumer = RouterConsumer(self.config) + + # Mock channel + self.channel = AsyncMock(spec=AbstractRobustChannel) + self.consumer.channel = self.channel + + # Mock queue + self.queue = AsyncMock(spec=AbstractRobustQueue) + self.channel.declare_queue.return_value = self.queue + + def test_init(self): + """Test initialization of RouterConsumer""" + self.assertIsNotNone(self.consumer) + self.assertEqual(self.consumer.amq_configuration, self.config) + + @patch("amqp.router.router_consumer.logging_info") + async def test_init_routing_paths_consumer_success(self, mock_logging): + """Test successful initialization of routing paths consumer""" + self.consumer.is_open = MagicMock(return_value=True) + + await self.consumer.init_routing_paths_consumer(self.channel) + + # Verify queue declaration + self.channel.declare_queue.assert_called_once_with( + name="test_" + CM_C_REQ_QUEUE, durable=True, exclusive=False, auto_delete=False + ) + + # Verify queue binding + self.queue.bind.assert_called_once_with(exchange=CM_REQUEST_EXCHANGE, routing_key="") + + # Verify consumer setup + self.queue.consume.assert_called_once() + + @patch("amqp.router.router_consumer.logging_error") + async def test_init_routing_paths_consumer_channel_closed(self, mock_logging): + """Test initialization when channel is closed""" + self.consumer.is_open = MagicMock(return_value=False) + + await self.consumer.init_routing_paths_consumer(self.channel) + + mock_logging.assert_called_once() + self.channel.declare_queue.assert_not_called() + + @patch("amqp.router.router_consumer.logging_debug") + @patch("amqp.router.router_consumer.logging_info") + async def test_handle_routing_data_request_message_valid( + self, mock_logging_info, mock_logging_debug + ): + """Test handling of valid routing data request message""" + # Setup + message = AsyncMock(spec=AbstractIncomingMessage) + message.delivery_tag = 1 + message.consumer_tag = "test_consumer" + message.properties = MagicMock() + message.body = b"test_body" + + # Mock service message factory + mock_service_message = ServiceMessage( + id=None, + payload_type=0, + message_type=ServiceMessageType.ROUTING_DATA_REQUEST, + recipient_name="test_service", + reply_to="other_service", + trace_info={}, + message="test_payload", + ) + self.consumer.service_message_factory = MagicMock() + self.consumer.service_message_factory.from_bytes = MagicMock( + return_value=mock_service_message + ) + self.consumer.service_message_factory.of = MagicMock() + self.consumer.wrap_own_routes = MagicMock(return_value="test_routes") + self.consumer.publish_service_message = AsyncMock(return_value=True) + + # Execute + await self.consumer.handle_routing_data_request_message(message) + + # Verify + message.ack.assert_called_once_with(False) + self.consumer.publish_service_message.assert_called_once() + + @patch("amqp.router.router_consumer.logging_info") + async def test_register_route_success(self, mock_logging): + """Test successful route registration""" + # Setup + route = AMQRoute( + component_name="test_service", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=0, + ) + self.consumer.is_open = MagicMock(return_value=True) + self.consumer.publish_service_message = AsyncMock(return_value=True) + self.consumer.add_own_route = MagicMock() + + # Execute + await self.consumer.register_route(route) + + # Verify + self.consumer.publish_service_message.assert_called_once() + self.consumer.add_own_route.assert_called_once_with(route) + + @patch("amqp.router.router_consumer.logging_warning") + async def test_register_route_channel_closed(self, mock_logging): + """Test route registration when channel is closed""" + # Setup + route = AMQRoute( + component_name="test_service", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=0, + ) + self.consumer.is_open = MagicMock(return_value=False) + + # Execute + await self.consumer.register_route(route) + + # Verify + mock_logging.assert_called_once() + + @patch("amqp.router.router_consumer.logging_info") + @patch("amqp.router.router_consumer.logging_warning") + async def test_remove_route_on_cleanup(self, mock_logging_warning, mock_logging_info): + """Test route removal during cleanup""" + # Setup + route = AMQRoute( + component_name="test_service", + exchange="test_exchange", + queue="test_queue", + key="test_key", + timeout=1000, + valid_until=0, + ) + self.consumer.publish_service_message = AsyncMock(return_value=True) + self.consumer.remove_route = MagicMock(return_value=True) + + # Execute + result = await self.consumer.remove_route_on_cleanup(route) + + # Verify + self.consumer.publish_service_message.assert_called_once() + self.consumer.remove_route.assert_called_once_with(route) + self.assertTrue(result) + + @patch("amqp.router.router_consumer.logging_info") + @patch("amqp.router.router_consumer.logging_warning") + def test_cleanup(self, mock_logging_warning, mock_logging_info): + """Test cleanup process""" + # Setup + routes = { + AMQRoute( + component_name="test_service", + exchange="test_exchange1", + queue="test_queue1", + key="test_key1", + timeout=1000, + valid_until=0, + ), + AMQRoute( + component_name="test_service", + exchange="test_exchange2", + queue="test_queue2", + key="test_key2", + timeout=1000, + valid_until=0, + ), + } + self.consumer.get_routes = MagicMock(return_value=routes) + self.consumer.remove_route_on_cleanup = AsyncMock(return_value=True) + self.consumer.wrap_routes = MagicMock(return_value="test_routes") + + # Execute + self.consumer.cleanup() + + # Verify + self.assertEqual(self.consumer.remove_route_on_cleanup.call_count, 2) + mock_logging_info.assert_called() diff --git a/tests/service/test_amq_message_handler.py b/tests/service/test_amq_message_handler.py new file mode 100644 index 0000000..e0312b1 --- /dev/null +++ b/tests/service/test_amq_message_handler.py @@ -0,0 +1,216 @@ +import asyncio +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from aio_pika import Message +from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange +from opentelemetry.trace import Tracer + +from amqp.adapter.backpressure_handler import BackpressureHandler +from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter +from amqp.adapter.data_message_factory import DataMessageFactory +from amqp.adapter.data_response_factory import DataResponseFactory +from amqp.adapter.file_handler import FileHandler +from amqp.adapter.service_message_factory import ServiceMessageFactory +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import ( + DataMessage, + DataMessageMagicByte, + DataResponse, + MultipartDataMessage, +) +from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient +from amqp.service.amq_message_handler import DataMessageHandler + + +class TestDataMessageHandler(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # Create mock objects for dependencies + self.service_adapter = MagicMock(spec=CleverThisServiceAdapter) + self.tracer = MagicMock(spec=Tracer) + self.message_factory = MagicMock(spec=DataMessageFactory) + self.service_message_factory = MagicMock(spec=ServiceMessageFactory) + self.reply_to_exchange = MagicMock(spec=AbstractRobustExchange) + self.rabbit_mq_client = MagicMock(spec=RabbitMQClient) + + # Create a proper mock configuration with nested attributes + self.amq_configuration = MagicMock(spec=AMQConfiguration) + dispatch_mock = MagicMock() + dispatch_mock.rabbit_mq_url = "amqp://test" + dispatch_mock.download_dir = "/test/dir" + self.amq_configuration.dispatch = dispatch_mock + + self.loop = asyncio.new_event_loop() + self.backpressure_handler = MagicMock(spec=BackpressureHandler) + self.backpressure_handler.parallel_workers = 10 + self.backpressure_handler.current_parallel_executions = 0 + self.file_handler = MagicMock(spec=FileHandler) + + # Create the handler instance + self.handler = DataMessageHandler( + service_adapter=self.service_adapter, + tracer=self.tracer, + message_factory=self.message_factory, + service_message_factory=self.service_message_factory, + reply_to_exchange=self.reply_to_exchange, + rabbit_mq_client=self.rabbit_mq_client, + amq_configuration=self.amq_configuration, + loop=self.loop, + backpressure_handler=self.backpressure_handler, + file_handler=self.file_handler, + ) + + def tearDown(self): + self.loop.close() + + @patch("amqp.service.amq_message_handler.threading.Thread") + async def test_inbound_data_message_callback_as_thread(self, mock_thread): + # Arrange + message = MagicMock(spec=AbstractIncomingMessage) + mock_thread.return_value.start = MagicMock() + + # Act + await self.handler.inbound_data_message_callback_as_thread(message) + + # Assert + mock_thread.assert_called_once() + mock_thread.return_value.start.assert_called_once() + + async def test_reconstitute_data_message_simple(self): + # Arrange + message = MagicMock(spec=AbstractIncomingMessage) + message.headers = {"clevermicro.addressing": 0} + message.body = ( + b"\002\027A64f3b5af4d00000a4000|m=POST|d=test.domain|p=/test|h={}|t={}|b=test_body" + ) + message.delivery_tag = 123 + message.consumer_tag = "test_consumer" + expected_message = MagicMock(spec=DataMessage) + self.message_factory.from_bytes = MagicMock(return_value=expected_message) + + # Act + result = await self.handler.reconstitute_data_message(message) + + # Assert + self.assertEqual(result, expected_message) + + async def test_reconstitute_data_message_multipart(self): + # Arrange + message = MagicMock(spec=AbstractIncomingMessage) + message.headers = {"clevermicro.addressing": 1} + message.body = b'{"req-info": "test_queue"}' + message.delivery_tag = 123 + message.consumer_tag = "test_consumer" + + base_message = MagicMock(spec=DataMessage) + base_message.magic.return_value = DataMessageMagicByte.HTTP_REQUEST.value + base_message.id.return_value = MagicMock(as_string=lambda: "test_id") + base_message.method.return_value = "POST" + base_message.domain.return_value = "test.domain" + base_message.path.return_value = "/test" + base_message.headers.return_value = {} + base_message.trace_info.return_value = {} + base_message.base64body.return_value = b"test_body" + base_message.body.return_value = b"test_body" + + # Mock the static methods + with ( + patch( + "amqp.adapter.data_message_factory.DataMessageFactory.from_bytes", + return_value=base_message, + ), + patch( + "amqp.adapter.data_message_factory.DataMessageFactory.from_stream", + new_callable=AsyncMock, + return_value=base_message, + ), + ): + self.file_handler.on_message = AsyncMock(return_value={"file1": "path1"}) + self.file_handler.download_buffer = AsyncMock(return_value=(b"test_body", 1)) + + # Act + result = await self.handler.reconstitute_data_message(message) + + # Assert + self.assertIsInstance(result, MultipartDataMessage) + self.file_handler.on_message.assert_called_once() + + async def test_send_reply(self): + # Arrange + reply_to = "test_queue" + response = MagicMock(spec=DataResponse) + snowflake_id = MagicMock() + snowflake_id.as_string.return_value = "test_id" + response.id = MagicMock(return_value=snowflake_id) + response.content_type = MagicMock(return_value="application/json") + response.body = MagicMock(return_value=b"test_response") + response.base64body = MagicMock(return_value=b"test_response") + response.headers = MagicMock(return_value={"Content-Type": ["application/json"]}) + response.trace_info = MagicMock(return_value={}) + response.response_code = MagicMock(return_value="200") + response.error = MagicMock(return_value="") + response.error_cause = MagicMock(return_value="") + + self.reply_to_exchange.publish = AsyncMock() + self.reply_to_exchange.name = "test_exchange" + + # Act + await self.handler.send_reply(reply_to, response) + + # Assert + self.reply_to_exchange.publish.assert_called_once() + call_args = self.reply_to_exchange.publish.call_args[1] + self.assertEqual(call_args["routing_key"], reply_to) + self.assertIsInstance(call_args["message"], Message) + + async def test_reply_received_callback(self): + # Arrange + message = MagicMock(spec=AbstractIncomingMessage) + message.correlation_id = "test_id" + # Create a properly formatted response message body with magic byte + message.body = ( + b"\002\027A64f3b5af4d00000a4000|r=200|c=application/json|e=|f=|t={}|b=test_response" + ) + message.delivery_tag = 123 + message.headers = {} + message.properties = {} + message.ack = AsyncMock() + + future = asyncio.Future() + self.handler.outstanding["test_id"] = future + + expected_body = MagicMock() + + with patch.object(DataResponseFactory, "from_bytes") as mock_from_bytes: + mock_from_bytes.return_value = expected_body + + # Act + await self.handler.reply_received_callback(message) + + # Assert + message.ack.assert_called_once() + self.assertNotIn("test_id", self.handler.outstanding) + self.assertEqual(future.result(), expected_body.body()) + + async def test_run_http_request_magic(self): + # Arrange + message = MagicMock(spec=AbstractIncomingMessage) + message.reply_to = "test_reply_to" + message.delivery_tag = 123 + + amq_message = MagicMock(spec=DataMessage) + amq_message.magic.return_value = DataMessageMagicByte.HTTP_REQUEST.value + + self.service_adapter.on_message = AsyncMock(return_value=MagicMock(spec=DataResponse)) + self.backpressure_handler.check_overload_condition = AsyncMock() + self.handler.send_reply = AsyncMock() + self.rabbit_mq_client.connection = MagicMock() + + # Act + await self.handler.run_http_request_magic(message, amq_message, self.loop) + + # Assert + self.backpressure_handler.increase_parallel_executions.assert_called_once() + self.backpressure_handler.decrease_parallel_executions.assert_called_once() + self.service_adapter.on_message.assert_called_once() + self.handler.send_reply.assert_called_once() diff --git a/tests/service/test_amq_service.py b/tests/service/test_amq_service.py new file mode 100644 index 0000000..0c57e9c --- /dev/null +++ b/tests/service/test_amq_service.py @@ -0,0 +1,185 @@ +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock, patch + +from amqp.config.amq_configuration import AMQAdapter, AMQConfiguration +from amqp.model.model import DataMessage +from amqp.service.amq_service import AMQService + + +class TestAMQService(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # Create a mock configuration + self.mock_adapter = Mock(spec=AMQAdapter) + self.mock_adapter.generator_id = 12345 + self.mock_adapter.service_name = "test-service" + self.mock_adapter.route_mapping = {} + self.mock_adapter.swarm_service_id = "test-swarm-id" + self.mock_adapter.swarm_task_id = "test-task-id" + + self.mock_config = Mock(spec=AMQConfiguration) + self.mock_config.amq_adapter = self.mock_adapter + + # Add dispatch configuration + self.mock_dispatch = Mock() + self.mock_dispatch.amq_host = "localhost" + self.mock_dispatch.amq_port = 5672 + self.mock_dispatch.rabbit_mq_user = "guest" + self.mock_dispatch.rabbit_mq_password = "guest" + self.mock_dispatch.use_dlq = False + self.mock_config.dispatch = self.mock_dispatch + + # Add backpressure configuration + self.mock_backpressure = Mock() + self.mock_backpressure.threshold_threads = 10 + self.mock_config.backpressure = self.mock_backpressure + + # Create service instance with mocked configuration + self.service = AMQService(self.mock_config) + + # Mock the event loop + self.mock_loop = Mock() + self.service.loop = self.mock_loop + + def tearDown(self): + # Clean up any resources + if hasattr(self, "service"): + self.service.cleanup() + + @patch("amqp.service.amq_service.RabbitMQClient") + @patch("amqp.service.amq_service.DataMessageFactory") + @patch("amqp.service.amq_service.ServiceMessageFactory") + @patch("amqp.service.amq_service.initialize_telemetry") + def test_init( + self, mock_telemetry, mock_service_factory, mock_data_factory, mock_rabbit_client + ): + # Test initialization of AMQService + service = AMQService(self.mock_config) + + # Verify that all required components are initialized + self.assertIsNotNone(service.amq_configuration) + self.assertIsNotNone(service.data_message_factory) + self.assertIsNotNone(service.rabbit_mq_client) + self.assertIsNotNone(service.tracer) + self.assertIsNotNone(service.service_message_factory) + + @patch("amqp.service.amq_service.RabbitMQClient") + async def test_register_routes_valid(self, mock_rabbit_client): + # Setup mock route mapping + self.mock_adapter.route_mapping = "test-service:test-exchange:test-queue:test.#:5:0" + + # Setup mock rabbit client + mock_client = Mock() + mock_client.register_inbound_routes = AsyncMock() + mock_client.register_data_reply_callback = AsyncMock() + mock_client.init = AsyncMock() + mock_client.request_routes = AsyncMock() + mock_client.channel = Mock() + mock_client.channel.is_closed = False + self.service.rabbit_mq_client = mock_client + + # Setup mock service adapter + mock_service_adapter = Mock() + mock_service_adapter.loop = self.mock_loop + + # Setup mock message handler + mock_handler = Mock() + mock_handler.inbound_data_message_callback_no_thread = AsyncMock() + mock_handler.reply_received_callback = AsyncMock() + self.service.amq_data_message_handler = mock_handler + + # Initialize the service + await self.service.init(self.mock_loop, mock_service_adapter) + + # Call register_routes + await self.service.register_routes() + + # Verify that routes were registered twice (once in init, once in register_routes) + self.assertEqual(mock_client.register_inbound_routes.call_count, 2) + self.assertEqual(mock_client.register_data_reply_callback.call_count, 2) + + @patch("amqp.service.amq_service.RabbitMQClient") + async def test_register_routes_invalid(self, mock_rabbit_client): + # Setup invalid route mapping + self.mock_adapter.route_mapping = None + + # Setup mock rabbit client + mock_client = Mock() + mock_client.register_inbound_routes = AsyncMock() + mock_client.register_data_reply_callback = AsyncMock() + self.service.rabbit_mq_client = mock_client + + # Call register_routes + result = await self.service.register_routes() + + # Verify that no routes were registered + mock_client.register_inbound_routes.assert_not_called() + mock_client.register_data_reply_callback.assert_not_called() + self.assertIsNone(result) + + @patch("amqp.service.amq_service.RabbitMQClient") + async def test_reinitialize_if_disconnected(self, mock_rabbit_client): + # Setup mock route mapping + self.mock_adapter.route_mapping = "test-service:test-exchange:test-queue:test.#:5:0" + + # Setup mock rabbit client with closed channel + mock_client = AsyncMock() + mock_client.channel = Mock() + mock_client.channel.is_closed = True + mock_client.init = AsyncMock() + mock_client.register_inbound_routes = AsyncMock() + mock_client.register_data_reply_callback = AsyncMock() + mock_client.request_routes = AsyncMock() + + # Make the constructor return our mock client + mock_rabbit_client.return_value = mock_client + self.service.rabbit_mq_client = mock_client + + # Setup mock message handler + mock_handler = Mock() + mock_handler.inbound_data_message_callback_no_thread = AsyncMock() + mock_handler.reply_received_callback = AsyncMock() + self.service.amq_data_message_handler = mock_handler + + # Call reinitialize_if_disconnected + await self.service.reinitialize_if_disconnected() + + # Verify that client was reinitialized + mock_client.init.assert_called_once() + + def test_send_message(self): + # Create mock message and future + mock_message = Mock(spec=DataMessage) + mock_message.id.return_value = "test-id" + mock_future = MagicMock() + + # Setup mock message handler + self.service.amq_data_message_handler = Mock() + self.service.amq_data_message_handler.outstanding = {} + + # Send message + self.service.send_message(mock_message, mock_future) + + # Verify message was added to outstanding + self.assertEqual(self.service.amq_data_message_handler.outstanding["test-id"], mock_future) + + def test_remove_outstanding_future(self): + # Setup mock message handler with outstanding message + self.service.amq_data_message_handler = Mock() + self.service.amq_data_message_handler.outstanding = {"test-id": Mock()} + + # Remove future + self.service.remove_outstanding_future("test-id") + + # Verify message was removed from outstanding + self.assertNotIn("test-id", self.service.amq_data_message_handler.outstanding) + + def test_cleanup(self): + # Setup mock rabbit client + mock_client = Mock() + self.service.rabbit_mq_client = mock_client + + # Call cleanup + self.service.cleanup() + + # Verify rabbit client cleanup was called + mock_client.cleanup.assert_called_once()