|
1 | 1 | """Test suite for the cProfile module.""" |
2 | | -import subprocess |
3 | 2 | import sys |
4 | 3 | import unittest |
5 | 4 |
|
|
9 | 8 | import textwrap |
10 | 9 | from test.test_profile import ProfileTest, regenerate_expected_output |
11 | 10 | from test.support.script_helper import assert_python_failure, assert_python_ok |
12 | | -from test.support import ( |
13 | | - SHORT_TIMEOUT, |
14 | | - SuppressCrashReport, |
15 | | - os_helper, |
16 | | - script_helper, |
17 | | -) |
18 | 11 | from test import support |
19 | 12 |
|
20 | 13 |
|
@@ -198,51 +191,30 @@ class Foo: |
198 | 191 | f.close() |
199 | 192 | assert_python_ok('-m', "cProfile", f.name) |
200 | 193 |
|
201 | | - |
202 | | -class TestProcessRunSupport(unittest.TestCase): |
203 | | - """ |
204 | | - Test that Process works correctly with cProfile. |
205 | | - """ |
206 | | - |
207 | 194 | def test_process_run_pickle(self): |
208 | 195 | # gh-140729: test use Process in cProfile. |
209 | 196 | val = 10 |
210 | | - test_script = f''' |
211 | | -import multiprocessing |
212 | | -
|
213 | | -def worker(x): |
214 | | - print(__name__) |
215 | | - exit(x ** 2) |
216 | | -
|
217 | | -if __name__ == "__main__": |
218 | | - multiprocessing.set_start_method("spawn") |
219 | | - p = multiprocessing.Process(target=worker, args=({val},)) |
220 | | - p.start() |
221 | | - p.join() |
222 | | - print("p.exitcode =", p.exitcode) |
223 | | -''' |
224 | | - |
225 | | - with os_helper.temp_dir() as temp_dir: |
226 | | - script = script_helper.make_script( |
227 | | - temp_dir, 'test_process_run_pickle', test_script |
228 | | - ) |
229 | | - with SuppressCrashReport(): |
230 | | - with script_helper.spawn_python( |
231 | | - "-m", "cProfile", |
232 | | - script, |
233 | | - stdout=subprocess.PIPE, |
234 | | - stderr=subprocess.PIPE, |
235 | | - text=True |
236 | | - ) as proc: |
237 | | - try: |
238 | | - stdout, stderr = proc.communicate(timeout=SHORT_TIMEOUT) |
239 | | - except subprocess.TimeoutExpired: |
240 | | - proc.kill() |
241 | | - stdout, stderr = proc.communicate() |
242 | | - |
243 | | - self.assertIn("__mp_main__", stdout) |
244 | | - self.assertIn(f"exitcode = {val**2}", stdout) |
245 | | - self.assertNotIn("Can't pickle", stderr) |
| 197 | + with tempfile.NamedTemporaryFile("w+", delete_on_close=False) as f: |
| 198 | + f.write(textwrap.dedent( |
| 199 | + f'''\ |
| 200 | + import multiprocessing |
| 201 | +
|
| 202 | + def worker(x): |
| 203 | + print(__name__) |
| 204 | + exit(x ** 2) |
| 205 | +
|
| 206 | + if __name__ == "__main__": |
| 207 | + multiprocessing.set_start_method("spawn") |
| 208 | + p = multiprocessing.Process(target=worker, args=({val},)) |
| 209 | + p.start() |
| 210 | + p.join() |
| 211 | + print("p.exitcode =", p.exitcode) |
| 212 | + ''')) |
| 213 | + f.close() |
| 214 | + _, out, err = assert_python_ok('-m', "cProfile", f.name) |
| 215 | + self.assertIn(b"__mp_main__", out) |
| 216 | + self.assertIn(bytes(f"exitcode = {val**2}", encoding='utf8'), out) |
| 217 | + self.assertNotIn(b"Can't pickle", err) |
246 | 218 |
|
247 | 219 |
|
248 | 220 | def main(): |
|
0 commit comments