Simplify StreamCapturer for subprocess testing

Rather than using a transient pipe for each subprocess started, the
StreamCapturer now makes a single pipe, and subprocesses redirect their
output to it.

So long as this works on Windows (I've done brief testing, and os.pipe()
seems to be functional), this will hopefully make this much more robust.
The recent failures in ShiningPanda on IPython.parallel have been caused
by StreamCapturer.

IPython/kernel/tests/utils.py

@@ -1,176 +1,175 b''
 """utilities for testing IPython kernels"""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2013  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
 #-------------------------------------------------------------------------------
 
 import atexit
 
 from contextlib import contextmanager
 from subprocess import PIPE, STDOUT
 try:
     from queue import Empty  # Py 3
 except ImportError:
     from Queue import Empty  # Py 2
 
 import nose
 import nose.tools as nt
 
 from IPython.kernel import KernelManager
 
 #-------------------------------------------------------------------------------
 # Globals
 #-------------------------------------------------------------------------------
 
 STARTUP_TIMEOUT = 60
 TIMEOUT = 15
 
 KM = None
 KC = None
 
 #-------------------------------------------------------------------------------
 # code
 #-------------------------------------------------------------------------------
 
 
 def start_new_kernel(argv=None):
     """start a new kernel, and return its Manager and Client"""
     km = KernelManager()
-    kwargs = dict(stdout=PIPE, stderr=STDOUT)
+    kwargs = dict(stdout=nose.ipy_stream_capturer.writefd, stderr=STDOUT)
     if argv:
         kwargs['extra_arguments'] = argv
     km.start_kernel(**kwargs)
-    nose.ipy_stream_capturer.add_stream(km.kernel.stdout.fileno())
     nose.ipy_stream_capturer.ensure_started()
     kc = km.client()
     kc.start_channels()
     
     msg_id = kc.kernel_info()
     kc.get_shell_msg(block=True, timeout=STARTUP_TIMEOUT)
     flush_channels(kc)
     return km, kc
 
 def flush_channels(kc=None):
     """flush any messages waiting on the queue"""
     from .test_message_spec import validate_message
     
     if kc is None:
         kc = KC
     for channel in (kc.shell_channel, kc.iopub_channel):
         while True:
             try:
                 msg = channel.get_msg(block=True, timeout=0.1)
             except Empty:
                 break
             else:
                 validate_message(msg)
 
 
 def execute(code='', kc=None, **kwargs):
     """wrapper for doing common steps for validating an execution request"""
     from .test_message_spec import validate_message
     if kc is None:
         kc = KC
     msg_id = kc.execute(code=code, **kwargs)
     reply = kc.get_shell_msg(timeout=TIMEOUT)
     validate_message(reply, 'execute_reply', msg_id)
     busy = kc.get_iopub_msg(timeout=TIMEOUT)
     validate_message(busy, 'status', msg_id)
     nt.assert_equal(busy['content']['execution_state'], 'busy')
     
     if not kwargs.get('silent'):
         pyin = kc.get_iopub_msg(timeout=TIMEOUT)
         validate_message(pyin, 'pyin', msg_id)
         nt.assert_equal(pyin['content']['code'], code)
     
     return msg_id, reply['content']
 
 def start_global_kernel():
     """start the global kernel (if it isn't running) and return its client"""
     global KM, KC
     if KM is None:
         KM, KC = start_new_kernel()
         atexit.register(stop_global_kernel)
     return KC
 
 @contextmanager
 def kernel():
     """Context manager for the global kernel instance
     
     Should be used for most kernel tests
     
     Returns
     -------
     kernel_client: connected KernelClient instance
     """
     yield start_global_kernel()
 
 def uses_kernel(test_f):
     """Decorator for tests that use the global kernel"""
     def wrapped_test():
         with kernel() as kc:
             test_f(kc)
     wrapped_test.__doc__ = test_f.__doc__
     wrapped_test.__name__ = test_f.__name__
     return wrapped_test
 
 def stop_global_kernel():
     """Stop the global shared kernel instance, if it exists"""
     global KM, KC
     KC.stop_channels()
     KC = None
     if KM is None:
         return
     KM.shutdown_kernel(now=True)
     KM = None
 
 @contextmanager
 def new_kernel(argv=None):
     """Context manager for a new kernel in a subprocess
     
     Should only be used for tests where the kernel must not be re-used.
     
     Returns
     -------
     kernel_client: connected KernelClient instance
     """
     km, kc = start_new_kernel(argv)
     try:
         yield kc
     finally:
         kc.stop_channels()
         km.shutdown_kernel(now=True)
 
 
 def assemble_output(iopub):
     """assemble stdout/err from an execution"""
     stdout = ''
     stderr = ''
     while True:
         msg = iopub.get_msg(block=True, timeout=1)
         msg_type = msg['msg_type']
         content = msg['content']
         if msg_type == 'status' and content['execution_state'] == 'idle':
             # idle message signals end of output
             break
         elif msg['msg_type'] == 'stream':
             if content['name'] == 'stdout':
                 stdout += content['data']
             elif content['name'] == 'stderr':
                 stderr += content['data']
             else:
                 raise KeyError("bad stream: %r" % content['name'])
         else:
             # other output, ignored
             pass
     return stdout, stderr
 
 
 

IPython/parallel/tests/__init__.py

@@ -1,133 +1,131 b''
 """toplevel setup/teardown for parallel tests."""
 from __future__ import print_function
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 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
 #-------------------------------------------------------------------------------
 
 import os
 import tempfile
 import time
 from subprocess import Popen, PIPE, STDOUT
 
 import nose
 
 from IPython.utils.path import get_ipython_dir
 from IPython.parallel import Client
 from IPython.parallel.apps.launcher import (LocalProcessLauncher,
                                                   ipengine_cmd_argv,
                                                   ipcontroller_cmd_argv,
                                                   SIGKILL,
                                                   ProcessStateError,
 )
 
 # globals
 launchers = []
 blackhole = open(os.devnull, 'w')
 
 # Launcher class
 class TestProcessLauncher(LocalProcessLauncher):
     """subclass LocalProcessLauncher, to prevent extra sockets and threads being created on Windows"""
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
-                stdout=PIPE, stderr=STDOUT,
+                stdout=nose.ipy_stream_capturer.writefd, stderr=STDOUT,
                 env=os.environ,
                 cwd=self.work_dir
             )
             self.notify_start(self.process.pid)
             self.poll = self.process.poll
             # Store stdout & stderr to show with failing tests.
             # This is defined in IPython.testing.iptest
-            nose.ipy_stream_capturer.add_stream(self.process.stdout.fileno())
             nose.ipy_stream_capturer.ensure_started()
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
 
 # nose setup/teardown
 
 def setup():
     cluster_dir = os.path.join(get_ipython_dir(), 'profile_iptest')
     engine_json = os.path.join(cluster_dir, 'security', 'ipcontroller-engine.json')
     client_json = os.path.join(cluster_dir, 'security', 'ipcontroller-client.json')
     for json in (engine_json, client_json):
         if os.path.exists(json):
             os.remove(json)
     
     cp = TestProcessLauncher()
     cp.cmd_and_args = ipcontroller_cmd_argv + \
                 ['--profile=iptest', '--log-level=20', '--ping=250', '--dictdb']
     cp.start()
     launchers.append(cp)
     tic = time.time()
     while not os.path.exists(engine_json) or not os.path.exists(client_json):
         if cp.poll() is not None:
             raise RuntimeError("The test controller exited with status %s" % cp.poll())
         elif time.time()-tic > 15:
             raise RuntimeError("Timeout waiting for the test controller to start.")
         time.sleep(0.1)
     add_engines(1)
 
 def add_engines(n=1, profile='iptest', total=False):
     """add a number of engines to a given profile.
     
     If total is True, then already running engines are counted, and only
     the additional engines necessary (if any) are started.
     """
     rc = Client(profile=profile)
     base = len(rc)
     
     if total:
         n = max(n - base, 0)
     
     eps = []
     for i in range(n):
         ep = TestProcessLauncher()
         ep.cmd_and_args = ipengine_cmd_argv + [
             '--profile=%s' % profile,
             '--log-level=50',
             '--InteractiveShell.colors=nocolor'
             ]
         ep.start()
         launchers.append(ep)
         eps.append(ep)
     tic = time.time()
     while len(rc) < base+n:
         if any([ ep.poll() is not None for ep in eps ]):
             raise RuntimeError("A test engine failed to start.")
         elif time.time()-tic > 15:
             raise RuntimeError("Timeout waiting for engines to connect.")
         time.sleep(.1)
         rc.spin()
     rc.close()
     return eps
 
 def teardown():
     time.sleep(1)
     while launchers:
         p = launchers.pop()
