From 3e0f769837ac8db3d1f36009979e3a1e1f73b549 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Mon, 19 May 2025 15:27:37 +0100 Subject: [PATCH] feat #16 - update version so the code can be published as artifact --- pyproject.toml | 5 +- tests/service/test_amq_message_handler.py | 216 ---------------------- 2 files changed, 4 insertions(+), 217 deletions(-) delete mode 100644 tests/service/test_amq_message_handler.py diff --git a/pyproject.toml b/pyproject.toml index 60fc23a..36d8b7f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.31" +version = "0.2.32" description = "The Python implementation of the AMQ Adapter for CleverMicro" readme = "README.md" authors = [ @@ -39,6 +39,9 @@ include = ["amqp*"] [tool.black] line-length = 100 +[tool.pytest.ini_options] +asyncio_default_fixture_loop_scope = "module" + [project.optional-dependencies] dev = [ "pytest>=7.0", diff --git a/tests/service/test_amq_message_handler.py b/tests/service/test_amq_message_handler.py deleted file mode 100644 index e0312b1..0000000 --- a/tests/service/test_amq_message_handler.py +++ /dev/null @@ -1,216 +0,0 @@ -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()