mirror of
https://github.com/python/cpython.git
synced 2025-11-04 07:31:38 +00:00
cb46f0ecb0
while it is holding a lock to a buffered I/O object, and the main thread tries to use the same I/O object (typically stdout or stderr). A fatal error is emitted instead.
244 lines
9.4 KiB
Python
244 lines
9.4 KiB
Python
# Common utility functions used by various script execution tests
|
|
# e.g. test_cmd_line, test_cmd_line_script and test_runpy
|
|
|
|
import collections
|
|
import importlib
|
|
import sys
|
|
import os
|
|
import os.path
|
|
import tempfile
|
|
import subprocess
|
|
import py_compile
|
|
import contextlib
|
|
import shutil
|
|
import zipfile
|
|
|
|
from importlib.util import source_from_cache
|
|
from test.support import make_legacy_pyc, strip_python_stderr, temp_dir
|
|
|
|
|
|
# Cached result of the expensive test performed in the function below.
|
|
__cached_interp_requires_environment = None
|
|
|
|
def interpreter_requires_environment():
|
|
"""
|
|
Returns True if our sys.executable interpreter requires environment
|
|
variables in order to be able to run at all.
|
|
|
|
This is designed to be used with @unittest.skipIf() to annotate tests
|
|
that need to use an assert_python*() function to launch an isolated
|
|
mode (-I) or no environment mode (-E) sub-interpreter process.
|
|
|
|
A normal build & test does not run into this situation but it can happen
|
|
when trying to run the standard library test suite from an interpreter that
|
|
doesn't have an obvious home with Python's current home finding logic.
|
|
|
|
Setting PYTHONHOME is one way to get most of the testsuite to run in that
|
|
situation. PYTHONPATH or PYTHONUSERSITE are other common envirnonment
|
|
variables that might impact whether or not the interpreter can start.
|
|
"""
|
|
global __cached_interp_requires_environment
|
|
if __cached_interp_requires_environment is None:
|
|
# Try running an interpreter with -E to see if it works or not.
|
|
try:
|
|
subprocess.check_call([sys.executable, '-E',
|
|
'-c', 'import sys; sys.exit(0)'])
|
|
except subprocess.CalledProcessError:
|
|
__cached_interp_requires_environment = True
|
|
else:
|
|
__cached_interp_requires_environment = False
|
|
|
|
return __cached_interp_requires_environment
|
|
|
|
|
|
_PythonRunResult = collections.namedtuple("_PythonRunResult",
|
|
("rc", "out", "err"))
|
|
|
|
|
|
# Executing the interpreter in a subprocess
|
|
def run_python_until_end(*args, **env_vars):
|
|
env_required = interpreter_requires_environment()
|
|
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 isolated:
|
|
# isolated mode: ignore Python environment variables, ignore user
|
|
# site-packages, and don't add the current directory to sys.path
|
|
cmd_line.append('-I')
|
|
elif not env_vars and not env_required:
|
|
# ignore Python environment variables
|
|
cmd_line.append('-E')
|
|
# Need to preserve the original environment, for in-place testing of
|
|
# shared library builds.
|
|
env = os.environ.copy()
|
|
# But a special flag that can be set to override -- in this case, the
|
|
# caller is responsible to pass the full environment.
|
|
if env_vars.pop('__cleanenv', None):
|
|
env = {}
|
|
env.update(env_vars)
|
|
cmd_line.extend(args)
|
|
p = subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
env=env)
|
|
try:
|
|
out, err = p.communicate()
|
|
finally:
|
|
subprocess._cleanup()
|
|
p.stdout.close()
|
|
p.stderr.close()
|
|
rc = p.returncode
|
|
err = strip_python_stderr(err)
|
|
return _PythonRunResult(rc, out, err), cmd_line
|
|
|
|
def _assert_python(expected_success, *args, **env_vars):
|
|
res, cmd_line = run_python_until_end(*args, **env_vars)
|
|
if (res.rc and expected_success) or (not res.rc and not expected_success):
|
|
# Limit to 80 lines to ASCII characters
|
|
maxlen = 80 * 100
|
|
out, err = res.out, res.err
|
|
if len(out) > maxlen:
|
|
out = b'(... truncated stdout ...)' + out[-maxlen:]
|
|
if len(err) > maxlen:
|
|
err = b'(... truncated stderr ...)' + err[-maxlen:]
|
|
out = out.decode('ascii', 'replace').rstrip()
|
|
err = err.decode('ascii', 'replace').rstrip()
|
|
raise AssertionError("Process return code is %d\n"
|
|
"command line: %r\n"
|
|
"\n"
|
|
"stdout:\n"
|
|
"---\n"
|
|
"%s\n"
|
|
"---\n"
|
|
"\n"
|
|
"stderr:\n"
|
|
"---\n"
|
|
"%s\n"
|
|
"---"
|
|
% (res.rc, cmd_line,
|
|
out,
|
|
err))
|
|
return res
|
|
|
|
def assert_python_ok(*args, **env_vars):
|
|
"""
|
|
Assert that running the interpreter with `args` and optional environment
|
|
variables `env_vars` succeeds (rc == 0) and return a (return code, stdout,
|
|
stderr) tuple.
|
|
|
|
If the __cleanenv keyword is set, env_vars is used a fresh environment.
|
|
|
|
Python is started in isolated mode (command line option -I),
|
|
except if the __isolated keyword is set to False.
|
|
"""
|
|
return _assert_python(True, *args, **env_vars)
|
|
|
|
def assert_python_failure(*args, **env_vars):
|
|
"""
|
|
Assert that running the interpreter with `args` and optional environment
|
|
variables `env_vars` fails (rc != 0) and return a (return code, stdout,
|
|
stderr) tuple.
|
|
|
|
See assert_python_ok() for more options.
|
|
"""
|
|
return _assert_python(False, *args, **env_vars)
|
|
|
|
def spawn_python(*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kw):
|
|
"""Run a Python subprocess with the given arguments.
|
|
|
|
kw is extra keyword args to pass to subprocess.Popen. Returns a Popen
|
|
object.
|
|
"""
|
|
cmd_line = [sys.executable, '-E']
|
|
cmd_line.extend(args)
|
|
# Under Fedora (?), GNU readline can output junk on stderr when initialized,
|
|
# depending on the TERM setting. Setting TERM=vt100 is supposed to disable
|
|
# that. References:
|
|
# - http://reinout.vanrees.org/weblog/2009/08/14/readline-invisible-character-hack.html
|
|
# - http://stackoverflow.com/questions/15760712/python-readline-module-prints-escape-character-during-import
|
|
# - http://lists.gnu.org/archive/html/bug-readline/2007-08/msg00004.html
|
|
env = kw.setdefault('env', dict(os.environ))
|
|
env['TERM'] = 'vt100'
|
|
return subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
|
|
stdout=stdout, stderr=stderr,
|
|
**kw)
|
|
|
|
def kill_python(p):
|
|
"""Run the given Popen process until completion and return stdout."""
|
|
p.stdin.close()
|
|
data = p.stdout.read()
|
|
p.stdout.close()
|
|
# try to cleanup the child so we don't appear to leak when running
|
|
# with regrtest -R.
|
|
p.wait()
|
|
subprocess._cleanup()
|
|
return data
|
|
|
|
def make_script(script_dir, script_basename, source, omit_suffix=False):
|
|
script_filename = script_basename
|
|
if not omit_suffix:
|
|
script_filename += os.extsep + 'py'
|
|
script_name = os.path.join(script_dir, script_filename)
|
|
# The script should be encoded to UTF-8, the default string encoding
|
|
script_file = open(script_name, 'w', encoding='utf-8')
|
|
script_file.write(source)
|
|
script_file.close()
|
|
importlib.invalidate_caches()
|
|
return script_name
|
|
|
|
def make_zip_script(zip_dir, zip_basename, script_name, name_in_zip=None):
|
|
zip_filename = zip_basename+os.extsep+'zip'
|
|
zip_name = os.path.join(zip_dir, zip_filename)
|
|
zip_file = zipfile.ZipFile(zip_name, 'w')
|
|
if name_in_zip is None:
|
|
parts = script_name.split(os.sep)
|
|
if len(parts) >= 2 and parts[-2] == '__pycache__':
|
|
legacy_pyc = make_legacy_pyc(source_from_cache(script_name))
|
|
name_in_zip = os.path.basename(legacy_pyc)
|
|
script_name = legacy_pyc
|
|
else:
|
|
name_in_zip = os.path.basename(script_name)
|
|
zip_file.write(script_name, name_in_zip)
|
|
zip_file.close()
|
|
#if test.support.verbose:
|
|
# zip_file = zipfile.ZipFile(zip_name, 'r')
|
|
# print 'Contents of %r:' % zip_name
|
|
# zip_file.printdir()
|
|
# zip_file.close()
|
|
return zip_name, os.path.join(zip_name, name_in_zip)
|
|
|
|
def make_pkg(pkg_dir, init_source=''):
|
|
os.mkdir(pkg_dir)
|
|
make_script(pkg_dir, '__init__', init_source)
|
|
|
|
def make_zip_pkg(zip_dir, zip_basename, pkg_name, script_basename,
|
|
source, depth=1, compiled=False):
|
|
unlink = []
|
|
init_name = make_script(zip_dir, '__init__', '')
|
|
unlink.append(init_name)
|
|
init_basename = os.path.basename(init_name)
|
|
script_name = make_script(zip_dir, script_basename, source)
|
|
unlink.append(script_name)
|
|
if compiled:
|
|
init_name = py_compile.compile(init_name, doraise=True)
|
|
script_name = py_compile.compile(script_name, doraise=True)
|
|
unlink.extend((init_name, script_name))
|
|
pkg_names = [os.sep.join([pkg_name]*i) for i in range(1, depth+1)]
|
|
script_name_in_zip = os.path.join(pkg_names[-1], os.path.basename(script_name))
|
|
zip_filename = zip_basename+os.extsep+'zip'
|
|
zip_name = os.path.join(zip_dir, zip_filename)
|
|
zip_file = zipfile.ZipFile(zip_name, 'w')
|
|
for name in pkg_names:
|
|
init_name_in_zip = os.path.join(name, init_basename)
|
|
zip_file.write(init_name, init_name_in_zip)
|
|
zip_file.write(script_name, script_name_in_zip)
|
|
zip_file.close()
|
|
for name in unlink:
|
|
os.unlink(name)
|
|
#if test.support.verbose:
|
|
# zip_file = zipfile.ZipFile(zip_name, 'r')
|
|
# print 'Contents of %r:' % zip_name
|
|
# zip_file.printdir()
|
|
# zip_file.close()
|
|
return zip_name, os.path.join(zip_name, script_name_in_zip)
|