adding discard with annotations test

Author: DanielePalaia

tests/conftest.py

@@ -101,6 +101,22 @@ def on_message(self, event: Event):
             raise ConsumerTestException("consumed")
 
 
+class MyMessageHandlerDiscardWithAnnotations(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.discard_with_annotations(event, annotations)
+        self._received = self._received + 1
+        if self._received == 1000:
+            event.connection.close()
+            raise ConsumerTestException("consumed")
+
+
 class MyMessageHandlerRequeue(MessagingHandler):
 
     def __init__(self):
@@ -125,7 +141,6 @@ 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()

tests/test_consumer.py

@@ -12,6 +12,7 @@
     ConsumerTestException,
     MyMessageHandlerAccept,
     MyMessageHandlerDiscard,
+    MyMessageHandlerDiscardWithAnnotations,
     MyMessageHandlerNoack,
     MyMessageHandlerRequeue,
     MyMessageHandlerRequeueWithAnnotations,
@@ -156,7 +157,7 @@ def test_consumer_async_queue_with_discard(connection: Connection) -> None:
         )
     )
     management.declare_queue(QuorumQueueSpecification(name=queue_dead_lettering))
-    management.bind(
+    bind_path = management.bind(
         BindingSpecification(
             source_exchange=exchange_dead_lettering,
             destination_queue=queue_dead_lettering,
@@ -203,6 +204,95 @@ def test_consumer_async_queue_with_discard(connection: Connection) -> None:
 
     message_count_dead_lettering = management.purge_queue(queue_dead_lettering)
 
+    management.unbind(bind_path)
+    management.delete_exchange(exchange_dead_lettering)
+    management.delete_queue(queue_dead_lettering)
+
+    management.close()
+
+    assert message_count == 0
+    # check dead letter queue
+    assert message_count_dead_lettering == messages_to_send
+
+
+def test_consumer_async_queue_with_discard_with_annotations(
+    connection: Connection,
+) -> None:
+    messages_to_send = 1000
+
+    exchange_dead_lettering = "exchange-dead-letter"
+    queue_dead_lettering = "queue-dead-letter"
+    queue_name = "test-queue_async_discard"
+    binding_key = "key_dead_letter"
+
+    management = connection.management()
+
+    # configuring dead lettering
+    management.declare_exchange(
+        ExchangeSpecification(
+            name=exchange_dead_lettering,
+            exchange_type=ExchangeType.fanout,
+            arguments={},
+        )
+    )
+    management.declare_queue(QuorumQueueSpecification(name=queue_dead_lettering))
+    bind_path = management.bind(
+        BindingSpecification(
+            source_exchange=exchange_dead_lettering,
+            destination_queue=queue_dead_lettering,
+            binding_key=binding_key,
+        )
+    )
+
+    management.declare_queue(
+        QuorumQueueSpecification(
+            name=queue_name,
+            dead_letter_exchange=exchange_dead_lettering,
+            dead_letter_routing_key=binding_key,
+        )
+    )
+
+    addr_queue = queue_address(queue_name)
+    addr_queue_dl = queue_address(queue_dead_lettering)
+
+    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=MyMessageHandlerDiscardWithAnnotations()
+    )
+
+    try:
+        consumer.run()
+    # ack to terminate the consumer
+    except ConsumerTestException:
+        pass
+
+    consumer.close()
+
+    # check for added annotation
+    new_consumer = connection.consumer(addr_queue_dl)
+    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)
+
+    message_count_dead_lettering = management.purge_queue(queue_dead_lettering)
+
+    management.unbind(bind_path)
+    management.delete_exchange(exchange_dead_lettering)
     management.delete_queue(queue_dead_lettering)
 
     management.close()