feat/#58 - fix unit test and deployment pipeline

.forgejo/workflows/build-and-publish.yml

@@ -22,7 +22,7 @@ env:
 
 jobs:
   build-and-push:
-    runs-on: ubuntu-latest
+    runs-on: general
     services:
       dind:
         image: docker:dind

amqp/service/amq_message_handler.py

@@ -10,6 +10,7 @@ from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
 from opentelemetry import trace
 from opentelemetry.trace import Tracer, TraceState
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+from pika.exchange_type import ExchangeType
 
 from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
@@ -282,23 +283,23 @@ class DataMessageHandler:
             future.set_result(response.body())
 
         await message.ack(multiple=False)
-        
+
     async def send_rpc(self, route, message: DataMessage) -> Future:
         """
         Sends an RPC message to the specified route and returns a Future that will be completed
         when the response is received.
-        
+
         :param route: AMQRoute object containing exchange and routing information
         :param message: DataMessage to be sent
         :return: Future that will be completed with the response payload
         """
         # Create a Future to be completed when the response is received
         future = asyncio.Future()
-        
+
         # Store the future in the outstanding dictionary using the message ID as the key
         message_id = message.id()
         self.outstanding[message_id] = future
-        
+
         # Create a Pika message with the correlation ID set to the message ID
         pika_message = Message(
             body=DataMessageFactory.serialize(message),
@@ -310,22 +311,20 @@ class DataMessageHandler:
                 else message.content_type()
             ),
         )
-        
+
         # Get the exchange from the route
-        exchange_name = route.exchange
+        rpc_exchange_name = route.exchange
-        exchange = await self.rabbit_mq_client.connection.get_exchange(exchange_name)
+        exchange: AbstractRobustExchange = await self.rabbit_mq_client.channel.declare_exchange(
-        
+            name=rpc_exchange_name, type=ExchangeType.topic
-        # Publish the message to the exchange with the component name as the routing key
-        await exchange.publish(
-            message=pika_message,
-            routing_key=route.component_name
         )
-        
+        # Publish the message to the exchange with the component name as the routing key
+        await exchange.publish(message=pika_message, routing_key=route.component_name)
+
         logging_debug(
             "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s",
-            exchange_name,
+            rpc_exchange_name,
             route.component_name,
-            message_id
+            message_id,
         )
-        
+
         return future