More test and test_util cleanup

Author: pedro-psb

pulpcore/tests/functional/test_pubsub.py

@@ -29,28 +29,58 @@ def django_connection_reset(request):
         django_db_blocker.block()
 
 
-def test_postgres_pubsub():
-    """Testing postgres low-level implementation."""
-    from django.db import connection
-
-    state = SimpleNamespace()
-    state.got_message = False
-    with connection.cursor() as cursor:
-        assert connection.connection is cursor.connection
-        conn = cursor.connection
-        # Listen and Notify
-        conn.execute("LISTEN abc")
-        conn.add_notify_handler(lambda notification: setattr(state, "got_message", True))
-        cursor.execute("NOTIFY abc, 'foo'")
-        assert state.got_message is True
-        conn.execute("SELECT 1")
-        assert state.got_message is True
-
-        # Reset and retry
+class TestPostgresSpecifics:
+    def test_listen_notify_in_same_process(self):
+        """Testing postgres low-level implementation."""
+        from django.db import connection
+
+        state = SimpleNamespace()
         state.got_message = False
-        conn.execute("UNLISTEN abc")
-        cursor.execute("NOTIFY abc, 'foo'")
-        assert state.got_message is False
+        with connection.cursor() as cursor:
+            assert connection.connection is cursor.connection
+            conn = cursor.connection
+            # Listen and Notify
+            conn.execute("LISTEN abc")
+            conn.add_notify_handler(lambda notification: setattr(state, "got_message", True))
+            cursor.execute("NOTIFY abc, 'foo'")
+            assert state.got_message is True
+            conn.execute("SELECT 1")
+            assert state.got_message is True
+
+            # Reset and retry
+            state.got_message = False
+            conn.execute("UNLISTEN abc")
+            cursor.execute("NOTIFY abc, 'foo'")
+            assert state.got_message is False
+
+    def test_low_level_assumptions_on_multiprocess(self):
+        """Asserts that we are really testing two different connections.
+
+        From psycopg, the backend_id is:
+        "The process ID (PID) of the backend process handling this connection."
+        """
+        from django.db import connection
+
+        def host_act(host_turn, log):
+            with host_turn():  # 1
+                assert connection.connection is None
+                with connection.cursor() as cursor:
+                    cursor.execute("select 1")
+                assert connection.connection is not None
+                log.put(connection.connection.info.backend_pid)
+
+        def child_act(child_turn, log):
+            with child_turn():  # 2
+                assert connection.connection is None
+                with connection.cursor() as cursor:
+                    cursor.execute("select 1")
+                assert connection.connection is not None
+                log.put(connection.connection.info.backend_pid)
+
+        log = IpcUtil.run(host_act, child_act)
+        assert len(log) == 2
+        host_connection_pid, child_connection_pid = log
+        assert host_connection_pid != child_connection_pid
 
 
 M = pubsub.PubsubMessage
@@ -59,20 +89,34 @@ def test_postgres_pubsub():
 ]
 
 
+def unsubscribe_all(channels, subscriber):
+    for channel in channels:
+        subscriber.unsubscribe(channel)
+
+
+def subscribe_all(channels, subscriber):
+    for channel in channels:
+        subscriber.subscribe(channel)
+
+
+def publish_all(messages, publisher):
+    for channel, payload in messages:
+        publisher.publish(channel, payload=payload)
+
+
 @pytest.mark.parametrize("pubsub_backend", PUBSUB_BACKENDS)
+@pytest.mark.parametrize(
+    "payload",
+    (
+        pytest.param(None, id="none"),
+        pytest.param("", id="empty-string"),
+        pytest.param("payload", id="non-empty-string"),
+        pytest.param(123, id="int"),
+        pytest.param(datetime.now(), id="datetime"),
+        pytest.param(True, id="bool"),
+    ),
+)
 class TestPublish:
-
-    @pytest.mark.parametrize(
-        "payload",
-        (
-            pytest.param(None, id="none"),
-            pytest.param("", id="empty-string"),
-            pytest.param("payload", id="non-empty-string"),
-            pytest.param(123, id="int"),
-            pytest.param(datetime.now(), id="datetime"),
-            pytest.param(True, id="bool"),
-        ),
-    )
     def test_with_payload_as(self, pubsub_backend: pubsub.BasePubSubBackend, payload):
         pubsub_backend.publish("channel", payload=payload)
 
@@ -90,29 +134,17 @@ def test_with_payload_as(self, pubsub_backend: pubsub.BasePubSubBackend, payload
     ),
 )
 class TestNoIpcSubscribeFetch:
-    def unsubscribe_all(self, channels, subscriber):
-        for channel in channels:
-            subscriber.unsubscribe(channel)
-
-    def subscribe_all(self, channels, subscriber):
-        for channel in channels:
-            subscriber.subscribe(channel)
-
-    def publish_all(self, messages, publisher):
-        for channel, payload in messages:
-            publisher.publish(channel, payload=payload)
-
     def test_with(
         self, pubsub_backend: pubsub.BasePubSubBackend, messages: list[pubsub.PubsubMessage]
     ):
         channels = {m.channel for m in messages}
         publisher = pubsub_backend
         with pubsub_backend() as subscriber:
