Add more coverage for various pollers

Author: eandersson

amqpstorm/tests/functional/test_reliability.py

@@ -1,4 +1,5 @@
 import importlib
+import select
 import sys
 import threading
 import time
@@ -21,8 +22,7 @@ class ReliabilityFunctionalTests(TestFunctionalFramework):
     @setup(new_connection=False, queue=True)
     def test_functional_open_new_connection_loop(self):
         for _ in range(25):
-            self.connection = self.connection = Connection(HOST, USERNAME,
-                                                           PASSWORD)
+            self.connection = self._make_connection()
             self.channel = self.connection.channel()
 
             # Make sure that it's a new channel.
@@ -47,7 +47,7 @@ def test_functional_open_new_connection_loop(self):
 
     @setup(new_connection=False, queue=True)
     def test_functional_open_close_connection_loop(self):
-        self.connection = Connection(HOST, USERNAME, PASSWORD, lazy=True)
+        self.connection = self._make_connection(lazy=True)
         for _ in range(25):
             self.connection.open()
             channel = self.connection.channel()
@@ -75,7 +75,7 @@ def test_functional_open_close_connection_loop(self):
 
     @setup(new_connection=True, new_channel=False, queue=True)
     def test_functional_close_gracefully_after_publish_mandatory_fails(self):
-        for index in range(3):
+        for _ in range(3):
             channel = self.connection.channel()
 
             # Try to publish 25 bad messages.
@@ -95,8 +95,7 @@ def test_functional_close_gracefully_after_publish_mandatory_fails(self):
 
     @setup(new_connection=False, queue=True)
     def test_functional_open_close_channel_loop(self):
-        self.connection = self.connection = Connection(HOST, USERNAME,
-                                                       PASSWORD)
+        self.connection = self._make_connection()
         for _ in range(25):
             channel = self.connection.channel()
 
@@ -109,19 +108,18 @@ def test_functional_open_close_channel_loop(self):
 
             channel.close()
 
-            # Verify that theChannel has been closed properly.
+            # Verify that the Channel has been closed properly.
             self.assertTrue(self.connection.is_open)
             self.assertTrue(channel.is_closed)
 
     @setup(new_connection=False, queue=True)
     def test_functional_open_multiple_channels(self):
-        self.connection = self.connection = Connection(HOST, USERNAME,
-                                                       PASSWORD, lazy=True)
+        self.connection = self._make_connection(lazy=True)
 
         for _ in range(5):
             channels = []
             self.connection.open()
-            for index in range(10):
+            for _ in range(10):
                 channel = self.connection.channel()
                 channels.append(channel)
 
@@ -132,13 +130,12 @@ def test_functional_open_multiple_channels(self):
 
     @setup(new_connection=False, queue=False)
     def test_functional_close_performance(self):
-        """Make sure closing a connection never takes longer than ~1 seconds.
+        """Make sure closing a connection never takes longer than ~1 second.
 
         :return:
         """
         for _ in range(10):
-            self.connection = self.connection = Connection(HOST, USERNAME,
-                                                           PASSWORD)
+            self.connection = self._make_connection()
             start_time = time.time()
             self.connection.close()
             self.assertLess(time.time() - start_time, 3)
@@ -151,9 +148,7 @@ def test_functional_close_after_channel_close_forced_by_server(self):
         :return:
         """
         for _ in range(10):
-            connection = Connection(
-                HOST, USERNAME, PASSWORD
-            )
+            connection = self._make_connection()
             channel = connection.channel()
             channel.confirm_deliveries()
             try:
@@ -179,6 +174,7 @@ def test_functional_uri_connection(self):
         self.assertTrue(self.connection.is_open)
 
     def test_functional_ssl_connection_without_ssl(self):
+        # Poller-agnostic: validates the SSL-absence error path.
         restore_func = sys.modules['ssl']
         try:
             sys.modules['ssl'] = None
@@ -195,9 +191,7 @@ def test_functional_ssl_connection_without_ssl(self):
 
     @setup(new_connection=False, queue=True)
     def test_functional_verify_passive_declare(self):
