feat #16 - update version so the code can be published as artifact
This commit is contained in:
Stanislav Hejny
@@ -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()
|
||||
Reference in New Issue
Block a user