pytest automation for testing all queue managers

pyinfra/pyampq/queue_manager/pika_queue_manager.py

@@ -27,7 +27,6 @@ def monkey_patch_queue_handle(channel, queue) -> QueueHandle:
 
         while True:
             message = channel.basic_get(queue)
-            print(message)
 
             if is_empty_message(message):
                 break

pyinfra/test/unit_tests/conftest.py

@@ -6,12 +6,15 @@ import pytest
 
 from pyinfra.exceptions import UnknownClient
 from pyinfra.locations import TEST_DIR
+from pyinfra.pyampq.queue_manager.pika_queue_manager import PikaQueueManager
+from pyinfra.pyampq.queue_manager.queue_manager import QueueManager
 from pyinfra.storage.adapters.azure import AzureStorageAdapter
 from pyinfra.storage.adapters.s3 import S3StorageAdapter
 from pyinfra.storage.clients.azure import get_azure_client
 from pyinfra.storage.clients.s3 import get_s3_client
 from pyinfra.storage.storage import Storage
 from pyinfra.test.config import CONFIG
+from pyinfra.test.queue_manager_mock import QueueManagerMock
 from pyinfra.test.storage.adapter_mock import StorageAdapterMock
 from pyinfra.test.storage.client_mock import StorageClientMock
 
@@ -71,3 +74,15 @@ def get_adapter(client_name, s3_backend):
         return S3StorageAdapter(get_s3_client(CONFIG.storage[s3_backend]))
     else:
         raise UnknownClient(client_name)
+
+
+def get_queue_manager(queue_manager_name) -> QueueManager:
+    if queue_manager_name == "mock":
+        return QueueManagerMock("input", "output")
+    if queue_manager_name == "pika":
+        return PikaQueueManager("input", "output")
+
+
+@pytest.fixture(scope="session")
+def queue_manager(queue_manager_name) -> QueueManager:
+    return get_queue_manager(queue_manager_name)

pyinfra/test/unit_tests/consumer_test.py

@@ -1,9 +1,6 @@
 import pytest
 
 from pyinfra.pyampq.consumer import Consumer
-from pyinfra.pyampq.queue_manager.pika_queue_manager import PikaQueueManager
-from pyinfra.pyampq.queue_manager.queue_manager import QueueManager
-from pyinfra.test.queue_manager_mock import QueueManagerMock
 
 
 @pytest.fixture
@@ -11,36 +8,34 @@ def callback():
     return lambda x: x**2
 
 
-@pytest.fixture(scope="session")
-def queue_manager() -> QueueManager:
-    return QueueManagerMock("input", "output")
-    return PikaQueueManager("input", "output")
-
-
 @pytest.fixture
 def consumer(queue_manager, callback):
     return Consumer(callback, queue_manager)
 
 
-def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(consumer, queue_manager):
+@pytest.mark.parametrize("queue_manager_name", ["mock", "pika"], scope="session")
-    queue_manager.clear()
+class TestConsumer:
-    consumer.consume()
-    assert queue_manager.output_queue.empty()
 
+    def test_consuming_empty_input_queue_does_not_put_anything_on_output_queue(self, consumer, queue_manager):
+        queue_manager.clear()
+        consumer.consume()
+        assert queue_manager.output_queue.empty()
 
-def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(consumer, queue_manager, callback):
+    def test_consuming_nonempty_input_queue_puts_messages_on_output_queue_in_fifo_order(
+            self, consumer, queue_manager, callback
+    ):
 
-    def produce_items():
+        def produce_items():
-        return range(3)
+            return range(3)
 
-    queue_manager.clear()
+        queue_manager.clear()
 
-    for item in produce_items():
+        for item in produce_items():
-        queue_manager.publish_request(item)
+            queue_manager.publish_request(item)
 
-    requests = consumer.consume()
+        requests = consumer.consume()
 
-    for _, r in zip(produce_items(), requests):
+        for _, r in zip(produce_items(), requests):
-        queue_manager.publish_response(r, callback)
+            queue_manager.publish_response(r, callback)
 
-    assert queue_manager.output_queue.to_list() == [0, 1, 4]
+        assert queue_manager.output_queue.to_list() == [0, 1, 4]