Lib/test/support/script_helper.py - platform/external/python/cpython3 - Git at Google

 # 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 subprocess
 import py_compile
 import zipfile

 from importlib.util import source_from_cache
 from test.support import make_legacy_pyc


 # 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 environment
     variables that might impact whether or not the interpreter can start.
     """
     global __cached_interp_requires_environment
     if __cached_interp_requires_environment is None:
         # If PYTHONHOME is set, assume that we need it
         if 'PYTHONHOME' in os.environ:
             __cached_interp_requires_environment = True
             return True

         # 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


 class _PythonRunResult(collections.namedtuple("_PythonRunResult",
                                           ("rc", "out", "err"))):
     """Helper for reporting Python subprocess run results"""
     def fail(self, cmd_line):
         """Provide helpful details about failed subcommand runs"""
         # Limit to 80 lines to ASCII characters
         maxlen = 80 * 100
         out, err = self.out, self.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"
                              "---"
                              % (self.rc, cmd_line,
                                 out,
                                 err))


 # Executing the interpreter in a subprocess
 def run_python_until_end(*args, **env_vars):
     env_required = interpreter_requires_environment()
     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 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')

     # 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 = {}
         if sys.platform == 'win32':
             # Windows requires at least the SYSTEMROOT environment variable to
             # start Python.
             env['SYSTEMROOT'] = os.environ['SYSTEMROOT']

         # Other interesting environment variables, not copied currently:
         # COMSPEC, HOME, PATH, TEMP, TMPDIR, TMP.
     else:
         # Need to preserve the original environment, for in-place testing of
         # shared library builds.
         env = os.environ.copy()

     # set TERM='' unless the TERM environment variable is passed explicitly
     # see issues #11390 and #18300
     if 'TERM' not in env_vars:
         env['TERM'] = ''

     env.update(env_vars)
     cmd_line.extend(args)
     proc = subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          env=env, cwd=cwd)
     with proc:
         try:
             out, err = proc.communicate()
         finally:
             proc.kill()
             subprocess._cleanup()
     rc = proc.returncode
     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):
         res.fail(cmd_line)
     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 as 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]
     if not interpreter_requires_environment():
         cmd_line.append('-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
     with open(script_name, 'w', encoding='utf-8') as script_file:
         script_file.write(source)
     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)
     with zipfile.ZipFile(zip_name, 'w') as zip_file:
         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)
     #if test.support.verbose:
     #    with zipfile.ZipFile(zip_name, 'r') as zip_file:
     #        print 'Contents of %r:' % zip_name
     #        zip_file.printdir()
     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)
     with zipfile.ZipFile(zip_name, 'w') as zip_file:
         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)
     for name in unlink:
         os.unlink(name)
     #if test.support.verbose:
     #    with zipfile.ZipFile(zip_name, 'r') as zip_file:
     #        print 'Contents of %r:' % zip_name
     #        zip_file.printdir()
     return zip_name, os.path.join(zip_name, script_name_in_zip)
	# 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 subprocess
	import py_compile
	import zipfile

	from importlib.util import source_from_cache
	from test.support import make_legacy_pyc


	# 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 environment
	variables that might impact whether or not the interpreter can start.
	"""
	global __cached_interp_requires_environment
	if __cached_interp_requires_environment is None:
	# If PYTHONHOME is set, assume that we need it
	if 'PYTHONHOME' in os.environ:
	__cached_interp_requires_environment = True
	return True

	# 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


	class _PythonRunResult(collections.namedtuple("_PythonRunResult",
	("rc", "out", "err"))):
	"""Helper for reporting Python subprocess run results"""
	def fail(self, cmd_line):
	"""Provide helpful details about failed subcommand runs"""
	# Limit to 80 lines to ASCII characters
	maxlen = 80 * 100
	out, err = self.out, self.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"
	"---"
	% (self.rc, cmd_line,
	out,
	err))


	# Executing the interpreter in a subprocess
	def run_python_until_end(args, *env_vars):
	env_required = interpreter_requires_environment()
	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 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')

	# 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 = {}
	if sys.platform == 'win32':
	# Windows requires at least the SYSTEMROOT environment variable to
	# start Python.
	env['SYSTEMROOT'] = os.environ['SYSTEMROOT']

	# Other interesting environment variables, not copied currently:
	# COMSPEC, HOME, PATH, TEMP, TMPDIR, TMP.
	else:
	# Need to preserve the original environment, for in-place testing of
	# shared library builds.
	env = os.environ.copy()

	# set TERM='' unless the TERM environment variable is passed explicitly
	# see issues #11390 and #18300
	if 'TERM' not in env_vars:
	env['TERM'] = ''

	env.update(env_vars)
	cmd_line.extend(args)
	proc = subprocess.Popen(cmd_line, stdin=subprocess.PIPE,
	stdout=subprocess.PIPE, stderr=subprocess.PIPE,
	env=env, cwd=cwd)
	with proc:
	try:
	out, err = proc.communicate()
	finally:
	proc.kill()
	subprocess._cleanup()
	rc = proc.returncode
	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):
	res.fail(cmd_line)
	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 as 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]
	if not interpreter_requires_environment():
	cmd_line.append('-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
	with open(script_name, 'w', encoding='utf-8') as script_file:
	script_file.write(source)
	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)
	with zipfile.ZipFile(zip_name, 'w') as zip_file:
	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)
	#if test.support.verbose:
	# with zipfile.ZipFile(zip_name, 'r') as zip_file:
	# print 'Contents of %r:' % zip_name
	# zip_file.printdir()
	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)
	with zipfile.ZipFile(zip_name, 'w') as zip_file:
	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)
	for name in unlink:
	os.unlink(name)
	#if test.support.verbose:
	# with zipfile.ZipFile(zip_name, 'r') as zip_file:
	# print 'Contents of %r:' % zip_name
	# zip_file.printdir()
	return zip_name, os.path.join(zip_name, script_name_in_zip)