test(General): finished unit test, coverage 88%

ISSUES CLOSED: clevermicro/amq-adapter-python#16

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

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()

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()

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()