Added "os.wait" tests from 3.13.5 (#6027)

ShaharNaveh · web-flow · commit 24f2524e6e93 · 2025-07-25T10:49:05.000+09:00
* Added "wait" tests from 3.13.5

* Trigger CI
diff --git a/Lib/test/fork_wait.py b/Lib/test/fork_wait.py
@@ -0,0 +1,80 @@
+"""This test case provides support for checking forking and wait behavior.
+
+To test different wait behavior, override the wait_impl method.
+
+We want fork1() semantics -- only the forking thread survives in the
+child after a fork().
+
+On some systems (e.g. Solaris without posix threads) we find that all
+active threads survive in the child after a fork(); this is an error.
+"""
+
+import os, time, unittest
+import threading
+from test import support
+from test.support import threading_helper
+import warnings
+
+
+LONGSLEEP = 2
+SHORTSLEEP = 0.5
+NUM_THREADS = 4
+
+class ForkWait(unittest.TestCase):
+
+    def setUp(self):
+        self._threading_key = threading_helper.threading_setup()
+        self.alive = {}
+        self.stop = 0
+        self.threads = []
+
+    def tearDown(self):
+        # Stop threads
+        self.stop = 1
+        for thread in self.threads:
+            thread.join()
+        thread = None
+        self.threads.clear()
+        threading_helper.threading_cleanup(*self._threading_key)
+
+    def f(self, id):
+        while not self.stop:
+            self.alive[id] = os.getpid()
+            try:
+                time.sleep(SHORTSLEEP)
+            except OSError:
+                pass
+
+    def wait_impl(self, cpid, *, exitcode):
+        support.wait_process(cpid, exitcode=exitcode)
+
+    def test_wait(self):
+        for i in range(NUM_THREADS):
+            thread = threading.Thread(target=self.f, args=(i,))
+            thread.start()
+            self.threads.append(thread)
+
+        # busy-loop to wait for threads
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            if len(self.alive) >= NUM_THREADS:
+                break
+
+        a = sorted(self.alive.keys())
+        self.assertEqual(a, list(range(NUM_THREADS)))
+
+        prefork_lives = self.alive.copy()
+
+        # Ignore the warning about fork with threads.
+        with warnings.catch_warnings(category=DeprecationWarning,
+                                     action="ignore"):
+            if (cpid := os.fork()) == 0:
+                # Child
+                time.sleep(LONGSLEEP)
+                n = 0
+                for key in self.alive:
+                    if self.alive[key] != prefork_lives[key]:
+                        n += 1
+                os._exit(n)
+            else:
+                # Parent
+                self.wait_impl(cpid, exitcode=0)
diff --git a/Lib/test/test_wait3.py b/Lib/test/test_wait3.py
@@ -0,0 +1,53 @@
+"""This test checks for correct wait3() behavior.
+"""
+
+import os
+import subprocess
+import sys
+import unittest
+from test.fork_wait import ForkWait
+from test import support
+
+if not support.has_fork_support:
+    raise unittest.SkipTest("requires working os.fork()")
+
+if not hasattr(os, 'wait3'):
+    raise unittest.SkipTest("os.wait3 not defined")
+
+class Wait3Test(ForkWait):
+    def wait_impl(self, cpid, *, exitcode):
+        # This many iterations can be required, since some previously run
+        # tests (e.g. test_ctypes) could have spawned a lot of children
+        # very quickly.
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            # wait3() shouldn't hang, but some of the buildbots seem to hang
+            # in the forking tests.  This is an attempt to fix the problem.
+            spid, status, rusage = os.wait3(os.WNOHANG)
+            if spid == cpid:
+                break
+
+        self.assertEqual(spid, cpid)
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
+        self.assertTrue(rusage)
+
+    def test_wait3_rusage_initialized(self):
+        # Ensure a successful wait3() call where no child was ready to report
+        # its exit status does not return uninitialized memory in the rusage
+        # structure. See bpo-36279.
+        args = [sys.executable, '-c', 'import sys; sys.stdin.read()']
+        proc = subprocess.Popen(args, stdin=subprocess.PIPE)
+        try:
+            pid, status, rusage = os.wait3(os.WNOHANG)
+            self.assertEqual(0, pid)
+            self.assertEqual(0, status)
+            self.assertEqual(0, sum(rusage))
+        finally:
+            proc.stdin.close()
+            proc.wait()
+
+
+def tearDownModule():
+    support.reap_children()
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_wait4.py b/Lib/test/test_wait4.py
@@ -0,0 +1,38 @@
+"""This test checks for correct wait4() behavior.
+"""
+
+import os
+import sys
+import unittest
+from test.fork_wait import ForkWait
+from test import support
+
+# If either of these do not exist, skip this test.
+if not support.has_fork_support:
+    raise unittest.SkipTest("requires working os.fork()")
+
+support.get_attribute(os, 'wait4')
+
+
+class Wait4Test(ForkWait):
+    def wait_impl(self, cpid, *, exitcode):
+        option = os.WNOHANG
+        if sys.platform.startswith('aix'):
+            # Issue #11185: wait4 is broken on AIX and will always return 0
+            # with WNOHANG.
+            option = 0
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            # wait4() shouldn't hang, but some of the buildbots seem to hang
+            # in the forking tests.  This is an attempt to fix the problem.
+            spid, status, rusage = os.wait4(cpid, option)
+            if spid == cpid:
+                break
+        self.assertEqual(spid, cpid)
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
+        self.assertTrue(rusage)
+
+def tearDownModule():
+    support.reap_children()
+
+if __name__ == "__main__":
+    unittest.main()
