Python code coverage for Lib/test/test_multiprocessing_main_handling.py

#	count	content
1	n/a	# tests __main__ module handling in multiprocessing
2	n/a	from test import support
3	n/a	# Skip tests if _thread or _multiprocessing wasn't built.
4	n/a	support.import_module('_thread')
5	n/a	support.import_module('_multiprocessing')
6	n/a
7	n/a	import importlib
8	n/a	import importlib.machinery
9	n/a	import unittest
10	n/a	import sys
11	n/a	import os
12	n/a	import os.path
13	n/a	import py_compile
14	n/a
15	n/a	from test.support.script_helper import (
16	n/a	make_pkg, make_script, make_zip_pkg, make_zip_script,
17	n/a	assert_python_ok)
18	n/a
19	n/a	if support.PGO:
20	n/a	raise unittest.SkipTest("test is not helpful for PGO")
21	n/a
22	n/a	# Look up which start methods are available to test
23	n/a	import multiprocessing
24	n/a	AVAILABLE_START_METHODS = set(multiprocessing.get_all_start_methods())
25	n/a
26	n/a	# Issue #22332: Skip tests if sem_open implementation is broken.
27	n/a	support.import_module('multiprocessing.synchronize')
28	n/a
29	n/a	verbose = support.verbose
30	n/a
31	n/a	test_source = """\
32	n/a	# multiprocessing includes all sorts of shenanigans to make __main__
33	n/a	# attributes accessible in the subprocess in a pickle compatible way.
34	n/a
35	n/a	# We run the "doesn't work in the interactive interpreter" example from
36	n/a	# the docs to make sure it does work from an executed __main__,
37	n/a	# regardless of the invocation mechanism
38	n/a
39	n/a	import sys
40	n/a	import time
41	n/a	from multiprocessing import Pool, set_start_method
42	n/a
43	n/a	# We use this __main__ defined function in the map call below in order to
44	n/a	# check that multiprocessing in correctly running the unguarded
45	n/a	# code in child processes and then making it available as __main__
46	n/a	def f(x):
47	n/a	return x*x
48	n/a
49	n/a	# Check explicit relative imports
50	n/a	if "check_sibling" in __file__:
51	n/a	# We're inside a package and not in a __main__.py file
52	n/a	# so make sure explicit relative imports work correctly
53	n/a	from . import sibling
54	n/a
55	n/a	if __name__ == '__main__':
56	n/a	start_method = sys.argv[1]
57	n/a	set_start_method(start_method)
58	n/a	p = Pool(5)
59	n/a	results = []
60	n/a	p.map_async(f, [1, 2, 3], callback=results.extend)
61	n/a	deadline = time.time() + 10 # up to 10 s to report the results
62	n/a	while not results:
63	n/a	time.sleep(0.05)
64	n/a	if time.time() > deadline:
65	n/a	raise RuntimeError("Timed out waiting for results")
66	n/a	results.sort()
67	n/a	print(start_method, "->", results)
68	n/a	"""
69	n/a
70	n/a	test_source_main_skipped_in_children = """\
71	n/a	# __main__.py files have an implied "if __name__ == '__main__'" so
72	n/a	# multiprocessing should always skip running them in child processes
73	n/a
74	n/a	# This means we can't use __main__ defined functions in child processes,
75	n/a	# so we just use "int" as a passthrough operation below
76	n/a
77	n/a	if __name__ != "__main__":
78	n/a	raise RuntimeError("Should only be called as __main__!")
79	n/a
80	n/a	import sys
81	n/a	import time
82	n/a	from multiprocessing import Pool, set_start_method
83	n/a
84	n/a	start_method = sys.argv[1]
85	n/a	set_start_method(start_method)
86	n/a	p = Pool(5)
87	n/a	results = []
88	n/a	p.map_async(int, [1, 4, 9], callback=results.extend)
89	n/a	deadline = time.time() + 10 # up to 10 s to report the results
90	n/a	while not results:
91	n/a	time.sleep(0.05)
92	n/a	if time.time() > deadline:
93	n/a	raise RuntimeError("Timed out waiting for results")
94	n/a	results.sort()
95	n/a	print(start_method, "->", results)
96	n/a	"""
97	n/a
98	n/a	# These helpers were copied from test_cmd_line_script & tweaked a bit...
99	n/a
100	n/a	def _make_test_script(script_dir, script_basename,
101	n/a	source=test_source, omit_suffix=False):
102	n/a	to_return = make_script(script_dir, script_basename,
103	n/a	source, omit_suffix)
104	n/a	# Hack to check explicit relative imports
105	n/a	if script_basename == "check_sibling":
106	n/a	make_script(script_dir, "sibling", "")
107	n/a	importlib.invalidate_caches()
108	n/a	return to_return
109	n/a
110	n/a	def _make_test_zip_pkg(zip_dir, zip_basename, pkg_name, script_basename,
111	n/a	source=test_source, depth=1):
112	n/a	to_return = make_zip_pkg(zip_dir, zip_basename, pkg_name, script_basename,
113	n/a	source, depth)
114	n/a	importlib.invalidate_caches()
115	n/a	return to_return
116	n/a
117	n/a	# There's no easy way to pass the script directory in to get
118	n/a	# -m to work (avoiding that is the whole point of making
119	n/a	# directories and zipfiles executable!)
120	n/a	# So we fake it for testing purposes with a custom launch script
121	n/a	launch_source = """\
122	n/a	import sys, os.path, runpy
123	n/a	sys.path.insert(0, %s)
124	n/a	runpy._run_module_as_main(%r)
125	n/a	"""
126	n/a
127	n/a	def _make_launch_script(script_dir, script_basename, module_name, path=None):
128	n/a	if path is None:
129	n/a	path = "os.path.dirname(__file__)"
130	n/a	else:
131	n/a	path = repr(path)
132	n/a	source = launch_source % (path, module_name)
133	n/a	to_return = make_script(script_dir, script_basename, source)
134	n/a	importlib.invalidate_caches()
135	n/a	return to_return
136	n/a
137	n/a	class MultiProcessingCmdLineMixin():
138	n/a	maxDiff = None # Show full tracebacks on subprocess failure
139	n/a
140	n/a	def setUp(self):
141	n/a	if self.start_method not in AVAILABLE_START_METHODS:
142	n/a	self.skipTest("%r start method not available" % self.start_method)
143	n/a
144	n/a	def _check_output(self, script_name, exit_code, out, err):
145	n/a	if verbose > 1:
146	n/a	print("Output from test script %r:" % script_name)
147	n/a	print(repr(out))
148	n/a	self.assertEqual(exit_code, 0)
149	n/a	self.assertEqual(err.decode('utf-8'), '')
150	n/a	expected_results = "%s -> [1, 4, 9]" % self.start_method
151	n/a	self.assertEqual(out.decode('utf-8').strip(), expected_results)
152	n/a
153	n/a	def _check_script(self, script_name, *cmd_line_switches):
154	n/a	if not __debug__:
155	n/a	cmd_line_switches += ('-' + 'O' * sys.flags.optimize,)
156	n/a	run_args = cmd_line_switches + (script_name, self.start_method)
157	n/a	rc, out, err = assert_python_ok(*run_args, __isolated=False)
158	n/a	self._check_output(script_name, rc, out, err)
159	n/a
160	n/a	def test_basic_script(self):
161	n/a	with support.temp_dir() as script_dir:
162	n/a	script_name = _make_test_script(script_dir, 'script')
163	n/a	self._check_script(script_name)
164	n/a
165	n/a	def test_basic_script_no_suffix(self):
166	n/a	with support.temp_dir() as script_dir:
167	n/a	script_name = _make_test_script(script_dir, 'script',
168	n/a	omit_suffix=True)
169	n/a	self._check_script(script_name)
170	n/a
171	n/a	def test_ipython_workaround(self):
172	n/a	# Some versions of the IPython launch script are missing the
173	n/a	# __name__ = "__main__" guard, and multiprocessing has long had
174	n/a	# a workaround for that case
175	n/a	# See https://github.com/ipython/ipython/issues/4698
176	n/a	source = test_source_main_skipped_in_children
177	n/a	with support.temp_dir() as script_dir:
178	n/a	script_name = _make_test_script(script_dir, 'ipython',
179	n/a	source=source)
180	n/a	self._check_script(script_name)
181	n/a	script_no_suffix = _make_test_script(script_dir, 'ipython',
182	n/a	source=source,
183	n/a	omit_suffix=True)
184	n/a	self._check_script(script_no_suffix)
185	n/a
186	n/a	def test_script_compiled(self):
187	n/a	with support.temp_dir() as script_dir:
188	n/a	script_name = _make_test_script(script_dir, 'script')
189	n/a	py_compile.compile(script_name, doraise=True)
190	n/a	os.remove(script_name)
191	n/a	pyc_file = support.make_legacy_pyc(script_name)
192	n/a	self._check_script(pyc_file)
193	n/a
194	n/a	def test_directory(self):
195	n/a	source = self.main_in_children_source
196	n/a	with support.temp_dir() as script_dir:
197	n/a	script_name = _make_test_script(script_dir, '__main__',
198	n/a	source=source)
199	n/a	self._check_script(script_dir)
200	n/a
201	n/a	def test_directory_compiled(self):
202	n/a	source = self.main_in_children_source
203	n/a	with support.temp_dir() as script_dir:
204	n/a	script_name = _make_test_script(script_dir, '__main__',
205	n/a	source=source)
206	n/a	py_compile.compile(script_name, doraise=True)
207	n/a	os.remove(script_name)
208	n/a	pyc_file = support.make_legacy_pyc(script_name)
209	n/a	self._check_script(script_dir)
210	n/a
211	n/a	def test_zipfile(self):
212	n/a	source = self.main_in_children_source
213	n/a	with support.temp_dir() as script_dir:
214	n/a	script_name = _make_test_script(script_dir, '__main__',
215	n/a	source=source)
216	n/a	zip_name, run_name = make_zip_script(script_dir, 'test_zip', script_name)
217	n/a	self._check_script(zip_name)
218	n/a
219	n/a	def test_zipfile_compiled(self):
220	n/a	source = self.main_in_children_source
221	n/a	with support.temp_dir() as script_dir:
222	n/a	script_name = _make_test_script(script_dir, '__main__',
223	n/a	source=source)
224	n/a	compiled_name = py_compile.compile(script_name, doraise=True)
225	n/a	zip_name, run_name = make_zip_script(script_dir, 'test_zip', compiled_name)
226	n/a	self._check_script(zip_name)
227	n/a
228	n/a	def test_module_in_package(self):
229	n/a	with support.temp_dir() as script_dir:
230	n/a	pkg_dir = os.path.join(script_dir, 'test_pkg')
231	n/a	make_pkg(pkg_dir)
232	n/a	script_name = _make_test_script(pkg_dir, 'check_sibling')
233	n/a	launch_name = _make_launch_script(script_dir, 'launch',
234	n/a	'test_pkg.check_sibling')
235	n/a	self._check_script(launch_name)
236	n/a
237	n/a	def test_module_in_package_in_zipfile(self):
238	n/a	with support.temp_dir() as script_dir:
239	n/a	zip_name, run_name = _make_test_zip_pkg(script_dir, 'test_zip', 'test_pkg', 'script')
240	n/a	launch_name = _make_launch_script(script_dir, 'launch', 'test_pkg.script', zip_name)
241	n/a	self._check_script(launch_name)
242	n/a
243	n/a	def test_module_in_subpackage_in_zipfile(self):
244	n/a	with support.temp_dir() as script_dir:
245	n/a	zip_name, run_name = _make_test_zip_pkg(script_dir, 'test_zip', 'test_pkg', 'script', depth=2)
246	n/a	launch_name = _make_launch_script(script_dir, 'launch', 'test_pkg.test_pkg.script', zip_name)
247	n/a	self._check_script(launch_name)
248	n/a
249	n/a	def test_package(self):
250	n/a	source = self.main_in_children_source
251	n/a	with support.temp_dir() as script_dir:
252	n/a	pkg_dir = os.path.join(script_dir, 'test_pkg')
253	n/a	make_pkg(pkg_dir)
254	n/a	script_name = _make_test_script(pkg_dir, '__main__',
255	n/a	source=source)
256	n/a	launch_name = _make_launch_script(script_dir, 'launch', 'test_pkg')
257	n/a	self._check_script(launch_name)
258	n/a
259	n/a	def test_package_compiled(self):
260	n/a	source = self.main_in_children_source
261	n/a	with support.temp_dir() as script_dir:
262	n/a	pkg_dir = os.path.join(script_dir, 'test_pkg')
263	n/a	make_pkg(pkg_dir)
264	n/a	script_name = _make_test_script(pkg_dir, '__main__',
265	n/a	source=source)
266	n/a	compiled_name = py_compile.compile(script_name, doraise=True)
267	n/a	os.remove(script_name)
268	n/a	pyc_file = support.make_legacy_pyc(script_name)
269	n/a	launch_name = _make_launch_script(script_dir, 'launch', 'test_pkg')
270	n/a	self._check_script(launch_name)
271	n/a
272	n/a	# Test all supported start methods (setupClass skips as appropriate)
273	n/a
274	n/a	class SpawnCmdLineTest(MultiProcessingCmdLineMixin, unittest.TestCase):
275	n/a	start_method = 'spawn'
276	n/a	main_in_children_source = test_source_main_skipped_in_children
277	n/a
278	n/a	class ForkCmdLineTest(MultiProcessingCmdLineMixin, unittest.TestCase):
279	n/a	start_method = 'fork'
280	n/a	main_in_children_source = test_source
281	n/a
282	n/a	class ForkServerCmdLineTest(MultiProcessingCmdLineMixin, unittest.TestCase):
283	n/a	start_method = 'forkserver'
284	n/a	main_in_children_source = test_source_main_skipped_in_children
285	n/a
286	n/a	def tearDownModule():
287	n/a	support.reap_children()
288	n/a
289	n/a	if __name__ == '__main__':
290	n/a	unittest.main()