Split out iptestcontroller to control test process.

IPython/testing/iptestcontroller.py

@@ -0,0 +1,319 b''
+# -*- coding: utf-8 -*-
+"""IPython Test Process Controller
+
+This module runs one or more subprocesses which will actually run the IPython
+test suite.
+
+"""
+
+#-----------------------------------------------------------------------------
+#  Copyright (C) 2009-2011  The IPython Development Team
+#
+#  Distributed under the terms of the BSD License.  The full license is in
+#  the file COPYING, distributed as part of this software.
+#-----------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+from __future__ import print_function
+
+import multiprocessing.pool
+import os
+import signal
+import sys
+import subprocess
+import tempfile
+import time
+
+from .iptest import have, special_test_suites
+from IPython.utils import py3compat
+from IPython.utils.path import get_ipython_module_path
+from IPython.utils.process import pycmd2argv
+from IPython.utils.sysinfo import sys_info
+from IPython.utils.tempdir import TemporaryDirectory
+
+
+class IPTester(object):
+    """Call that calls iptest or trial in a subprocess.
+    """
+    #: string, name of test runner that will be called
+    runner = None
+    #: list, parameters for test runner
+    params = None
+    #: list, arguments of system call to be made to call test runner
+    call_args = None
+    #: list, subprocesses we start (for cleanup)
+    processes = None
+    #: str, coverage xml output file
+    coverage_xml = None
+    buffer_output = False
+
+    def __init__(self, runner='iptest', params=None):
+        """Create new test runner."""
+        if runner == 'iptest':
+            iptest_app = os.path.abspath(get_ipython_module_path('IPython.testing.iptest'))
+            self.runner = pycmd2argv(iptest_app) + sys.argv[1:]
+        else:
+            raise Exception('Not a valid test runner: %s' % repr(runner))
+        if params is None:
+            params = []
+        if isinstance(params, str):
+            params = [params]
+        self.params = params
+
+        # Assemble call
+        self.call_args = self.runner+self.params
+        
+        # Find the section we're testing (IPython.foo)
+        for sect in self.params:
+            if sect.startswith('IPython') or sect in special_test_suites: break
+        else:
+            raise ValueError("Section not found", self.params)
+        
+        if '--with-xunit' in self.call_args:
+            
+            self.call_args.append('--xunit-file')
+            # FIXME: when Windows uses subprocess.call, these extra quotes are unnecessary:
+            xunit_file = os.path.abspath(sect+'.xunit.xml')
+            if sys.platform == 'win32':
+                xunit_file = '"%s"' % xunit_file
+            self.call_args.append(xunit_file)
+        
+        if '--with-xml-coverage' in self.call_args:
+            self.coverage_xml = os.path.abspath(sect+".coverage.xml")
+            self.call_args.remove('--with-xml-coverage')
+            self.call_args = ["coverage", "run", "--source="+sect] + self.call_args[1:]
+
+        # Store anything we start to clean up on deletion
+        self.processes = []
+
+    def _run_cmd(self):
+        with TemporaryDirectory() as IPYTHONDIR:
+            env = os.environ.copy()
+            env['IPYTHONDIR'] = IPYTHONDIR
+            # print >> sys.stderr, '*** CMD:', ' '.join(self.call_args) # dbg
+            output = subprocess.PIPE if self.buffer_output else None
+            subp = subprocess.Popen(self.call_args, stdout=output,
+                    stderr=output, env=env)
+            self.processes.append(subp)
+            # If this fails, the process will be left in self.processes and
+            # cleaned up later, but if the wait call succeeds, then we can
+            # clear the stored process.
+            retcode = subp.wait()
+            self.processes.pop()
+            self.stdout = subp.stdout
+            self.stderr = subp.stderr
+            return retcode
+
+    def run(self):
+        """Run the stored commands"""
+        try:
+            retcode = self._run_cmd()
+        except KeyboardInterrupt:
+            return -signal.SIGINT
+        except:
+            import traceback
+            traceback.print_exc()
+            return 1  # signal failure
+        
+        if self.coverage_xml:
+            subprocess.call(["coverage", "xml", "-o", self.coverage_xml])
+        return retcode
+
+    def __del__(self):
+        """Cleanup on exit by killing any leftover processes."""
+        for subp in self.processes:
+            if subp.poll() is not None:
+                continue # process 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 make_runners(inc_slow=False):
+    """Define the top-level packages that need to be tested.
+    """
+
+    # Packages to be tested via nose, that only depend on the stdlib
+    nose_pkg_names = ['config', 'core', 'extensions', 'lib', 'terminal',
+                      'testing', 'utils', 'nbformat']
+
+    if have['qt']:
+        nose_pkg_names.append('qt')
+
+    if have['tornado']:
+        nose_pkg_names.append('html')
+        
+    if have['zmq']:
+        nose_pkg_names.insert(0, 'kernel')
+        nose_pkg_names.insert(1, 'kernel.inprocess')
+        if inc_slow:
+            nose_pkg_names.insert(0, 'parallel')
+
+    if all((have['pygments'], have['jinja2'], have['sphinx'])):
+        nose_pkg_names.append('nbconvert')
+
+    # For debugging this code, only load quick stuff
+    #nose_pkg_names = ['core', 'extensions']  # dbg
+
+    # Make fully qualified package names prepending 'IPython.' to our name lists
+    nose_packages = ['IPython.%s' % m for m in nose_pkg_names ]
+
+    # Make runners
+    runners = [ (v, IPTester('iptest', params=v)) for v in nose_packages ]
+    
+    for name in special_test_suites:
+        runners.append((name, IPTester('iptest', params=name)))
+
+    return runners
+
+def do_run(x):
+    print('IPython test group:',x[0])
+    ret = x[1].run()
+    return ret
+
+def report():
+    """Return a string with a summary report of test-related variables."""
+
+    out = [ sys_info(), '\n']
+
+    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')
+
+    return ''.join(out)
+
+def run_iptestall(inc_slow=False, fast=False):
+    """Run the entire IPython test suite by calling nose and trial.
+
+    This function constructs :class:`IPTester` instances for all IPython
+    modules and package and then runs each of them.  This causes the modules
+    and packages of IPython to be tested each in their own subprocess using
+    nose.
+    
+    Parameters
+    ----------
+    
+    inc_slow : bool, optional
+      Include slow tests, like IPython.parallel. By default, these tests aren't
+      run.
+
+    fast : bool, option
+      Run the test suite in parallel, if True, using as many threads as there
+      are processors
+    """
+    if fast:
+        p = multiprocessing.pool.ThreadPool()
+    else:
+        p = multiprocessing.pool.ThreadPool(1)
+
+    runners = make_runners(inc_slow=inc_slow)
+
+    # Run the test runners in a temporary dir so we can nuke it when finished
+    # to clean up any junk files left over by accident.  This also makes it
+    # robust against being run in non-writeable directories by mistake, as the
+    # temp dir will always be user-writeable.
+    curdir = os.getcwdu()
+    testdir = tempfile.gettempdir()
+    os.chdir(testdir)
+
+    # Run all test runners, tracking execution time
+    failed = []
+    t_start = time.time()
+
+    try:
+        all_res = p.map(do_run, runners)
+        print('*'*70)
+        for ((name, runner), res) in zip(runners, all_res):
+            tgroup = 'IPython test group: ' + name
+            res_string = 'OK' if res == 0 else 'FAILED'
+            res_string = res_string.rjust(70 - len(tgroup), '.')
+            print(tgroup + res_string)
+            if res:
+                failed.append( (name, runner) )
+                if res == -signal.SIGINT:
+                    print("Interrupted")
+                    break
+    finally:
+        os.chdir(curdir)
+    t_end = time.time()
+    t_tests = t_end - t_start
+    nrunners = len(runners)
+    nfail = len(failed)
+    # summarize results
+    print()
+    print('*'*70)
+    print('Test suite completed for system with the following information:')
+    print(report())
+    print('Ran %s test groups in %.3fs' % (nrunners, t_tests))
+    print()
+    print('Status:')
+    if not failed:
+        print('OK')
+    else:
+        # If anything went wrong, point out what command to rerun manually to
+        # see the actual errors and individual summary
+        print('ERROR - %s out of %s test groups failed.' % (nfail, nrunners))
+        for name, failed_runner in failed:
+            print('-'*40)
+            print('Runner failed:',name)
+            print('You may wish to rerun this one individually, with:')
+            failed_call_args = [py3compat.cast_unicode(x) for x in failed_runner.call_args]
+            print(u' '.join(failed_call_args))
+            print()
+        # Ensure that our exit code indicates failure
+        sys.exit(1)
+
+
+def main():
+    for arg in sys.argv[1:]:
+        if arg.startswith('IPython') or arg in special_test_suites:
+            from .iptest import run_iptest
+            # This is in-process
+            run_iptest()
+    else:
+        inc_slow =  "--all" in sys.argv
+        if inc_slow:
+            sys.argv.remove("--all")
+
+        fast =  "--fast" in sys.argv
+        if fast:
+            sys.argv.remove("--fast")
+            IPTester.buffer_output = True
+
+        # This starts subprocesses
+        run_iptestall(inc_slow=inc_slow, fast=fast)
+
+
+if __name__ == '__main__':
+    main()

IPython/scripts/iptest

@@ -20,5 +20,5 b' Exiting."""'
     import sys
     print >> sys.stderr, error
 else:
-    from IPython.testing import iptest
-    iptest.main()
+    from IPython.testing import iptestcontroller
+    iptestcontroller.main()

IPython/testing/iptest.py

@@ -30,13 +30,8 b' from __future__ import print_function'
 import glob
 import os
 import os.path as path
-import signal
 import sys
-import subprocess
-import tempfile
-import time
 import warnings
-import multiprocessing.pool
 
 # Now, proceed to import nose itself
 import nose.plugins.builtin
@@ -45,12 +40,8 b' from nose import SkipTest'
 from nose.core import TestProgram
 
 # Our own imports
-from IPython.utils import py3compat
 from IPython.utils.importstring import import_item
-from IPython.utils.path import get_ipython_module_path, get_ipython_package_dir
-from IPython.utils.process import pycmd2argv
-from IPython.utils.sysinfo import sys_info
-from IPython.utils.tempdir import TemporaryDirectory
+from IPython.utils.path import get_ipython_package_dir
 from IPython.utils.warn import warn
 
 from IPython.testing import globalipapp
@@ -167,32 +158,6 b" have['zmq'] = test_for('zmq.pyzmq_version_info', min_zmq, callback=lambda x: x()"
 # Functions and classes
 #-----------------------------------------------------------------------------
 
-def report():
-    """Return a string with a summary report of test-related variables."""
-
-    out = [ sys_info(), '\n']
-
-    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')
-
-    return ''.join(out)
-
 
 def make_exclude():
     """Make patterns of modules and packages to exclude from testing.
@@ -325,160 +290,9 b' def make_exclude():'
 
     return exclusions
 
-
-class IPTester(object):
-    """Call that calls iptest or trial in a subprocess.
-    """
-    #: string, name of test runner that will be called
-    runner = None
-    #: list, parameters for test runner
-    params = None
-    #: list, arguments of system call to be made to call test runner
-    call_args = None
-    #: list, subprocesses we start (for cleanup)
-    processes = None
-    #: str, coverage xml output file
-    coverage_xml = None
-    buffer_output = False
-
-    def __init__(self, runner='iptest', params=None):
-        """Create new test runner."""
-        p = os.path
-        if runner == 'iptest':
-            iptest_app = os.path.abspath(get_ipython_module_path('IPython.testing.iptest'))
-            self.runner = pycmd2argv(iptest_app) + sys.argv[1:]
-        else:
-            raise Exception('Not a valid test runner: %s' % repr(runner))
-        if params is None:
-            params = []
-        if isinstance(params, str):
-            params = [params]
-        self.params = params
-
-        # Assemble call
-        self.call_args = self.runner+self.params
-        
-        # Find the section we're testing (IPython.foo)
-        for sect in self.params:
-            if sect.startswith('IPython') or sect in special_test_suites: break
-        else:
-            raise ValueError("Section not found", self.params)
-        
-        if '--with-xunit' in self.call_args:
-            
-            self.call_args.append('--xunit-file')
-            # FIXME: when Windows uses subprocess.call, these extra quotes are unnecessary:
-            xunit_file = path.abspath(sect+'.xunit.xml')
-            if sys.platform == 'win32':
-                xunit_file = '"%s"' % xunit_file
-            self.call_args.append(xunit_file)
-        
-        if '--with-xml-coverage' in self.call_args:
-            self.coverage_xml = path.abspath(sect+".coverage.xml")
-            self.call_args.remove('--with-xml-coverage')
-            self.call_args = ["coverage", "run", "--source="+sect] + self.call_args[1:]
-
-        # Store anything we start to clean up on deletion
-        self.processes = []
-
-    def _run_cmd(self):
-        with TemporaryDirectory() as IPYTHONDIR:
-            env = os.environ.copy()
-            env['IPYTHONDIR'] = IPYTHONDIR
-            # print >> sys.stderr, '*** CMD:', ' '.join(self.call_args) # dbg
-            output = subprocess.PIPE if self.buffer_output else None
-            subp = subprocess.Popen(self.call_args, stdout=output,
-                    stderr=output, env=env)
-            self.processes.append(subp)
-            # If this fails, the process will be left in self.processes and
-            # cleaned up later, but if the wait call succeeds, then we can
-            # clear the stored process.
-            retcode = subp.wait()
-            self.processes.pop()
-            self.stdout = subp.stdout
-            self.stderr = subp.stderr
-            return retcode
-
-    def run(self):
-        """Run the stored commands"""
-        try:
-            retcode = self._run_cmd()
-        except KeyboardInterrupt:
-            return -signal.SIGINT
-        except:
-            import traceback
-            traceback.print_exc()
-            return 1  # signal failure
-        
-        if self.coverage_xml:
-            subprocess.call(["coverage", "xml", "-o", self.coverage_xml])
-        return retcode
-
-    def __del__(self):
-        """Cleanup on exit by killing any leftover processes."""
-        for subp in self.processes:
-            if subp.poll() is not None:
-                continue # process 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.')
-
-
 special_test_suites = {
     'autoreload': ['IPython.extensions.autoreload', 'IPython.extensions.tests.test_autoreload'],
 }
-                
-def make_runners(inc_slow=False):
-    """Define the top-level packages that need to be tested.
-    """
-
-    # Packages to be tested via nose, that only depend on the stdlib
-    nose_pkg_names = ['config', 'core', 'extensions', 'lib', 'terminal',
-                      'testing', 'utils', 'nbformat']
-
-    if have['qt']:
-        nose_pkg_names.append('qt')
-
-    if have['tornado']:
-        nose_pkg_names.append('html')
-        
-    if have['zmq']:
-        nose_pkg_names.insert(0, 'kernel')
-        nose_pkg_names.insert(1, 'kernel.inprocess')
-        if inc_slow:
-            nose_pkg_names.insert(0, 'parallel')
-
-    if all((have['pygments'], have['jinja2'], have['sphinx'])):
-        nose_pkg_names.append('nbconvert')
-
-    # For debugging this code, only load quick stuff
-    #nose_pkg_names = ['core', 'extensions']  # dbg
-
-    # Make fully qualified package names prepending 'IPython.' to our name lists
-    nose_packages = ['IPython.%s' % m for m in nose_pkg_names ]
-
-    # Make runners
-    runners = [ (v, IPTester('iptest', params=v)) for v in nose_packages ]
-    
-    for name in special_test_suites:
-        runners.append((name, IPTester('iptest', params=name)))
-
-    return runners
 
 
 def run_iptest():
@@ -546,111 +360,6 b' def run_iptest():'
     # Now nose can run
     TestProgram(argv=argv, addplugins=plugins)
 
-def do_run(x):
-    print('IPython test group:',x[0])
-    ret = x[1].run()
-    return ret
-
-def run_iptestall(inc_slow=False, fast=False):
-    """Run the entire IPython test suite by calling nose and trial.
-
-    This function constructs :class:`IPTester` instances for all IPython
-    modules and package and then runs each of them.  This causes the modules
-    and packages of IPython to be tested each in their own subprocess using
-    nose.
-    
-    Parameters
-    ----------
-    
-    inc_slow : bool, optional
-      Include slow tests, like IPython.parallel. By default, these tests aren't
-      run.
-
-    fast : bool, option
-      Run the test suite in parallel, if True, using as many threads as there
-      are processors
-    """
-    if fast:
-        p = multiprocessing.pool.ThreadPool()
-    else:
-        p = multiprocessing.pool.ThreadPool(1)
-
-    runners = make_runners(inc_slow=inc_slow)
-
-    # Run the test runners in a temporary dir so we can nuke it when finished
-    # to clean up any junk files left over by accident.  This also makes it
-    # robust against being run in non-writeable directories by mistake, as the
-    # temp dir will always be user-writeable.
-    curdir = os.getcwdu()
-    testdir = tempfile.gettempdir()
-    os.chdir(testdir)
-
-    # Run all test runners, tracking execution time
-    failed = []
-    t_start = time.time()
-
-    try:
-        all_res = p.map(do_run, runners)
-        print('*'*70)
-        for ((name, runner), res) in zip(runners, all_res):
-            tgroup = 'IPython test group: ' + name
-            res_string = 'OK' if res == 0 else 'FAILED'
-            res_string = res_string.rjust(70 - len(tgroup), '.')
-            print(tgroup + res_string)
-            if res:
-                failed.append( (name, runner) )
-                if res == -signal.SIGINT:
-                    print("Interrupted")
-                    break
-    finally:
-        os.chdir(curdir)
-    t_end = time.time()
-    t_tests = t_end - t_start
-    nrunners = len(runners)
-    nfail = len(failed)
-    # summarize results
-    print()
-    print('*'*70)
-    print('Test suite completed for system with the following information:')
-    print(report())
-    print('Ran %s test groups in %.3fs' % (nrunners, t_tests))
-    print()
-    print('Status:')
-    if not failed:
-        print('OK')
-    else:
-        # If anything went wrong, point out what command to rerun manually to
-        # see the actual errors and individual summary
-        print('ERROR - %s out of %s test groups failed.' % (nfail, nrunners))
-        for name, failed_runner in failed:
-            print('-'*40)
-            print('Runner failed:',name)
-            print('You may wish to rerun this one individually, with:')
-            failed_call_args = [py3compat.cast_unicode(x) for x in failed_runner.call_args]
-            print(u' '.join(failed_call_args))
-            print()
-        # Ensure that our exit code indicates failure
-        sys.exit(1)
-
-
-def main():
-    for arg in sys.argv[1:]:
-        if arg.startswith('IPython') or arg in special_test_suites:
-            # This is in-process
-            run_iptest()
-    else:
-        inc_slow =  "--all" in sys.argv
-        if inc_slow:
-            sys.argv.remove("--all")
-
-        fast =  "--fast" in sys.argv
-        if fast:
-            sys.argv.remove("--fast")
-            IPTester.buffer_output = True
-
-        # This starts subprocesses
-        run_iptestall(inc_slow=inc_slow, fast=fast)
-
 
 if __name__ == '__main__':
-    main()
+    run_iptest()