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):
@@ -49,30 +45,25 @@ def test_consumer_sync_queue_accept(connection: Connection) -> None:
 
     consumer.close()
 
-    assert consumed > 0
-
     management.delete_queue(queue_name)
     management.close()
 
+    assert consumed > 0
+
 
 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()
@@ -283,20 +228,18 @@ def test_consumer_async_queue_with_discard_with_annotations(
     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)
+    cleanup_dead_lettering(management, bind_path)
 
     management.close()
 
+    assert "x-opt-string" in message.annotations
+
     assert message_count == 0
     # check dead letter queue
     assert message_count_dead_lettering == messages_to_send
@@ -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
@@ -385,11 +318,11 @@ def test_consumer_async_queue_with_requeue_with_annotations(
     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 "x-opt-string" in message.annotations
+
     assert message_count > 0

tests/test_publisher.py

@@ -25,13 +25,13 @@ def test_publish_queue(connection: Connection) -> None:
     except Exception:
         raised = True
 
-    assert raised is False
-
     publisher.close()
 
     management.delete_queue(queue_name)
     management.close()
 
+    assert raised is False
+
 
 def test_publish_exchange(connection: Connection) -> None:
 
@@ -62,19 +62,17 @@ def test_publish_exchange(connection: Connection) -> None:
     except Exception:
         raised = True
 
-    assert raised is False
-
     publisher.close()
 
     management.delete_exchange(exchange_name)
     management.delete_queue(queue_name)
     management.close()
 
+    assert raised is False
+
 
 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,21 +90,21 @@ 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 == 0:
+        message_purged = management.purge_queue(queue_name)
         time.sleep(1)
         attempt = attempt + 1
         if attempt > 60:
             break
 
-    assert raised is False
-    assert message_purged == 20
-
     publisher.close()
 
     management.delete_queue(queue_name)
     management.close()
+
+    assert raised is False
+    assert message_purged == 20