-
-        self.connection = self.connection = Connection(HOST, USERNAME,
-                                                       PASSWORD)
+        self.connection = self._make_connection()
         self.channel = self.connection.channel()
         self.assertEqual(int(self.channel), 1)
 
@@ -223,13 +217,22 @@ def test_functional_verify_passive_declare(self):
         self.connection.close()
 
 
+class ReliabilityFunctionalTestsSelect(ReliabilityFunctionalTests):
+    poller = 'select'
+
+
+if hasattr(select, 'poll'):
+    class ReliabilityFunctionalTestsPoll(ReliabilityFunctionalTests):
+        poller = 'poll'
+
+
 class PublishAndConsume1kTest(TestFunctionalFramework):
     messages_to_send = 1000
-    messages_consumed = 0
-    lock = threading.Lock()
 
     def configure(self):
         self.disable_logging_validation()
+        self.messages_consumed = 0
+        self.lock = threading.Lock()
 
     def publish_messages(self):
         for _ in range(self.messages_to_send):
@@ -255,12 +258,12 @@ def increment_message_count(self):
     def test_functional_publish_and_consume_1k_messages(self):
         self.channel.queue.declare(self.queue_name, durable=True)
 
-        publish_thread = threading.Thread(target=self.publish_messages, )
+        publish_thread = threading.Thread(target=self.publish_messages)
         publish_thread.daemon = True
         publish_thread.start()
 
         for _ in range(4):
-            consumer_thread = threading.Thread(target=self.consume_messages, )
+            consumer_thread = threading.Thread(target=self.consume_messages)
             consumer_thread.daemon = True
             consumer_thread.start()
 
@@ -278,6 +281,17 @@ def test_functional_publish_and_consume_1k_messages(self):
                          'test took too long')
 
 
+class PublishAndConsume1kTestSelect(PublishAndConsume1kTest):
+    """1k publish/consume run against the ``select.select`` poller."""
+    poller = 'select'
+
+
+if hasattr(select, 'poll'):
+    class PublishAndConsume1kTestPoll(PublishAndConsume1kTest):
+        """1k publish/consume run against the ``select.poll`` poller."""
+        poller = 'poll'
+
+
 class Consume1kUntilEmpty(TestFunctionalFramework):
     messages_to_send = 1000
 
@@ -309,3 +323,14 @@ def test_functional_publish_and_consume_until_empty(self):
                          'not all messages consumed')
 
         channel.close()
+
+
+class Consume1kUntilEmptySelect(Consume1kUntilEmpty):
+    """Drain-the-queue test run against the ``select.select`` poller."""
+    poller = 'select'
+
+
+if hasattr(select, 'poll'):
+    class Consume1kUntilEmptyPoll(Consume1kUntilEmpty):
+        """Drain-the-queue test run against the ``select.poll`` poller."""
+        poller = 'poll'

amqpstorm/tests/functional/utility.py

@@ -14,13 +14,20 @@
 class TestFunctionalFramework(TestFramework):
     """Extended Test Base for functional unit-tests."""
 
+    poller: str = 'select'
+
     def __init__(self, *args, **kwargs):
         self.api = ManagementApi(HTTP_URL, USERNAME, PASSWORD, timeout=1)
         self.queue_name = None
         self.exchange_name = None
         self.virtual_host_name = None
         super(TestFunctionalFramework, self).__init__(*args, **kwargs)
 
+    def _make_connection(self, **kwargs):
+        kwargs.setdefault('timeout', 1)
+        kwargs.setdefault('poller', self.poller)
+        return Connection(HOST, USERNAME, PASSWORD, **kwargs)
+
 
 def retry_function_wrapper(callable_function, retry_limit=10,
                            sleep_interval=1):
@@ -63,8 +70,7 @@ def setup_wrapper(self, *args, **kwargs):
             self.exchange_name = name
             self.virtual_host_name = name
             if new_connection:
-                self.connection = Connection(HOST, USERNAME, PASSWORD,
-                                             timeout=1)
+                self.connection = self._make_connection()
                 if new_channel:
                     self.channel = self.connection.channel()
             try: