# Executing the interpreter in a subprocess
@support.requires_subprocess()
def run_python_until_end(*args, **env_vars):
+ """Used to implement assert_python_*.
+
+ *args are the command line flags to pass to the python interpreter.
+ **env_vars keyword arguments are environment variables to set on the process.
+
+ If __run_using_command= is supplied, it must be a list of
+ command line arguments to prepend to the command line used.
+ Useful when you want to run another command that should launch the
+ python interpreter via its own arguments. ["/bin/echo", "--"] for
+ example could print the unquoted python command line instead of
+ run it.
+ """
env_required = interpreter_requires_environment()
+ run_using_command = env_vars.pop('__run_using_command', None)
cwd = env_vars.pop('__cwd', None)
if '__isolated' in env_vars:
isolated = env_vars.pop('__isolated')
else:
isolated = not env_vars and not env_required
cmd_line = [sys.executable, '-X', 'faulthandler']
+ if run_using_command:
+ cmd_line = run_using_command + cmd_line
if isolated:
# isolated mode: ignore Python environment variables, ignore user
# site-packages, and don't add the current directory to sys.path
self.assertIsInstance(subprocess.Popen[bytes], types.GenericAlias)
self.assertIsInstance(subprocess.CompletedProcess[str], types.GenericAlias)
- @unittest.skipIf(not sysconfig.get_config_var("HAVE_VFORK"),
- "vfork() not enabled by configure.")
- @mock.patch("subprocess._fork_exec")
- def test__use_vfork(self, mock_fork_exec):
- self.assertTrue(subprocess._USE_VFORK) # The default value regardless.
- mock_fork_exec.side_effect = RuntimeError("just testing args")
- with self.assertRaises(RuntimeError):
- subprocess.run([sys.executable, "-c", "pass"])
- mock_fork_exec.assert_called_once()
- self.assertTrue(mock_fork_exec.call_args.args[-1])
- with mock.patch.object(subprocess, '_USE_VFORK', False):
- with self.assertRaises(RuntimeError):
- subprocess.run([sys.executable, "-c", "pass"])
- self.assertFalse(mock_fork_exec.call_args_list[-1].args[-1])
-
class RunFuncTestCase(BaseTestCase):
def run_python(self, code, **kwargs):
self.assertEqual(out, b'')
self.assertIn(b"preexec_fn not supported at interpreter shutdown", err)
+ @unittest.skipIf(not sysconfig.get_config_var("HAVE_VFORK"),
+ "vfork() not enabled by configure.")
+ @mock.patch("subprocess._fork_exec")
+ def test__use_vfork(self, mock_fork_exec):
+ self.assertTrue(subprocess._USE_VFORK) # The default value regardless.
+ mock_fork_exec.side_effect = RuntimeError("just testing args")
+ with self.assertRaises(RuntimeError):
+ subprocess.run([sys.executable, "-c", "pass"])
+ mock_fork_exec.assert_called_once()
+ # NOTE: These assertions are *ugly* as they require the last arg
+ # to remain the have_vfork boolean. We really need to refactor away
+ # from the giant "wall of args" internal C extension API.
+ self.assertTrue(mock_fork_exec.call_args.args[-1])
+ with mock.patch.object(subprocess, '_USE_VFORK', False):
+ with self.assertRaises(RuntimeError):
+ subprocess.run([sys.executable, "-c", "pass"])
+ self.assertFalse(mock_fork_exec.call_args_list[-1].args[-1])
+
+ @unittest.skipIf(not sysconfig.get_config_var("HAVE_VFORK"),
+ "vfork() not enabled by configure.")
+ @unittest.skipIf(sys.platform != "linux", "Linux only, requires strace.")
+ def test_vfork_used_when_expected(self):
+ # This is a performance regression test to ensure we default to using
+ # vfork() when possible.
+ strace_binary = "/usr/bin/strace"
+ # The only system calls we are interested in.
+ strace_filter = "--trace=clone,clone2,clone3,fork,vfork,exit,exit_group"
+ true_binary = "/bin/true"
+ strace_command = [strace_binary, strace_filter]
+
+ try:
+ does_strace_work_process = subprocess.run(
+ strace_command + [true_binary],
+ stderr=subprocess.PIPE,
+ stdout=subprocess.DEVNULL,
+ )
+ rc = does_strace_work_process.returncode
+ stderr = does_strace_work_process.stderr
+ except OSError:
+ rc = -1
+ stderr = ""
+ if rc or (b"+++ exited with 0 +++" not in stderr):
+ self.skipTest("strace not found or not working as expected.")
+
+ with self.subTest(name="default_is_vfork"):
+ vfork_result = assert_python_ok(
+ "-c",
+ textwrap.dedent(f"""\
+ import subprocess
+ subprocess.check_call([{true_binary!r}])"""),
+ __run_using_command=strace_command,
+ )
+ # Match both vfork() and clone(..., flags=...|CLONE_VFORK|...)
+ self.assertRegex(vfork_result.err, br"(?i)vfork")
+ # Do NOT check that fork() or other clones did not happen.
+ # If the OS denys the vfork it'll fallback to plain fork().
+
+ # Test that each individual thing that would disable the use of vfork
+ # actually disables it.
+ for sub_name, preamble, sp_kwarg, expect_permission_error in (
+ ("!use_vfork", "subprocess._USE_VFORK = False", "", False),
+ ("preexec", "", "preexec_fn=lambda: None", False),
+ ("setgid", "", f"group={os.getgid()}", True),
+ ("setuid", "", f"user={os.getuid()}", True),
+ ("setgroups", "", "extra_groups=[]", True),
+ ):
+ with self.subTest(name=sub_name):
+ non_vfork_result = assert_python_ok(
+ "-c",
+ textwrap.dedent(f"""\
+ import subprocess
+ {preamble}
+ try:
+ subprocess.check_call(
+ [{true_binary!r}], **dict({sp_kwarg}))
+ except PermissionError:
+ if not {expect_permission_error}:
+ raise"""),
+ __run_using_command=strace_command,
+ )
+ # Ensure neither vfork() or clone(..., flags=...|CLONE_VFORK|...).
+ self.assertNotRegex(non_vfork_result.err, br"(?i)vfork")
+
@unittest.skipUnless(mswindows, "Windows specific tests")
class Win32ProcessTestCase(BaseTestCase):
projects / thirdparty / Python / cpython.git / commitdiff