-        nose.ipy_stream_capturer.remove_stream(p.process.stdout.fileno())
         if p.poll() is None:
             try:
                 p.stop()
             except Exception as e:
                 print(e)
                 pass
         if p.poll() is None:
             time.sleep(.25)
         if p.poll() is None:
             try:
                 print('cleaning up test process...')
                 p.signal(SIGKILL)
             except:
                 print("couldn't shutdown process: ", p)
     blackhole.close()
     

IPython/testing/iptest.py

@@ -1,539 +1,514 b''
 # -*- coding: utf-8 -*-
 """IPython Test Suite Runner.
 
 This module provides a main entry point to a user script to test IPython
 itself from the command line. There are two ways of running this script:
 
 1. With the syntax `iptest all`.  This runs our entire test suite by
    calling this script (with different arguments) recursively.  This
    causes modules and package to be tested in different processes, using nose
    or trial where appropriate.
 2. With the regular nose syntax, like `iptest -vvs IPython`.  In this form
    the script simply calls nose, but with special command line flags and
    plugins loaded.
 
 """
 
 #-----------------------------------------------------------------------------
 #  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
 
 # Stdlib
 import glob
 from io import BytesIO
 import os
 import os.path as path
 from select import select
 import sys
 from threading import Thread, Lock, Event
 import warnings
 
 # Now, proceed to import nose itself
 import nose.plugins.builtin
 from nose.plugins.xunit import Xunit
 from nose import SkipTest
 from nose.core import TestProgram
 from nose.plugins import Plugin
 from nose.util import safe_str
 
 # Our own imports
 from IPython.utils.process import is_cmd_found
 from IPython.utils.importstring import import_item
 from IPython.testing.plugin.ipdoctest import IPythonDoctest
 from IPython.external.decorators import KnownFailure, knownfailureif
 
 pjoin = path.join
 
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 
 #-----------------------------------------------------------------------------
 # Warnings control
 #-----------------------------------------------------------------------------
 
 # Twisted generates annoying warnings with Python 2.6, as will do other code
 # that imports 'sets' as of today
 warnings.filterwarnings('ignore', 'the sets module is deprecated',
                         DeprecationWarning )
 
 # This one also comes from Twisted
 warnings.filterwarnings('ignore', 'the sha module is deprecated',
                         DeprecationWarning)
 
 # Wx on Fedora11 spits these out
 warnings.filterwarnings('ignore', 'wxPython/wxWidgets release number mismatch',
                         UserWarning)
 
 # ------------------------------------------------------------------------------
 # Monkeypatch Xunit to count known failures as skipped.
 # ------------------------------------------------------------------------------
 def monkeypatch_xunit():
     try:
         knownfailureif(True)(lambda: None)()
     except Exception as e:
         KnownFailureTest = type(e)
 
     def addError(self, test, err, capt=None):
         if issubclass(err[0], KnownFailureTest):
             err = (SkipTest,) + err[1:]
         return self.orig_addError(test, err, capt)
 
     Xunit.orig_addError = Xunit.addError
     Xunit.addError = addError
 
 #-----------------------------------------------------------------------------
 # Check which dependencies are installed and greater than minimum version.
 #-----------------------------------------------------------------------------
 def extract_version(mod):
     return mod.__version__
 
 def test_for(item, min_version=None, callback=extract_version):
     """Test to see if item is importable, and optionally check against a minimum
     version.
 
     If min_version is given, the default behavior is to check against the
     `__version__` attribute of the item, but specifying `callback` allows you to
     extract the value you are interested in. e.g::
 
         In [1]: import sys
 
         In [2]: from IPython.testing.iptest import test_for
 
         In [3]: test_for('sys', (2,6), callback=lambda sys: sys.version_info)
         Out[3]: True
 
     """
     try:
         check = import_item(item)
     except (ImportError, RuntimeError):
         # GTK reports Runtime error if it can't be initialized even if it's
         # importable.
         return False
     else:
         if min_version:
             if callback:
                 # extra processing step to get version to compare
                 check = callback(check)
 
             return check >= min_version
         else:
             return True
 
 # Global dict where we can store information on what we have and what we don't
 # have available at test run time
 have = {}
 
 have['curses'] = test_for('_curses')
 have['matplotlib'] = test_for('matplotlib')
 have['numpy'] = test_for('numpy')
 have['pexpect'] = test_for('IPython.external.pexpect')
 have['pymongo'] = test_for('pymongo')
 have['pygments'] = test_for('pygments')
 have['qt'] = test_for('IPython.external.qt')
 have['rpy2'] = test_for('rpy2')
 have['sqlite3'] = test_for('sqlite3')
 have['cython'] = test_for('Cython')
 have['oct2py'] = test_for('oct2py')
 have['tornado'] = test_for('tornado.version_info', (3,1,0), callback=None)
 have['jinja2'] = test_for('jinja2')
 have['wx'] = test_for('wx')
 have['wx.aui'] = test_for('wx.aui')
 have['azure'] = test_for('azure')
 have['requests'] = test_for('requests')
 have['sphinx'] = test_for('sphinx')
 have['casperjs'] = is_cmd_found('casperjs')
 
 min_zmq = (2,1,11)
 
 have['zmq'] = test_for('zmq.pyzmq_version_info', min_zmq, callback=lambda x: x())
 
 #-----------------------------------------------------------------------------
 # Test suite definitions
 #-----------------------------------------------------------------------------
 
 test_group_names = ['parallel', 'kernel', 'kernel.inprocess', 'config', 'core',
                     'extensions', 'lib', 'terminal', 'testing', 'utils',
                     'nbformat', 'qt', 'html', 'nbconvert'
                    ]
 
 class TestSection(object):
     def __init__(self, name, includes):
         self.name = name
         self.includes = includes
         self.excludes = []
         self.dependencies = []
         self.enabled = True
     
     def exclude(self, module):
         if not module.startswith('IPython'):
             module = self.includes[0] + "." + module
         self.excludes.append(module.replace('.', os.sep))
     
     def requires(self, *packages):
         self.dependencies.extend(packages)
     
     @property
     def will_run(self):
         return self.enabled and all(have[p] for p in self.dependencies)
 
 # Name -> (include, exclude, dependencies_met)
 test_sections = {n:TestSection(n, ['IPython.%s' % n]) for n in test_group_names}
 
 # Exclusions and dependencies
 # ---------------------------
 
 # core:
 sec = test_sections['core']
 if not have['sqlite3']:
     sec.exclude('tests.test_history')
     sec.exclude('history')
 if not have['matplotlib']:
     sec.exclude('pylabtools'),
     sec.exclude('tests.test_pylabtools')
 
 # lib:
 sec = test_sections['lib']
 if not have['wx']:
     sec.exclude('inputhookwx')
 if not have['pexpect']:
     sec.exclude('irunner')
     sec.exclude('tests.test_irunner')
 if not have['zmq']:
     sec.exclude('kernel')
 # We do this unconditionally, so that the test suite doesn't import
 # gtk, changing the default encoding and masking some unicode bugs.
 sec.exclude('inputhookgtk')
 # Testing inputhook will need a lot of thought, to figure out
 # how to have tests that don't lock up with the gui event
 # loops in the picture
 sec.exclude('inputhook')
 
 # testing:
 sec = test_sections['testing']
 # This guy is probably attic material
 sec.exclude('mkdoctests')
 # These have to be skipped on win32 because they use echo, rm, cd, etc.
 # See ticket https://github.com/ipython/ipython/issues/87
 if sys.platform == 'win32':
     sec.exclude('plugin.test_exampleip')
     sec.exclude('plugin.dtexample')
 
 # terminal:
 if (not have['pexpect']) or (not have['zmq']):
     test_sections['terminal'].exclude('console')
 
 # parallel
 sec = test_sections['parallel']
 sec.requires('zmq')
 if not have['pymongo']:
     sec.exclude('controller.mongodb')
     sec.exclude('tests.test_mongodb')
 
 # kernel:
 sec = test_sections['kernel']
 sec.requires('zmq')
 # The in-process kernel tests are done in a separate section
 sec.exclude('inprocess')
 # importing gtk sets the default encoding, which we want to avoid
 sec.exclude('zmq.gui.gtkembed')
 if not have['matplotlib']:
     sec.exclude('zmq.pylab')
 
 # kernel.inprocess:
 test_sections['kernel.inprocess'].requires('zmq')
 
 # extensions:
 sec = test_sections['extensions']
 if not have['cython']:
     sec.exclude('cythonmagic')
     sec.exclude('tests.test_cythonmagic')
 if not have['oct2py']:
     sec.exclude('octavemagic')
     sec.exclude('tests.test_octavemagic')
 if not have['rpy2'] or not have['numpy']:
     sec.exclude('rmagic')
     sec.exclude('tests.test_rmagic')
 # autoreload does some strange stuff, so move it to its own test section
 sec.exclude('autoreload')
 sec.exclude('tests.test_autoreload')
 test_sections['autoreload'] = TestSection('autoreload',
         ['IPython.extensions.autoreload', 'IPython.extensions.tests.test_autoreload'])
 test_group_names.append('autoreload')
 
 # qt:
 test_sections['qt'].requires('zmq', 'qt', 'pygments')
 
 # html:
 sec = test_sections['html']
 sec.requires('zmq', 'tornado', 'requests')
 # The notebook 'static' directory contains JS, css and other
 # files for web serving.  Occasionally projects may put a .py
 # file in there (MathJax ships a conf.py), so we might as
 # well play it safe and skip the whole thing.
 sec.exclude('static')
 sec.exclude('fabfile')
 if not have['jinja2']:
     sec.exclude('notebookapp')
 if not have['azure']:
     sec.exclude('services.notebooks.azurenbmanager')
 
 # config:
 # Config files aren't really importable stand-alone
 test_sections['config'].exclude('profile')
 
 # nbconvert:
 sec = test_sections['nbconvert']
 sec.requires('pygments', 'jinja2', 'sphinx')
 # Exclude nbconvert directories containing config files used to test.
 # Executing the config files with iptest would cause an exception.
 sec.exclude('tests.files')
 sec.exclude('exporters.tests.files')
 if not have['tornado']:
     sec.exclude('nbconvert.post_processors.serve')
     sec.exclude('nbconvert.post_processors.tests.test_serve')
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 def check_exclusions_exist():
     from IPython.utils.path import get_ipython_package_dir
     from IPython.utils.warn import warn
     parent = os.path.dirname(get_ipython_package_dir())
     for sec in test_sections:
         for pattern in sec.exclusions:
             fullpath = pjoin(parent, pattern)
             if not os.path.exists(fullpath) and not glob.glob(fullpath + '.*'):
                 warn("Excluding nonexistent file: %r" % pattern)
 
 
 class ExclusionPlugin(Plugin):
     """A nose plugin to effect our exclusions of files and directories.
     """
     name = 'exclusions'
     score = 3000  # Should come before any other plugins
     
     def __init__(self, exclude_patterns=None):
         """
         Parameters
         ----------
 
         exclude_patterns : sequence of strings, optional
           Filenames containing these patterns (as raw strings, not as regular
           expressions) are excluded from the tests.
         """
         self.exclude_patterns = exclude_patterns or []
         super(ExclusionPlugin, self).__init__()
 
     def options(self, parser, env=os.environ):
         Plugin.options(self, parser, env)
     
     def configure(self, options, config):
         Plugin.configure(self, options, config)
         # Override nose trying to disable plugin.
         self.enabled = True
         
     def wantFile(self, filename):
         """Return whether the given filename should be scanned for tests.
         """
         if any(pat in filename for pat in self.exclude_patterns):
             return False
         return None
 
     def wantDirectory(self, directory):
         """Return whether the given directory should be scanned for tests.
         """
         if any(pat in directory for pat in self.exclude_patterns):
             return False
         return None
 
 
 class StreamCapturer(Thread):
     started = False
     def __init__(self):
         super(StreamCapturer, self).__init__()
         self.streams = []
         self.buffer = BytesIO()