-            self.subscribe_all(channels, subscriber)
-            self.publish_all(messages, publisher)
+            subscribe_all(channels, subscriber)
+            publish_all(messages, publisher)
             assert subscriber.fetch() == messages
 
-            self.unsubscribe_all(channels, subscriber)
+            unsubscribe_all(channels, subscriber)
             assert subscriber.fetch() == []
 
     def test_select_readiness_with(
@@ -122,14 +154,14 @@ def test_select_readiness_with(
         CHANNELS = {m.channel for m in messages}
         publisher = pubsub_backend
         with pubsub_backend() as subscriber:
-            self.subscribe_all(CHANNELS, subscriber)
+            subscribe_all(CHANNELS, subscriber)
             assert subscriber.get_subscriptions() == CHANNELS
 
             ready, _, _ = select.select([subscriber], [], [], TIMEOUT)
             assert subscriber not in ready
             assert subscriber.fetch() == []
 
-            self.publish_all(messages, publisher)
+            publish_all(messages, publisher)
             ready, _, _ = select.select([subscriber], [], [], TIMEOUT)
             assert subscriber in ready
 
@@ -141,49 +173,19 @@ def test_select_readiness_with(
             assert subscriber not in ready
             assert subscriber.fetch() == []
 
-            self.unsubscribe_all(CHANNELS, subscriber)
-            self.publish_all(messages, publisher)
+            unsubscribe_all(CHANNELS, subscriber)
+            publish_all(messages, publisher)
             ready, _, _ = select.select([subscriber], [], [], TIMEOUT)
             assert subscriber not in ready
             assert subscriber.fetch() == []
 
 
-def test_postgres_backend_ipc():
-    """Asserts that we are really testing two different connections.
-
-    From psycopg, the backend_id is:
-    "The process ID (PID) of the backend process handling this connection."
-    """
-    from django.db import connection
-
-    def host_act(host_turn, log):
-        with host_turn():  # 1
-            assert connection.connection is None
-            with connection.cursor() as cursor:
-                cursor.execute("select 1")
-            assert connection.connection is not None
-            log.put(connection.connection.info.backend_pid)
-
-    def child_act(child_turn, log):
-        with child_turn():  # 2
-            assert connection.connection is None
-            with connection.cursor() as cursor:
-                cursor.execute("select 1")
-            assert connection.connection is not None
-            log.put(connection.connection.info.backend_pid)
-
-    log = IpcUtil.run(host_act, child_act)
-    assert len(log) == 2
-    host_connection_pid, child_connection_pid = log
-    assert host_connection_pid != child_connection_pid
-
-
 @pytest.mark.parametrize("pubsub_backend", PUBSUB_BACKENDS)
 @pytest.mark.parametrize(
     "messages",
     (
         pytest.param([M("a", "A1")], id="single-message"),
-        pytest.param([M("a", "A1")], id="test-leaking"),
+        pytest.param([M("a", "A1")], id="test-if-leaking"),
         pytest.param([M("b", "B1"), M("b", "B2")], id="two-messages-in-same-channel"),
         pytest.param(
             [M("c", "C1"), M("c", "C2"), M("d", "D1"), M("d", "D1")],
@@ -212,8 +214,7 @@ def subscriber_act(subscriber_turn, log):
             with pubsub_backend() as subscriber:
                 with subscriber_turn():  # 1
                     log.put("subscribe")
-                    for channel in CHANNELS:
-                        subscriber.subscribe(channel)
+                    subscribe_all(CHANNELS, subscriber)
 
                 with subscriber_turn():  # 3
                     log.put("fetch")
@@ -225,8 +226,7 @@ def subscriber_act(subscriber_turn, log):
                     log.put("fetch+unsubscribe")
                     assert subscriber.fetch() == messages
                     assert subscriber.fetch() == []
-                    for channel in CHANNELS:
-                        subscriber.unsubscribe(channel)
+                    unsubscribe_all(CHANNELS, subscriber)
 
                 with subscriber_turn(done=True):  # 7
                     log.put("fetch-empty")
@@ -235,20 +235,17 @@ def subscriber_act(subscriber_turn, log):
         # child
         def publisher_act(publisher_turn, log):
             publisher = pubsub_backend
-            with publisher_turn():
+            with publisher_turn():  # 2
                 log.put("publish")
-                for message in messages:  # 2
-                    publisher.publish(message.channel, payload=message.payload)
+                publish_all(messages, publisher)
 
-            with publisher_turn():
+            with publisher_turn():  # 4
                 log.put("publish")
-                for message in messages:  # 4
-                    publisher.publish(message.channel, payload=message.payload)
+                publish_all(messages, publisher)
 
-            with publisher_turn():
+            with publisher_turn():  # 6
                 log.put("publish")
-                for message in messages:  # 6
-                    publisher.publish(message.channel, payload=message.payload)
+                publish_all(messages, publisher)
 
         log = IpcUtil.run(subscriber_act, publisher_act)
         assert log == EXPECTED_LOG
@@ -270,8 +267,7 @@ def subscriber_act(subscriber_turn, log):
             with pubsub_backend() as subscriber:
                 with subscriber_turn():  # 1
                     log.put("subscribe/select-empty")
-                    for channel in CHANNELS:
-                        subscriber.subscribe(channel)
+                    subscribe_all(CHANNELS, subscriber)
                     assert subscriber.get_subscriptions() == CHANNELS
                     ready, _, _ = select.select([subscriber], [], [], TIMEOUT)
                     assert subscriber not in ready
@@ -286,8 +282,7 @@ def subscriber_act(subscriber_turn, log):
                     assert subscriber in ready
                     assert subscriber.fetch() == messages
                     assert subscriber.fetch() == []
-                    for channel in CHANNELS:
-                        subscriber.unsubscribe(channel)
+                    unsubscribe_all(CHANNELS, subscriber)
 
                 with subscriber_turn(done=True):  # 5
                     log.put("fetch/select-empty")
@@ -299,13 +294,11 @@ def publisher_act(publisher_turn, log):
             publisher = pubsub_backend
             with publisher_turn():  # 2
                 log.put("publish")
-                for message in messages:
-                    publisher.publish(message.channel, payload=message.payload)
+                publish_all(messages, publisher)
 
             with publisher_turn():  # 4
                 log.put("publish")
-                for message in messages:
-                    publisher.publish(message.channel, payload=message.payload)
+                publish_all(messages, publisher)
 
         log = IpcUtil.run(subscriber_act, publisher_act)
         assert log == EXPECTED_LOG

pulpcore/tests/functional/test_utils.py

@@ -19,7 +19,7 @@ def child_act(child_turn, log):
         with pytest.raises(Exception, match=error_msg):
             IpcUtil.run(host_act, child_act)
 
-    def test_turns_are_respected(self):
+    def test_turns_are_deterministic(self):
         RUNS = 1000
         errors = 0
 

pulpcore/tests/functional/utils.py

@@ -171,10 +171,27 @@ class ProcessErrorData(NamedTuple):
 
 
 class IpcUtil:
+    TIMEOUT_ERROR_MESSAGE = (
+        "Tip: make sure the last 'with turn()' (in execution order) "
+        "is called with 'actor_turn(done=True)', otherwise it may hang."
+    )
+    SUBPROCESS_ERROR_HEADER_TEMPLATE = "Error from sub-process (pid={pid) on test using IpcUtil"
+    TURN_WAIT_TIMEOUT = 1
 
     @staticmethod
     def run(host_act, child_act) -> list:
-        # ensures a connection from one run doesn't interfere with the other
+        """Run two processes in synchronous alternate turns.
+
+        The act are functions with the signature (act_turn, log), where act_turn is
+        a context manager where each step of the act takes place, and log is a
+        queue where each actor can put messages in using Q.put(item).
+
+        Args:
+            host_act: The function of the act that start the communication
+            child_act: The function of the act that follows host_act
+        Returns:
+            A list with the items collected via log.
+        """
         conn_1, conn_2 = mp.Pipe()
         log = mp.SimpleQueue()
         lock = mp.Lock()
@@ -208,19 +225,12 @@ def run(host_act, child_act) -> list:
     @staticmethod
     @contextmanager
     def _actor_turn(conn: Connection, starts: bool, log, lock: mp.Lock, done: bool = False):
-        TIMEOUT = 1
+        def flush_conn(conn: Connection):
+            if not conn.poll(IpcUtil.TURN_WAIT_TIMEOUT):
+                raise TimeoutError(IpcUtil.TIMEOUT_ERROR_MESSAGE)
+            conn.recv()
 
         try:
-
-            def flush_conn(conn):
-                if not conn.poll(TIMEOUT):
-                    err_msg = (
-                        "Tip: make sure the last 'with turn()' (in execution order) "
-                        "is called with 'actor_turn(done=True)', otherwise it may hang."
-                    )
-                    raise TimeoutError(err_msg)
-                conn.recv()
-
             if starts:
                 with lock:
                     conn.send("done")
@@ -234,9 +244,8 @@ def flush_conn(conn):
                     conn.send("done")
         except Exception as e:
             traceback.print_exc(file=sys.stderr)
-            err_header = f"Error from sub-process (pid={os.getpid()}) on test using IpcUtil"
-            traceback_str = f"{err_header}\n\n{traceback.format_exc()}"
-
+            error_header = IpcUtil.SUBPROCESS_ERROR_HEADER_TEMPLATE.format(pid=os.getpid())
+            traceback_str = f"{error_header}\n\n{traceback.format_exc()}"
             error = ProcessErrorData(e, traceback_str)
             log.put(error)
             exit(1)
@@ -246,4 +255,6 @@ def read_log(log: mp.SimpleQueue) -> list:
         result = []
         while not log.empty():
             result.append(log.get())
+        for item in result:
+            log.put(item)
         return result