adding fixtures in tests

rabbitmq_amqp_python_client/__init__.py

@@ -7,6 +7,7 @@
     BindingSpecification,
     ExchangeSpecification,
 )
+from .management import Management
 from .publisher import Publisher
 from .qpid.proton._message import Message
 from .queues import (
@@ -26,6 +27,7 @@
 
 __all__ = [
     "Connection",
+    "Management",
     "ExchangeSpecification",
     "QuorumQueueSpecification",
     "ClassicQueueSpecification",

rabbitmq_amqp_python_client/management.py

@@ -330,8 +330,6 @@ def purge_queue(self, queue_name: str) -> int:
         logger.debug("purge_queue operation called")
         path = purge_queue_address(queue_name)
 
-        print("path: " + path)
-
         response = self.request(
             None,
             path,

tests/conftest.py

@@ -0,0 +1,27 @@
+import pytest
+
+from rabbitmq_amqp_python_client import Connection
+
+
+@pytest.fixture()
+def connection(pytestconfig):
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+    try:
+        yield connection
+
+    finally:
+        connection.close()
+
+
+@pytest.fixture()
+def management(pytestconfig):
+    connection = Connection("amqp://guest:guest@localhost:5672/")
+    connection.dial()
+    try:
+        management = connection.management()
+        yield management
+
+    finally:
+        management.close()
+        connection.close()

tests/test_management.py

@@ -1,8 +1,8 @@
 from rabbitmq_amqp_python_client import (
     BindingSpecification,
     ClassicQueueSpecification,
-    Connection,
     ExchangeSpecification,
+    Management,
     QueueType,
     QuorumQueueSpecification,
     StreamSpecification,
@@ -12,12 +12,9 @@
 )
 
 
-def test_declare_delete_exchange() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_declare_delete_exchange(management: Management) -> None:
 
     exchange_name = "test-exchange"
-    management = connection.management()
 
     exchange_info = management.declare_exchange(
         ExchangeSpecification(name=exchange_name, arguments={})
@@ -27,15 +24,9 @@ def test_declare_delete_exchange() -> None:
 
     management.delete_exchange(exchange_name)
 
-    connection.close()
-
-
-def test_declare_purge_delete_queue() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
 
+def test_declare_purge_delete_queue(management: Management) -> None:
     queue_name = "my_queue"
-    management = connection.management()
 
     queue_info = management.declare_queue(QuorumQueueSpecification(name=queue_name))
 
@@ -45,17 +36,12 @@ def test_declare_purge_delete_queue() -> None:
 
     management.delete_queue(queue_name)
 
-    connection.close()
-
 
-def test_bind_exchange_to_queue() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_bind_exchange_to_queue(management: Management) -> None:
 
     exchange_name = "test-bind-exchange-to-queue-exchange"
     queue_name = "test-bind-exchange-to-queue-queue"
     routing_key = "routing-key"
-    management = connection.management()
 
     management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
 
@@ -89,12 +75,9 @@ def test_bind_exchange_to_queue() -> None:
     management.unbind(binding_exchange_queue_path)
 
 
-def test_queue_info_with_validations() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_queue_info_with_validations(management: Management) -> None:
 
     queue_name = "test_queue_info_with_validation"
-    management = connection.management()
 
     queue_specification = QuorumQueueSpecification(
         name=queue_name,
@@ -111,12 +94,9 @@ def test_queue_info_with_validations() -> None:
     assert queue_info.message_count == 0
 
 
-def test_queue_info_for_stream_with_validations() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_queue_info_for_stream_with_validations(management: Management) -> None:
 
     stream_name = "test_stream_info_with_validation"
-    management = connection.management()
 
     queue_specification = StreamSpecification(
         name=stream_name,
@@ -132,13 +112,10 @@ def test_queue_info_for_stream_with_validations() -> None:
     assert stream_info.message_count == 0
 
 
-def test_queue_precondition_fail() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_queue_precondition_fail(management: Management) -> None:
     test_failure = True
 
     queue_name = "test-queue_precondition_fail"
-    management = connection.management()
 
     queue_specification = QuorumQueueSpecification(
         name=queue_name, is_auto_delete=False
@@ -162,12 +139,9 @@ def test_queue_precondition_fail() -> None:
     assert test_failure is False
 
 
-def test_declare_classic_queue() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_declare_classic_queue(management: Management) -> None:
 
     queue_name = "test-declare_classic_queue"
-    management = connection.management()
 
     queue_specification = QuorumQueueSpecification(
         name=queue_name,
@@ -182,12 +156,9 @@ def test_declare_classic_queue() -> None:
     management.delete_queue(queue_name)
 
 
-def test_declare_classic_queue_with_args() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_declare_classic_queue_with_args(management: Management) -> None:
 
     queue_name = "test-queue_with_args"
-    management = connection.management()
 
     queue_specification = ClassicQueueSpecification(
         name=queue_name,
@@ -220,12 +191,8 @@ 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()
-
+def test_declare_classic_queue_with_invalid_args(management: Management) -> None:
     queue_name = "test-queue_with_args"
-    management = connection.management()
     test_failure = True
 
     queue_specification = ClassicQueueSpecification(
@@ -244,12 +211,8 @@ def test_declare_classic_queue_with_invalid_args() -> None:
     assert test_failure is False
 
 
-def test_declare_stream_with_args() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
-
+def test_declare_stream_with_args(management: Management) -> None:
     stream_name = "test-stream_with_args"
-    management = connection.management()
 
     stream_specification = StreamSpecification(
         name=stream_name,

tests/test_publisher.py

@@ -1,13 +1,13 @@
+import time
+
 from rabbitmq_amqp_python_client import (
     Connection,
     Message,
     QuorumQueueSpecification,
 )
 
 
-def test_publish_exchange() -> None:
-    connection = Connection("amqp://guest:guest@localhost:5672/")
-    connection.dial()
+def test_publish_exchange(connection: Connection) -> None:
 
     queue_name = "test-queue"
     management = connection.management()
@@ -29,7 +29,7 @@ def test_publish_exchange() -> None:
     management.delete_queue(queue_name)
 
 
-def test_publish_purge() -> None:
+def test_publish_purge(connection: Connection) -> None:
     connection = Connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
 
@@ -42,14 +42,18 @@ def test_publish_purge() -> None:
 
     try:
         publisher = connection.publisher("/queues/" + queue_name)
-        publisher.publish(Message(body="test"))
+        for i in range(20):
+            print("im here")
+            publisher.publish(Message(body="test"))
     except Exception:
         raised = True
 
+    time.sleep(1)
+
     message_purged = management.purge_queue(queue_name)
 
     assert raised is False
-    assert message_purged == 1
+    assert message_purged == 20
 
     publisher.close()