adding dead lettering test

Author: DanielePalaia

examples/getting_started/main.py

@@ -22,11 +22,22 @@ def __init__(self):
     def on_message(self, event: Event):
         print("received message: " + str(event.message.annotations))
 
+        # accepting
+        MessageAck.accept(event)
+
+        # in case of rejection (+eventually deadlettering)
+        # MessageAck.discard(event)
+
+        # in case of requeuing
+        # MessageAck.requeue(event)
+
         # annotations = {}
         # annotations[symbol('x-opt-string')] = 'x-test1'
+        # in case of requeuing with annotations added
         # MessageAck.requeue_with_annotations(event, annotations)
 
-        MessageAck.accept(event)
+        # in case of rejection with annotations added
+        # MessageAck.discard_with_annotations(event)
 
         print("count " + str(self._count))
 
@@ -67,9 +78,8 @@ def main() -> None:
     management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
     management.declare_queue(
-        QuorumQueueSpecification(
-            name=queue_name, dead_letter_exchange="dead-letter-test"
-        )
+        QuorumQueueSpecification(name=queue_name)
+        # QuorumQueueSpecification(name=queue_name, dead_letter_exchange="dead-letter")
     )
 
     print("binding queue to exchange")

rabbitmq_amqp_python_client/__init__.py

@@ -1,7 +1,7 @@
 from importlib import metadata
 
 from .address_helper import exchange_address, queue_address
-from .common import QueueType
+from .common import ExchangeType, QueueType
 from .connection import Connection
 from .consumer import Consumer
 from .entities import (
@@ -50,4 +50,5 @@
     "Delivery",
     "MessageAck",
     "symbol",
+    "ExchangeType",
 ]

tests/conftest.py

@@ -5,6 +5,7 @@
     Event,
     MessageAck,
     MessagingHandler,
+    queue_address,
     symbol,
 )
 
@@ -33,6 +34,21 @@ def management(pytestconfig):
         connection.close()
 
 
+@pytest.fixture()
+def consumer(pytestconfig):
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+    try:
+        queue_name = "test-queue"
+        addr_queue = queue_address(queue_name)
+        consumer = connection.consumer(addr_queue)
+        yield consumer
+
+    finally:
+        consumer.close()
+        connection.close()
+
+
 class ConsumerTestException(BaseException):
     # Constructor or Initializer
     def __init__(self, msg: str):
@@ -67,17 +83,9 @@ def on_message(self, event: Event):
         if self._received == 1000:
             event.receiver.close()
             event.connection.close()
+            # Workaround to terminate the Consumer and notify the test when all messages are consumed
             raise ConsumerTestException("consumed")
 
-    def on_connection_closed(self, event: Event):
-        print("connection closed")
-
-    def on_link_closed(self, event: Event) -> None:
-        print("link closed")
-
-    def on_rejected(self, event: Event) -> None:
-        print("rejected")
-
 
 class MyMessageHandlerDiscard(MessagingHandler):
 
@@ -107,13 +115,6 @@ def on_message(self, event: Event):
             raise ConsumerTestException("consumed")
 
 
-def create_connection() -> Connection:
-    connection_consumer = Connection("amqp://guest:guest@localhost:5672/")
-    connection_consumer.dial()
-
-    return connection_consumer
-
-
 class MyMessageHandlerRequeueWithAnnotations(MessagingHandler):
 
     def __init__(self):

tests/test_consumer.py

@@ -1,5 +1,8 @@
 from rabbitmq_amqp_python_client import (
+    BindingSpecification,
     Connection,
+    ExchangeSpecification,
+    ExchangeType,
     Message,
     QuorumQueueSpecification,
     queue_address,
@@ -16,9 +19,9 @@
 from .utils import create_connection
 
 
-def test_consumer_sync_queue_ack(connection: Connection) -> None:
+def test_consumer_sync_queue_accept(connection: Connection) -> None:
 
-    queue_name = "test-queue-ack"
+    queue_name = "test-queue-sync-accept"
     messages_to_send = 100
     management = connection.management()
 
@@ -31,21 +34,22 @@ def test_consumer_sync_queue_ack(connection: Connection) -> None:
 
     consumed = 0
 
-    # publish 10 messages
+    # publish messages_to_send messages
     for i in range(messages_to_send):
         publisher.publish(Message(body="test" + str(i)))
 
+    publisher.close()
+
     # consumer synchronously without handler
     for i in range(messages_to_send):
         message = consumer.consume()
         if message.body == "test" + str(i):
             consumed = consumed + 1
 
-    assert consumed > 0
-
-    publisher.close()
     consumer.close()
 
+    assert consumed > 0
+
     management.delete_queue(queue_name)
     management.close()
 
@@ -91,11 +95,11 @@ def test_consumer_async_queue_accept(connection: Connection) -> None:
     assert message_count == 0
 
 
-def test_consumer_async_queue_noack(connection: Connection) -> None:
+def test_consumer_async_queue_no_ack(connection: Connection) -> None:
 
     messages_to_send = 1000
 
-    queue_name = "test-queue_async_noack"
+    queue_name = "test-queue_async_no_ack"
 
     management = connection.management()
 
@@ -136,11 +140,37 @@ def test_consumer_async_queue_noack(connection: Connection) -> None:
 def test_consumer_async_queue_with_discard(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()
 
-    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+    # configuring dead lettering
+    management.declare_exchange(
+        ExchangeSpecification(
+            name=exchange_dead_lettering,
+            exchange_type=ExchangeType.fanout,
+            arguments={},
+        )
+    )
+    management.declare_queue(QuorumQueueSpecification(name=queue_dead_lettering))
+    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)
 
@@ -171,9 +201,15 @@ def test_consumer_async_queue_with_discard(connection: Connection) -> None:
 
     management.delete_queue(queue_name)
 
+    message_count_dead_lettering = management.purge_queue(queue_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_requeue(connection: Connection) -> None:

tests/test_publisher.py

@@ -1,3 +1,5 @@
+import time
+
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     Connection,
@@ -69,7 +71,6 @@ def test_publish_exchange(connection: Connection) -> None:
     management.close()
 
 
-"""
 def test_publish_purge(connection: Connection) -> None:
     messages_to_publish = 20
     connection = Connection("amqp://guest:guest@localhost:5672/")
@@ -98,7 +99,7 @@ def test_publish_purge(connection: Connection) -> None:
     while purged_messages != messages_to_publish:
         purged_messages = management.purge_queue(queue_name)
         time.sleep(1)
-        attempt = attempt+1
+        attempt = attempt + 1
         if attempt > 60:
             break
 
@@ -109,4 +110,3 @@ def test_publish_purge(connection: Connection) -> None:
 
     management.delete_queue(queue_name)
     management.close()
-"""