refactoring tests

Author: DanielePalaia

tests/test_consumer.py

@@ -1,9 +1,5 @@
 from rabbitmq_amqp_python_client import (
-    BindingSpecification,
     Connection,
-    ExchangeSpecification,
-    ExchangeType,
-    Message,
     QuorumQueueSpecification,
     queue_address,
 )
@@ -17,7 +13,12 @@
     MyMessageHandlerRequeue,
     MyMessageHandlerRequeueWithAnnotations,
 )
-from .utils import create_connection
+from .utils import (
+    cleanup_dead_lettering,
+    create_connection,
+    publish_messages,
+    setup_dead_lettering,
+)
 
 
 def test_consumer_sync_queue_accept(connection: Connection) -> None:
@@ -29,17 +30,12 @@ def test_consumer_sync_queue_accept(connection: Connection) -> None:
     management.declare_queue(QuorumQueueSpecification(name=queue_name))
 
     addr_queue = queue_address(queue_name)
-
-    publisher = connection.publisher("/queues/" + queue_name)
     consumer = connection.consumer(addr_queue)
 
     consumed = 0
 
     # publish messages_to_send messages
-    for i in range(messages_to_send):
-        publisher.publish(Message(body="test" + str(i)))
-
-    publisher.close()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # consumer synchronously without handler
     for i in range(messages_to_send):
@@ -67,12 +63,7 @@ def test_consumer_async_queue_accept(connection: Connection) -> None:
 
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated
@@ -108,12 +99,7 @@ def test_consumer_async_queue_no_ack(connection: Connection) -> None:
 
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated
@@ -141,46 +127,16 @@ def test_consumer_async_queue_no_ack(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()
 
     # 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,
-        )
-    )
-
+    bind_path = setup_dead_lettering(management)
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated
@@ -204,9 +160,7 @@ 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)
+    cleanup_dead_lettering(management, bind_path)
 
     management.close()
 
@@ -220,47 +174,16 @@ def test_consumer_async_queue_with_discard_with_annotations(
 ) -> 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,
-        )
-    )
-
+    bind_path = setup_dead_lettering(management)
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated
@@ -291,9 +214,7 @@ def test_consumer_async_queue_with_discard_with_annotations(
 
     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)
+    cleanup_dead_lettering(management, bind_path)
 
     management.close()
 
@@ -313,12 +234,7 @@ def test_consumer_async_queue_with_requeue(connection: Connection) -> None:
 
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated
@@ -357,12 +273,7 @@ def test_consumer_async_queue_with_requeue_with_annotations(
 
     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()
+    publish_messages(connection, messages_to_send, queue_name)
 
     # workaround: it looks like when the consumer finish to consume invalidate the connection
     # so for the moment we need to use one dedicated

tests/utils.py

@@ -1,8 +1,60 @@
-from rabbitmq_amqp_python_client import Connection
+from rabbitmq_amqp_python_client import (
+    BindingSpecification,
+    Connection,
+    ExchangeSpecification,
+    ExchangeType,
+    Management,
+    Message,
+    QuorumQueueSpecification,
+)
 
 
 def create_connection() -> Connection:
     connection_consumer = Connection("amqp://guest:guest@localhost:5672/")
     connection_consumer.dial()
 
     return connection_consumer
+
+
+def publish_messages(connection: Connection, messages_to_send: int, queue_name) -> None:
+    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()
+
+
+def setup_dead_lettering(management: Management) -> str:
+
+    exchange_dead_lettering = "exchange-dead-letter"
+    queue_dead_lettering = "queue-dead-letter"
+    binding_key = "key_dead_letter"
+
+    # 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,
+        )
+    )
+
+    return bind_path
+
+
+def cleanup_dead_lettering(management: Management, bind_path: str) -> None:
+
+    exchange_dead_lettering = "exchange-dead-letter"
+    queue_dead_lettering = "queue-dead-letter"
+
+    management.unbind(bind_path)
+    management.delete_exchange(exchange_dead_lettering)
+    management.delete_queue(queue_dead_lettering)