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):
@@ -59,20 +55,15 @@ def test_consumer_async_queue_accept(connection: Connection) -> None:
 
     messages_to_send = 1000
 
-    queue_name = "test-queue_async_accept"
+    queue_name = "test-queue-async-accept"
 
     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()
+    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
@@ -91,6 +82,8 @@ def test_consumer_async_queue_accept(connection: Connection) -> None:
 
     message_count = management.purge_queue(queue_name)
 
+    management.delete_queue(queue_name)
+
     management.close()
 
     assert message_count == 0
@@ -100,20 +93,15 @@ def test_consumer_async_queue_no_ack(connection: Connection) -> None:
 
     messages_to_send = 1000
 
-    queue_name = "test-queue_async_no_ack"
+    queue_name = "test-queue-async-no-ack"
 
     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()
+    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,29 +129,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"
+    queue_name = "test-queue-async-discard"
+    exchange_dead_lettering = "exchange-dead-letter"
+    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,
-        )
-    )
+    bind_path = setup_dead_lettering(management)
+    addr_queue = queue_address(queue_name)
 
     management.declare_queue(
         QuorumQueueSpecification(
@@ -173,14 +148,7 @@ def test_consumer_async_queue_with_discard(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
@@ -204,9 +172,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,30 +186,13 @@ 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"
+    queue_name = "test-queue-async-discard"
+    exchange_dead_lettering = "exchange-dead-letter"
+    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,
@@ -252,16 +201,12 @@ def test_consumer_async_queue_with_discard_with_annotations(
         )
     )
 
+    publish_messages(connection, messages_to_send, queue_name)
+
+    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()
-
     # 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()
@@ -291,9 +236,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()
 
@@ -305,20 +248,15 @@ def test_consumer_async_queue_with_discard_with_annotations(
 def test_consumer_async_queue_with_requeue(connection: Connection) -> None:
     messages_to_send = 1000
 
-    queue_name = "test-queue_async_requeue"
+    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()
+    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
@@ -349,20 +287,15 @@ def test_consumer_async_queue_with_requeue_with_annotations(
 ) -> None:
     messages_to_send = 1000
 
-    queue_name = "test-queue_async_requeue"
+    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()
+    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/test_publisher.py

@@ -73,8 +73,6 @@ def test_publish_exchange(connection: Connection) -> None:
 
 def test_publish_purge(connection: Connection) -> None:
     messages_to_publish = 20
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
 
     queue_name = "test-queue"
     management = connection.management()
@@ -92,12 +90,12 @@ def test_publish_purge(connection: Connection) -> None:
 
     time.sleep(4)
 
-    message_purged = management.purge_queue(queue_name)
-
     attempt = 0
-    purged_messages = 0
-    while purged_messages != messages_to_publish:
-        purged_messages = management.purge_queue(queue_name)
+    message_purged = 0
+    # check right number of messages purged
+    # publish may delay so we loop several times till the condition is met
+    while message_purged != messages_to_publish:
+        message_purged = management.purge_queue(queue_name)
         time.sleep(1)
         attempt = attempt + 1
         if attempt > 60:

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)