[3.12] gh-108973: Fix asyncio test_subprocess_consistent_callbacks() (GH-109431) (#109609)

gh-108973: Fix asyncio test_subprocess_consistent_callbacks() (GH-109431)

SubprocessProtocol process_exited() method can be called before
pipe_data_received() and pipe_connection_lost() methods. Document it
and adapt the test for that.

Revert commit 282edd7b2a.
_child_watcher_callback() calls immediately _process_exited(): don't
add an additional delay with call_soon(). The reverted change didn't
make _process_exited() more determistic: it can still be called
before pipe_connection_lost() for example.

(cherry picked from commit ced6924630)

Co-authored-by: Victor Stinner <vstinner@python.org>
Co-authored-by: Davide Rizzo <sorcio@gmail.com>

Lib/test/test_asyncio/test_subprocess.py

@@ -753,21 +753,44 @@ class SubprocessMixin:
 
         self.loop.run_until_complete(main())
 
-    def test_subprocess_consistent_callbacks(self):
+    def test_subprocess_protocol_events(self):
+        # gh-108973: Test that all subprocess protocol methods are called.
+        # The protocol methods are not called in a determistic order.
+        # The order depends on the event loop and the operating system.
         events = []
+        fds = [1, 2]
+        expected = [
+            ('pipe_data_received', 1, b'stdout'),
+            ('pipe_data_received', 2, b'stderr'),
+            ('pipe_connection_lost', 1),
+            ('pipe_connection_lost', 2),
+            'process_exited',
+        ]
+        per_fd_expected = [
+            'pipe_data_received',
+            'pipe_connection_lost',
+        ]
+
         class MyProtocol(asyncio.SubprocessProtocol):
             def __init__(self, exit_future: asyncio.Future) -> None:
                 self.exit_future = exit_future
 
             def pipe_data_received(self, fd, data) -> None:
                 events.append(('pipe_data_received', fd, data))
+                self.exit_maybe()
 
             def pipe_connection_lost(self, fd, exc) -> None:
-                events.append('pipe_connection_lost')
+                events.append(('pipe_connection_lost', fd))
+                self.exit_maybe()
 
             def process_exited(self) -> None:
                 events.append('process_exited')
-                self.exit_future.set_result(True)
+                self.exit_maybe()
+
+            def exit_maybe(self):
+                # Only exit when we got all expected events
+                if len(events) >= len(expected):
+                    self.exit_future.set_result(True)
 
         async def main() -> None:
             loop = asyncio.get_running_loop()
@@ -777,15 +800,24 @@ class SubprocessMixin:
                                                       sys.executable, '-c', code, stdin=None)
             await exit_future
             transport.close()
-            self.assertEqual(events, [
-                ('pipe_data_received', 1, b'stdout'),
-                ('pipe_data_received', 2, b'stderr'),
-                'pipe_connection_lost',
-                'pipe_connection_lost',
-                'process_exited',
-            ])
 
-        self.loop.run_until_complete(main())
+            return events
+
+        events = self.loop.run_until_complete(main())
+
+        # First, make sure that we received all events
+        self.assertSetEqual(set(events), set(expected))
+
+        # Second, check order of pipe events per file descriptor
+        per_fd_events = {fd: [] for fd in fds}
+        for event in events:
+            if event == 'process_exited':
+                continue
+            name, fd = event[:2]
+            per_fd_events[fd].append(name)
+
+        for fd in fds:
+            self.assertEqual(per_fd_events[fd], per_fd_expected, (fd, events))
 
     def test_subprocess_communicate_stdout(self):
         # See https://github.com/python/cpython/issues/100133