-        self.streams_lock = Lock()
+        self.readfd, self.writefd = os.pipe()
         self.buffer_lock = Lock()
-        self.stream_added = Event()
         self.stop = Event()
 
     def run(self):
         self.started = True
-        while not self.stop.is_set():
-            with self.streams_lock:
-                streams = self.streams
 
-            if not streams:
-                self.stream_added.wait(timeout=1)
-                self.stream_added.clear()
-                continue
+        while not self.stop.is_set():
+            ready = select([self.readfd], [], [], 1)[0]
 
-            ready = select(streams, [], [], 0.5)[0]
-            dead = []
+            if ready:
             with self.buffer_lock:
-                for fd in ready:
-                    try:
-                        self.buffer.write(os.read(fd, 1024))
-                    except OSError as e:
-                        import errno
-                        if e.errno == errno.EBADF:
-                            dead.append(fd)
-                        else:
-                            raise
-
-            with self.streams_lock:
-                for fd in dead:
-                    self.streams.remove(fd)
-    
-    def add_stream(self, fd):
-        with self.streams_lock:
-            self.streams.append(fd)
-        self.stream_added.set()
+                    self.buffer.write(os.read(self.readfd, 1024))
     
-    def remove_stream(self, fd):
-        with self.streams_lock:
-            self.streams.remove(fd)
+        os.close(self.readfd)
+        os.close(self.writefd)
         
     def reset_buffer(self):
         with self.buffer_lock:
             self.buffer.truncate(0)
             self.buffer.seek(0)
     
     def get_buffer(self):
         with self.buffer_lock:
             return self.buffer.getvalue()
     
     def ensure_started(self):
         if not self.started:
             self.start()
 
 class SubprocessStreamCapturePlugin(Plugin):
     name='subprocstreams'
     def __init__(self):
         Plugin.__init__(self)
         self.stream_capturer = StreamCapturer()
         # This is ugly, but distant parts of the test machinery need to be able
