1"""This test checks for correct fork() behavior.
2"""
3
4import imp
5import os
6import signal
7import sys
8import time
9
10from test.fork_wait import ForkWait
11from test.test_support import run_unittest, reap_children, get_attribute, import_module
12threading = import_module('threading')
13
14#Skip test if fork does not exist.
15get_attribute(os, 'fork')
16
17
18class ForkTest(ForkWait):
19    def wait_impl(self, cpid):
20        for i in range(10):
21            # waitpid() shouldn't hang, but some of the buildbots seem to hang
22            # in the forking tests.  This is an attempt to fix the problem.
23            spid, status = os.waitpid(cpid, os.WNOHANG)
24            if spid == cpid:
25                break
26            time.sleep(1.0)
27
28        self.assertEqual(spid, cpid)
29        self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
30
31    def test_import_lock_fork(self):
32        import_started = threading.Event()
33        fake_module_name = "fake test module"
34        partial_module = "partial"
35        complete_module = "complete"
36        def importer():
37            imp.acquire_lock()
38            sys.modules[fake_module_name] = partial_module
39            import_started.set()
40            time.sleep(0.01) # Give the other thread time to try and acquire.
41            sys.modules[fake_module_name] = complete_module
42            imp.release_lock()
43        t = threading.Thread(target=importer)
44        t.start()
45        import_started.wait()
46        pid = os.fork()
47        try:
48            if not pid:
49                m = __import__(fake_module_name)
50                if m == complete_module:
51                    os._exit(0)
52                else:
53                    os._exit(1)
54            else:
55                t.join()
56                # Exitcode 1 means the child got a partial module (bad.) No
57                # exitcode (but a hang, which manifests as 'got pid 0')
58                # means the child deadlocked (also bad.)
59                self.wait_impl(pid)
60        finally:
61            try:
62                os.kill(pid, signal.SIGKILL)
63            except OSError:
64                pass
65
66def test_main():
67    run_unittest(ForkTest)
68    reap_children()
69
70if __name__ == "__main__":
71    test_main()
72