ยปCore Development>Code coverage>Lib/test/test_multiprocessing_main_handling.py

Python code coverage for Lib/test/test_multiprocessing_main_handling.py

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