"""Run tests in a subprocess

#: str, IPython test suite to be executed.

section = None

#: list, command line arguments to be executed

cmd = None

#: dict, extra environment variables to set for the subprocess

env: Dict[str, str] = {}

#: list, TemporaryDirectory instances to clear up when the process finishes

dirs = None

#: subprocess.Popen instance

process = None

#: str, process stdout+stderr

stdout = None

def __init__(self):

self.cmd = []

self.env = {}

self.dirs = []

def setUp(self):

"""Create temporary directories etc.

pass

def launch(self, buffer_output=False, capture_output=False):

# print('*** ENV:', self.env) # dbg

# print('*** CMD:', self.cmd) # dbg

env = os.environ.copy()

env.update(self.env)

if buffer_output:

capture_output = True

self.stdout_capturer = c = StreamCapturer(echo=not buffer_output)

c.start()

stdout = c.writefd if capture_output else None

stderr = subprocess.STDOUT if capture_output else None

for k, v in env.items():

assert isinstance(v, str), f"env[{repr(k)}] is not a str: {v}"

self.process = subprocess.Popen(self.cmd, stdout=stdout,

stderr=stderr, env=env)

def wait(self):

self.process.wait()

self.stdout_capturer.halt()

self.stdout = self.stdout_capturer.get_buffer()

return self.process.returncode

def cleanup_process(self):

"""Cleanup on exit by killing any leftover processes."""

subp = self.process

if subp is None or (subp.poll() is not None):

return # Process doesn't exist, or is already dead.

try:

print('Cleaning up stale PID: %d' % subp.pid)

subp.kill()

except: # (OSError, WindowsError) ?

# This is just a best effort, if we fail or the process was

# really gone, ignore it.

pass

else:

for i in range(10):

if subp.poll() is None:

time.sleep(0.1)

else:

break

if subp.poll() is None:

# The process did not die...

print('... failed. Manual cleanup may be required.')

def cleanup(self):

"Kill process if it's still alive, and clean up temporary directories"

self.cleanup_process()

for td in self.dirs:

td.cleanup()

"""Run Python tests using IPython.testing.iptest"""

#: str, Python command to execute in subprocess

pycmd = None

def __init__(self, section, options):

"""Create new test runner."""

TestController.__init__(self)

self.section = section

# pycmd is put into cmd[2] in PyTestController.launch()

self.cmd = [sys.executable, '-c', None, section]

self.pycmd = "from IPython.testing.iptest import run_iptest; run_iptest()"

self.options = options

def setup(self):

ipydir = TemporaryDirectory()

self.dirs.append(ipydir)

self.env['IPYTHONDIR'] = ipydir.name

self.workingdir = workingdir = TemporaryDirectory()

self.dirs.append(workingdir)

self.env['IPTEST_WORKING_DIR'] = workingdir.name

# This means we won't get odd effects from our own matplotlib config

self.env['MPLCONFIGDIR'] = workingdir.name

# For security reasons (http://bugs.python.org/issue16202), use

# a temporary directory to which other users have no access.

self.env['TMPDIR'] = workingdir.name

# Add a non-accessible directory to PATH (see gh-7053)

noaccess = Path(self.workingdir.name) / "_no_access_"

self.noaccess = noaccess.resolve()

noaccess.mkdir(0, exist_ok=True)

PATH = os.environ.get('PATH', '')

if PATH:

PATH = noaccess / PATH

else:

PATH = noaccess

self.env["PATH"] = str(PATH)

# From options:

if self.options.xunit:

self.add_xunit()

if self.options.coverage:

self.add_coverage()

self.env['IPTEST_SUBPROC_STREAMS'] = self.options.subproc_streams

self.cmd.extend(self.options.extra_args)

def cleanup(self):

"""

self.noaccess.chmod(stat.S_IRWXU)

TestController.cleanup(self)

@property

def will_run(self):

try:

return test_sections[self.section].will_run

except KeyError:

return True

def add_xunit(self):

xunit_file = Path(f"{self.section}.xunit.xml")

self.cmd.extend(["--with-xunit", "--xunit-file", xunit_file.absolute()])

def add_coverage(self):

try:

sources = test_sections[self.section].includes

except KeyError:

sources = ["IPython"]

coverage_rc = (

"[run]\n" "data_file = {data_file}\n" "source =\n" " {source}\n"

).format(

data_file=Path(f".coverage.{self.section}").absolute(),

source="\n ".join(sources),

)

config_file: Path = Path(self.workingdir.name) / ".coveragerc"

config_file.touch(exist_ok=True)

config_file.write_text(coverage_rc)

self.env["COVERAGE_PROCESS_START"] = str(config_file.resolve())

self.pycmd = "import coverage; coverage.process_startup(); " + self.pycmd

def launch(self, buffer_output=False):

self.cmd[2] = self.pycmd

"""Returns two lists of TestController instances, those to run, and those

testgroups = options.testgroups

if not testgroups:

testgroups = py_test_group_names

controllers = [PyTestController(name, options) for name in testgroups]

to_run = [c for c in controllers if c.will_run]

not_run = [c for c in controllers if not c.will_run]

"""Setup and run a test controller.

try:

try:

controller.setup()

controller.launch(buffer_output=buffer_output)

except Exception:

import traceback

traceback.print_exc()

return controller, 1 # signal failure

exitcode = controller.wait()

return controller, exitcode

except KeyboardInterrupt:

return controller, -signal.SIGINT

finally:

"""Return a string with a summary report of test-related variables."""

inf = get_sys_info()

out = []

def _add(name, value):

out.append((name, value))

_add('IPython version', inf['ipython_version'])

_add('IPython commit', "{} ({})".format(inf['commit_hash'], inf['commit_source']))

_add('IPython package', compress_user(inf['ipython_path']))

_add('Python version', inf['sys_version'].replace('\n',''))

_add('sys.executable', compress_user(inf['sys_executable']))

_add('Platform', inf['platform'])

width = max(len(n) for (n,v) in out)

out = ["{:<{width}}: {}\n".format(n, v, width=width) for (n,v) in out]

avail = []

not_avail = []

for k, is_avail in have.items():

if is_avail:

avail.append(k)

else:

not_avail.append(k)

if avail:

out.append('\nTools and libraries available at test time:\n')

avail.sort()

out.append(' ' + ' '.join(avail)+'\n')

if not_avail:

out.append('\nTools and libraries NOT available at test time:\n')

not_avail.sort()

out.append(' ' + ' '.join(not_avail)+'\n')

"""Run the entire IPython test suite by calling nose and trial.

to_run, not_run = prepare_controllers(options)

def justify(ltext, rtext, width=70, fill='-'):

ltext += ' '

rtext = (' ' + rtext).rjust(width - len(ltext), fill)

return ltext + rtext

# Run all test runners, tracking execution time

failed = []

t_start = time.time()

print()

if options.fast == 1:

# This actually means sequential, i.e. with 1 job

for controller in to_run:

print('Test group:', controller.section)

sys.stdout.flush() # Show in correct order when output is piped

controller, res = do_run(controller, buffer_output=False)

if res:

failed.append(controller)

if res == -signal.SIGINT:

print("Interrupted")

break

print()

else:

# Run tests concurrently

try:

pool = multiprocessing.pool.ThreadPool(options.fast)

for (controller, res) in pool.imap_unordered(do_run, to_run):

res_string = 'OK' if res == 0 else 'FAILED'

print(justify('Test group: ' + controller.section, res_string))

if res:

print(decode(controller.stdout))

failed.append(controller)

if res == -signal.SIGINT:

print("Interrupted")

break

except KeyboardInterrupt:

return

for controller in not_run:

print(justify('Test group: ' + controller.section, 'NOT RUN'))

t_end = time.time()

t_tests = t_end - t_start

nrunners = len(to_run)

nfail = len(failed)

# summarize results

print('_'*70)

print('Test suite completed for system with the following information:')

print(report())

took = "Took %.3fs." % t_tests

print('Status: ', end='')

if not failed:

print('OK (%d test groups).' % nrunners, took)

else:

# If anything went wrong, point out what command to rerun manually to

# see the actual errors and individual summary

failed_sections = [c.section for c in failed]

print('ERROR - {} out of {} test groups failed ({}).'.format(nfail,

nrunners, ', '.join(failed_sections)), took)

print()

print('You may wish to rerun these, with:')

print(' iptest', *failed_sections)

print()

if options.coverage:

from coverage import coverage, CoverageException

cov = coverage(data_file='.coverage')

cov.combine()

cov.save()

# Coverage HTML report

if options.coverage == 'html':

html_dir = 'ipy_htmlcov'

shutil.rmtree(html_dir, ignore_errors=True)

print("Writing HTML coverage report to %s/ ... " % html_dir, end="")

sys.stdout.flush()

# Custom HTML reporter to clean up module names.

from coverage.html import HtmlReporter

class CustomHtmlReporter(HtmlReporter):

def find_code_units(self, morfs):

super(CustomHtmlReporter, self).find_code_units(morfs)

for cu in self.code_units:

nameparts = cu.name.split(os.sep)

if 'IPython' not in nameparts:

continue

ix = nameparts.index('IPython')

cu.name = '.'.join(nameparts[ix:])

# Reimplement the html_report method with our custom reporter

cov.get_data()

cov.config.from_args(omit='*{0}tests{0}*'.format(os.sep), html_dir=html_dir,

html_title='IPython test coverage',

)

reporter = CustomHtmlReporter(cov, cov.config)

reporter.report(None)

print('done.')

# Coverage XML report

elif options.coverage == 'xml':

try:

cov.xml_report(outfile='ipy_coverage.xml')

except CoverageException as e:

print('Generating coverage report failed. Are you running javascript tests only?')

import traceback

traceback.print_exc()

if failed:

# Ensure that our exit code indicates failure

"""Get an argparse Namespace object with the default arguments, to pass to

options = argparser.parse_args([])

options.extra_args = []

# iptest doesn't work correctly if the working directory is the

# root of the IPython source tree. Tell the user to avoid

# frustration.

main_file = Path.cwd() / Path("IPython/testing/__main__.py")

if main_file.exists():

print("Don't run iptest from the IPython source directory", file=sys.stderr)

sys.exit(1)

# Arguments after -- should be passed through to nose. Argparse treats

# everything after -- as regular positional arguments, so we separate them

# first.

try:

ix = sys.argv.index('--')

except ValueError:

to_parse = sys.argv[1:]

extra_args = []

else:

to_parse = sys.argv[1:ix]

extra_args = sys.argv[ix+1:]

options = argparser.parse_args(to_parse)

options.extra_args = extra_args