-        # to add streams, so we make the object globally accessible.
+        # to redirect streams, so we make the object globally accessible.
         nose.ipy_stream_capturer = self.stream_capturer
     
     def configure(self, options, config):
         Plugin.configure(self, options, config)
         # Override nose trying to disable plugin.
         self.enabled = True
     
     def startTest(self, test):
         # Reset log capture
         self.stream_capturer.reset_buffer()
     
     def formatFailure(self, test, err):
         # Show output
         ec, ev, tb = err
         captured = self.stream_capturer.get_buffer().decode('utf-8', 'replace')
         if captured.strip():
             ev = safe_str(ev)
             out = [ev, '>> begin captured subprocess output <<',
                     captured,
                     '>> end captured subprocess output <<']
             return ec, '\n'.join(out), tb
 
         return err
     
     formatError = formatFailure
     
     def finalize(self, result):
         if self.stream_capturer.started:
             self.stream_capturer.stop.set()
             self.stream_capturer.join()
 
 
 def run_iptest():
     """Run the IPython test suite using nose.
 
     This function is called when this script is **not** called with the form
     `iptest all`.  It simply calls nose with appropriate command line flags
     and accepts all of the standard nose arguments.
     """
     # Apply our monkeypatch to Xunit
     if '--with-xunit' in sys.argv and not hasattr(Xunit, 'orig_addError'):
         monkeypatch_xunit()
 
     warnings.filterwarnings('ignore',
         'This will be removed soon.  Use IPython.testing.util instead')
     
     arg1 = sys.argv[1]
     if arg1 in test_sections:
         section = test_sections[arg1]
         sys.argv[1:2] = section.includes
     elif arg1.startswith('IPython.') and arg1[8:] in test_sections:
         section = test_sections[arg1[8:]]
         sys.argv[1:2] = section.includes
     else:
         section = TestSection(arg1, includes=[arg1])
         
 
     argv = sys.argv + [ '--detailed-errors',  # extra info in tracebacks
 
                         '--with-ipdoctest',
                         '--ipdoctest-tests','--ipdoctest-extension=txt',
 
                         # We add --exe because of setuptools' imbecility (it
                         # blindly does chmod +x on ALL files).  Nose does the
                         # right thing and it tries to avoid executables,
                         # setuptools unfortunately forces our hand here.  This
                         # has been discussed on the distutils list and the
                         # setuptools devs refuse to fix this problem!
                         '--exe',
                         ]
     if '-a' not in argv and '-A' not in argv:
         argv = argv + ['-a', '!crash']
 
     if nose.__version__ >= '0.11':
         # I don't fully understand why we need this one, but depending on what
         # directory the test suite is run from, if we don't give it, 0 tests
         # get run.  Specifically, if the test suite is run from the source dir
         # with an argument (like 'iptest.py IPython.core', 0 tests are run,
         # even if the same call done in this directory works fine).  It appears
         # that if the requested package is in the current dir, nose bails early
         # by default.  Since it's otherwise harmless, leave it in by default
         # for nose >= 0.11, though unfortunately nose 0.10 doesn't support it.
         argv.append('--traverse-namespace')
 
     # use our plugin for doctesting.  It will remove the standard doctest plugin
     # if it finds it enabled
     plugins = [ExclusionPlugin(section.excludes), IPythonDoctest(), KnownFailure(),
                SubprocessStreamCapturePlugin() ]
     
     # Use working directory set by parent process (see iptestcontroller)
     if 'IPTEST_WORKING_DIR' in os.environ:
         os.chdir(os.environ['IPTEST_WORKING_DIR'])
     
     # We need a global ipython running in this process, but the special
     # in-process group spawns its own IPython kernels, so for *that* group we
     # must avoid also opening the global one (otherwise there's a conflict of
     # singletons).  Ultimately the solution to this problem is to refactor our
     # assumptions about what needs to be a singleton and what doesn't (app
     # objects should, individual shells shouldn't).  But for now, this
     # workaround allows the test suite for the inprocess module to complete.
     if 'kernel.inprocess' not in section.name:
         from IPython.testing import globalipapp
         globalipapp.start_ipython()
 
     # Now nose can run
     TestProgram(argv=argv, addplugins=plugins)
 
 if __name__ == '__main__':
     run_iptest()