adding purge test

Author: DanielePalaia

rabbitmq_amqp_python_client/management.py

@@ -105,7 +105,7 @@ def _request(
     def declare_exchange(
         self, exchange_specification: ExchangeSpecification
     ) -> ExchangeSpecification:
-        logger.debug("delete_exchange operation called")
+        logger.debug("declare_exchange operation called")
         body = {}
         body["auto_delete"] = exchange_specification.is_auto_delete
         body["durable"] = exchange_specification.is_durable
@@ -326,11 +326,13 @@ def unbind(self, binding_exchange_queue_path: str) -> None:
             ],
         )
 
-    def purge_queue(self, queue_name: str) -> None:
+    def purge_queue(self, queue_name: str) -> int:
         logger.debug("purge_queue operation called")
         path = purge_queue_address(queue_name)
 
-        self.request(
+        print("path: " + path)
+
+        response = self.request(
             None,
             path,
             CommonValues.command_delete.value,
@@ -339,6 +341,8 @@ def purge_queue(self, queue_name: str) -> None:
             ],
         )
 
+        return int(response.body["message_count"])
+
     def queue_info(self, queue_name: str) -> QueueInfo:
         logger.debug("queue_info operation called")
         path = queue_address(queue_name)

tests/test_management.py

@@ -199,6 +199,30 @@ def test_declare_classic_queue_with_args() -> None:
     management.delete_queue(queue_name)
 
 
+def test_declare_classic_queue_with_invalid_args() -> None:
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+
+    queue_name = "test-queue_with_args"
+    management = connection.management()
+    test_failure = True
+
+    queue_specification = ClassicQueueSpecification(
+        name=queue_name,
+        queue_type=QueueType.classic,
+        max_len=-5,
+    )
+
+    try:
+        management.declare_queue(queue_specification)
+    except ValidationCodeException:
+        test_failure = False
+
+    management.delete_queue(queue_name)
+
+    assert test_failure is False
+
+
 def test_declare_stream_with_args() -> None:
     connection = Connection("amqp://guest:guest@localhost:5672/")
     connection.dial()

tests/test_publisher.py

@@ -5,7 +5,7 @@
 )
 
 
-def test_bind_exchange_to_queue() -> None:
+def test_publish_exchange() -> None:
     connection = Connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
 
@@ -27,3 +27,30 @@ def test_bind_exchange_to_queue() -> None:
     publisher.close()
 
     management.delete_queue(queue_name)
+
+
+def test_publish_purge() -> None:
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+
+    queue_name = "test-queue"
+    management = connection.management()
+
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+
+    raised = False
+
+    try:
+        publisher = connection.publisher("/queues/" + queue_name)
+        publisher.publish(Message(body="test"))
+    except Exception:
+        raised = True
+
+    message_purged = management.purge_queue(queue_name)
+
+    assert raised is False
+    assert message_purged == 1
+
+    publisher.close()
+
+    management.delete_queue(queue_name)