fix(aio-pika): Extract message & routing_key from either args or kwargs

src/instana/instrumentation/aio_pika.py

@@ -51,11 +51,15 @@ async def publish_with_instana(
             "rabbitmq", span_context=parent_context
         ) as span:
             connection = instance.channel._connection
+            message = kwargs["message"] if kwargs.get("message") else args[0]
+            routing_key = (
+                kwargs["routing_key"] if kwargs.get("routing_key") else args[1]
+            )
+
             _extract_span_attributes(
-                span, connection, "publish", kwargs["routing_key"], instance.name
+                span, connection, "publish", routing_key, instance.name
             )
 
-            message = args[0]
             tracer.inject(
                 span.context,
                 Format.HTTP_HEADERS,

tests/clients/test_aio_pika.py

@@ -30,7 +30,7 @@ def _resource(self) -> Generator[None, None, None]:
         # Ensure that allow_exit_as_root has the default value
         agent.options.allow_exit_as_root = False
 
-    async def publish_message(self) -> None:
+    async def publish_message(self, params_combination: str = "both_args") -> None:
         # Perform connection
         connection = await connect()
 
@@ -46,11 +46,22 @@ async def publish_message(self) -> None:
             exchange = await channel.declare_exchange("test.exchange")
             await queue.bind(exchange, routing_key=queue_name)
 
+            message = Message(f"Hello {queue_name}".encode())
+
+            args = ()
+            kwargs = {}
+
+            if params_combination == "both_kwargs":
+                kwargs = {"message": message, "routing_key": queue_name}
+            elif params_combination == "arg_kwarg":
+                args = (message,)
+                kwargs = {"routing_key": queue_name}
+            else:
+                # params_combination == "both_args"
+                args = (message, queue_name)
+
             # Sending the message
-            await exchange.publish(
-                Message(f"Hello {queue_name}".encode()),
-                routing_key=queue_name,
-            )
+            await exchange.publish(*args, **kwargs)
 
     async def delete_queue(self) -> None:
         connection = await connect()
@@ -75,9 +86,13 @@ async def consume_message(self, connect_method) -> None:
                         if queue.name in message.body.decode():
                             break
 
-    def test_basic_publish(self) -> None:
+    @pytest.mark.parametrize(
+        "params_combination",
+        ["both_args", "both_kwargs", "arg_kwarg"],
+    )
+    def test_basic_publish(self, params_combination) -> None:
         with tracer.start_as_current_span("test"):
-            self.loop.run_until_complete(self.publish_message())
+            self.loop.run_until_complete(self.publish_message(params_combination))
 
         spans = self.recorder.queued_spans()
         assert len(spans) == 2