Python code coverage for Lib/test/test_fork1.py

#	count	content
1	n/a	"""This test checks for correct fork() behavior.
2	n/a	"""
3	n/a
4	n/a	import _imp as imp
5	n/a	import os
6	n/a	import signal
7	n/a	import sys
8	n/a	import time
9	n/a	import unittest
10	n/a
11	n/a	from test.fork_wait import ForkWait
12	n/a	from test.support import (reap_children, get_attribute,
13	n/a	import_module, verbose)
14	n/a
15	n/a	threading = import_module('threading')
16	n/a
17	n/a	# Skip test if fork does not exist.
18	n/a	get_attribute(os, 'fork')
19	n/a
20	n/a	class ForkTest(ForkWait):
21	n/a	def wait_impl(self, cpid):
22	n/a	deadline = time.monotonic() + 10.0
23	n/a	while time.monotonic() <= deadline:
24	n/a	# waitpid() shouldn't hang, but some of the buildbots seem to hang
25	n/a	# in the forking tests. This is an attempt to fix the problem.
26	n/a	spid, status = os.waitpid(cpid, os.WNOHANG)
27	n/a	if spid == cpid:
28	n/a	break
29	n/a	time.sleep(0.1)
30	n/a
31	n/a	self.assertEqual(spid, cpid)
32	n/a	self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
33	n/a
34	n/a	def test_threaded_import_lock_fork(self):
35	n/a	"""Check fork() in main thread works while a subthread is doing an import"""
36	n/a	import_started = threading.Event()
37	n/a	fake_module_name = "fake test module"
38	n/a	partial_module = "partial"
39	n/a	complete_module = "complete"
40	n/a	def importer():
41	n/a	imp.acquire_lock()
42	n/a	sys.modules[fake_module_name] = partial_module
43	n/a	import_started.set()
44	n/a	time.sleep(0.01) # Give the other thread time to try and acquire.
45	n/a	sys.modules[fake_module_name] = complete_module
46	n/a	imp.release_lock()
47	n/a	t = threading.Thread(target=importer)
48	n/a	t.start()
49	n/a	import_started.wait()
50	n/a	pid = os.fork()
51	n/a	try:
52	n/a	# PyOS_BeforeFork should have waited for the import to complete
53	n/a	# before forking, so the child can recreate the import lock
54	n/a	# correctly, but also won't see a partially initialised module
55	n/a	if not pid:
56	n/a	m = __import__(fake_module_name)
57	n/a	if m == complete_module:
58	n/a	os._exit(0)
59	n/a	else:
60	n/a	if verbose > 1:
61	n/a	print("Child encountered partial module")
62	n/a	os._exit(1)
63	n/a	else:
64	n/a	t.join()
65	n/a	# Exitcode 1 means the child got a partial module (bad.) No
66	n/a	# exitcode (but a hang, which manifests as 'got pid 0')
67	n/a	# means the child deadlocked (also bad.)
68	n/a	self.wait_impl(pid)
69	n/a	finally:
70	n/a	try:
71	n/a	os.kill(pid, signal.SIGKILL)
72	n/a	except OSError:
73	n/a	pass
74	n/a
75	n/a
76	n/a	def test_nested_import_lock_fork(self):
77	n/a	"""Check fork() in main thread works while the main thread is doing an import"""
78	n/a	# Issue 9573: this used to trigger RuntimeError in the child process
79	n/a	def fork_with_import_lock(level):
80	n/a	release = 0
81	n/a	in_child = False
82	n/a	try:
83	n/a	try:
84	n/a	for i in range(level):
85	n/a	imp.acquire_lock()
86	n/a	release += 1
87	n/a	pid = os.fork()
88	n/a	in_child = not pid
89	n/a	finally:
90	n/a	for i in range(release):
91	n/a	imp.release_lock()
92	n/a	except RuntimeError:
93	n/a	if in_child:
94	n/a	if verbose > 1:
95	n/a	print("RuntimeError in child")
96	n/a	os._exit(1)
97	n/a	raise
98	n/a	if in_child:
99	n/a	os._exit(0)
100	n/a	self.wait_impl(pid)
101	n/a
102	n/a	# Check this works with various levels of nested
103	n/a	# import in the main thread
104	n/a	for level in range(5):
105	n/a	fork_with_import_lock(level)
106	n/a
107	n/a
108	n/a	def tearDownModule():
109	n/a	reap_children()
110	n/a
111	n/a	if __name__ == "__main__":
112	n/a	unittest.main()