• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""This test checks for correct wait3() behavior.
2"""
3
4import os
5import subprocess
6import sys
7import time
8import unittest
9from test.fork_wait import ForkWait
10from test import support
11
12if not hasattr(os, 'fork'):
13    raise unittest.SkipTest("os.fork not defined")
14
15if not hasattr(os, 'wait3'):
16    raise unittest.SkipTest("os.wait3 not defined")
17
18class Wait3Test(ForkWait):
19    def wait_impl(self, cpid, *, exitcode):
20        # This many iterations can be required, since some previously run
21        # tests (e.g. test_ctypes) could have spawned a lot of children
22        # very quickly.
23        deadline = time.monotonic() + support.SHORT_TIMEOUT
24        while time.monotonic() <= deadline:
25            # wait3() shouldn't hang, but some of the buildbots seem to hang
26            # in the forking tests.  This is an attempt to fix the problem.
27            spid, status, rusage = os.wait3(os.WNOHANG)
28            if spid == cpid:
29                break
30            time.sleep(0.1)
31
32        self.assertEqual(spid, cpid)
33        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
34        self.assertTrue(rusage)
35
36    def test_wait3_rusage_initialized(self):
37        # Ensure a successful wait3() call where no child was ready to report
38        # its exit status does not return uninitialized memory in the rusage
39        # structure. See bpo-36279.
40        args = [sys.executable, '-c', 'import sys; sys.stdin.read()']
41        proc = subprocess.Popen(args, stdin=subprocess.PIPE)
42        try:
43            pid, status, rusage = os.wait3(os.WNOHANG)
44            self.assertEqual(0, pid)
45            self.assertEqual(0, status)
46            self.assertEqual(0, sum(rusage))
47        finally:
48            proc.stdin.close()
49            proc.wait()
50
51
52def tearDownModule():
53    support.reap_children()
54
55if __name__ == "__main__":
56    unittest.main()
57