adding requeue with annotations test

Author: DanielePalaia

rabbitmq_amqp_python_client/message_ack.py

@@ -42,8 +42,6 @@ def requeue_with_annotations(
         event: Event, annotations: Dict[str, "PythonAMQPData"]
     ) -> None:
         dlv = event.delivery
-        if dlv.local.annotations is None:
-            print("is none")
         dlv.local.annotations = annotations
         dlv.local.failed = False
         dlv.local.undeliverable = False

tests/conftest.py

@@ -5,6 +5,7 @@
     Event,
     MessageAck,
     MessagingHandler,
+    symbol,
 )
 
 
@@ -111,3 +112,20 @@ def create_connection() -> Connection:
     connection_consumer.dial()
 
     return connection_consumer
+
+
+class MyMessageHandlerRequeueWithAnnotations(MessagingHandler):
+
+    def __init__(self):
+        super().__init__(auto_accept=False, auto_settle=False)
+        self._received = 0
+
+    def on_message(self, event: Event):
+        annotations = {}
+        annotations[symbol("x-opt-string")] = "x-test1"
+        MessageAck.requeue_with_annotations(event, annotations)
+        MessageAck.requeue_with_annotations(event, annotations)
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.connection.close()
+            raise ConsumerTestException("consumed")

tests/test_consumer.py

@@ -11,8 +11,9 @@
     MyMessageHandlerDiscard,
     MyMessageHandlerNoack,
     MyMessageHandlerRequeue,
-    create_connection,
+    MyMessageHandlerRequeueWithAnnotations,
 )
+from .utils import create_connection
 
 
 def test_consumer_sync_queue_ack(connection: Connection) -> None:
@@ -81,9 +82,9 @@ def test_consumer_async_queue_accept(connection: Connection) -> None:
     except ConsumerTestException:
         pass
 
-    message_count = management.purge_queue(queue_name)
+    consumer.close()
 
-    management.delete_queue(queue_name)
+    message_count = management.purge_queue(queue_name)
 
     management.close()
 
@@ -215,3 +216,54 @@ def test_consumer_async_queue_with_requeue(connection: Connection) -> None:
     management.close()
 
     assert message_count > 0
+
+
+def test_consumer_async_queue_with_requeue_with_annotations(
+    connection: Connection,
+) -> None:
+    messages_to_send = 1000
+
+    queue_name = "test-queue_async_requeue"
+
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    addr_queue = queue_address(queue_name)
+
+    publisher = connection.publisher("/queues/" + queue_name)
+
+    # publish messages_to_send messages
+    for i in range(messages_to_send):
+        publisher.publish(Message(body="test" + str(i)))
+    publisher.close()
+
+    # workaround: it looks like when the consumer finish to consume invalidate the connection
+    # so for the moment we need to use one dedicated
+    connection_consumer = create_connection()
+
+    consumer = connection_consumer.consumer(
+        addr_queue, handler=MyMessageHandlerRequeueWithAnnotations()
+    )
+
+    try:
+        consumer.run()
+    # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    consumer.close()
+
+    # check for added annotation
+    new_consumer = connection.consumer(addr_queue)
+    message = new_consumer.consume()
+    new_consumer.close()
+
+    assert "x-opt-string" in message.annotations
+
+    message_count = management.purge_queue(queue_name)
+
+    management.delete_queue(queue_name)
+    management.close()
+
+    assert message_count > 0

tests/utils.py

@@ -0,0 +1,8 @@
+from rabbitmq_amqp_python_client import Connection
+
+
+def create_connection() -> Connection:
+    connection_consumer = Connection("amqp://guest:guest@localhost:5672/")
+    connection_consumer.dial()
+
+    return connection_consumer