MCPcopy Index your code
hub / github.com/python/cpython / create_worker_process

Function create_worker_process

Lib/test/libregrtest/worker.py:23–80  ·  view source on GitHub ↗
(runtests: WorkerRunTests, output_fd: int,
                          tmp_dir: StrPath | None = None)

Source from the content-addressed store, hash-verified

21
22
23def create_worker_process(runtests: WorkerRunTests, output_fd: int,
24 tmp_dir: StrPath | None = None) -> subprocess.Popen[str]:
25 worker_json = runtests.as_json()
26
27 cmd = runtests.create_python_cmd()
28 cmd.extend(['-m', 'test.libregrtest.worker', worker_json])
29
30 env = dict(os.environ)
31 if tmp_dir is not None:
32 env['TMPDIR'] = tmp_dir
33 env['TEMP'] = tmp_dir
34 env['TMP'] = tmp_dir
35
36 # The subcommand is run with a temporary output which means it is not a TTY
37 # and won't auto-color. The test results are printed to stdout so if we can
38 # color that have the subprocess use color.
39 if can_colorize(file=sys.stdout):
40 env['FORCE_COLOR'] = '1'
41
42 # Running the child from the same working directory as regrtest's original
43 # invocation ensures that TEMPDIR for the child is the same when
44 # sysconfig.is_python_build() is true. See issue 15300.
45 #
46 # Emscripten and WASI Python must start in the Python source code directory
47 # to get 'python.js' or 'python.wasm' file. Then worker_process() changes
48 # to a temporary directory created to run tests.
49 work_dir = os_helper.SAVEDCWD
50
51 kwargs: dict[str, Any] = dict(
52 env=env,
53 stdout=output_fd,
54 # bpo-45410: Write stderr into stdout to keep messages order
55 stderr=output_fd,
56 text=True,
57 close_fds=True,
58 cwd=work_dir,
59 )
60
61 # Don't use setsid() in tests using TTY
62 test_name = runtests.tests[0]
63 if USE_PROCESS_GROUP and test_name not in NEED_TTY:
64 kwargs['start_new_session'] = True
65
66 # Include the test name in the TSAN log file name
67 if 'TSAN_OPTIONS' in env:
68 parts = env['TSAN_OPTIONS'].split(' ')
69 for i, part in enumerate(parts):
70 if part.startswith('log_path='):
71 parts[i] = f'{part}.{test_name}'
72 break
73 env['TSAN_OPTIONS'] = ' '.join(parts)
74
75 # Pass json_file to the worker process
76 json_file = runtests.json_file
77 json_file.configure_subprocess(kwargs)
78
79 with json_file.inherit_subprocess():
80 return subprocess.Popen(cmd, **kwargs)

Callers 1

_run_processMethod · 0.85

Calls 10

can_colorizeFunction · 0.90
enumerateFunction · 0.85
as_jsonMethod · 0.80
create_python_cmdMethod · 0.80
configure_subprocessMethod · 0.80
inherit_subprocessMethod · 0.80
extendMethod · 0.45
splitMethod · 0.45
startswithMethod · 0.45
joinMethod · 0.45

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…