aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/ipython/py2/IPython/testing/iptest.py
diff options
context:
space:
mode:
authorMikhail Borisov <borisov.mikhail@gmail.com>2022-02-10 16:45:40 +0300
committerDaniil Cherednik <dcherednik@yandex-team.ru>2022-02-10 16:45:40 +0300
commit5d50718e66d9c037dc587a0211110b7d25a66185 (patch)
treee98df59de24d2ef7c77baed9f41e4875a2fef972 /contrib/python/ipython/py2/IPython/testing/iptest.py
parenta6a92afe03e02795227d2641b49819b687f088f8 (diff)
downloadydb-5d50718e66d9c037dc587a0211110b7d25a66185.tar.gz
Restoring authorship annotation for Mikhail Borisov <borisov.mikhail@gmail.com>. Commit 2 of 2.
Diffstat (limited to 'contrib/python/ipython/py2/IPython/testing/iptest.py')
-rw-r--r--contrib/python/ipython/py2/IPython/testing/iptest.py856
1 files changed, 428 insertions, 428 deletions
diff --git a/contrib/python/ipython/py2/IPython/testing/iptest.py b/contrib/python/ipython/py2/IPython/testing/iptest.py
index 9a5a3d6f58..4018264125 100644
--- a/contrib/python/ipython/py2/IPython/testing/iptest.py
+++ b/contrib/python/ipython/py2/IPython/testing/iptest.py
@@ -1,55 +1,55 @@
-# -*- 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) IPython Development Team.
-# Distributed under the terms of the Modified BSD License.
-
-from __future__ import print_function
-
-import glob
-from io import BytesIO
-import os
-import os.path as path
-import sys
-from threading import Thread, Lock, Event
-import warnings
-
-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
-
-from IPython import version_info
-from IPython.utils.py3compat import bytes_to_str
-from IPython.utils.importstring import import_item
-from IPython.testing.plugin.ipdoctest import IPythonDoctest
-from IPython.external.decorators import KnownFailure, knownfailureif
-
-pjoin = path.join
-
-
-# Enable printing all warnings raise by IPython's modules
+# -*- 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) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
+
+from __future__ import print_function
+
+import glob
+from io import BytesIO
+import os
+import os.path as path
+import sys
+from threading import Thread, Lock, Event
+import warnings
+
+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
+
+from IPython import version_info
+from IPython.utils.py3compat import bytes_to_str
+from IPython.utils.importstring import import_item
+from IPython.testing.plugin.ipdoctest import IPythonDoctest
+from IPython.external.decorators import KnownFailure, knownfailureif
+
+pjoin = path.join
+
+
+# Enable printing all warnings raise by IPython's modules
warnings.filterwarnings('ignore', message='.*Matplotlib is building the font cache.*', category=UserWarning, module='.*')
if sys.version_info > (3,0):
warnings.filterwarnings('error', message='.*', category=ResourceWarning, module='.*')
warnings.filterwarnings('error', message=".*{'config': True}.*", category=DeprecationWarning, module='IPy.*')
-warnings.filterwarnings('default', message='.*', category=Warning, module='IPy.*')
-
+warnings.filterwarnings('default', message='.*', category=Warning, module='IPy.*')
+
warnings.filterwarnings('error', message='.*apply_wrapper.*', category=DeprecationWarning, module='.*')
warnings.filterwarnings('error', message='.*make_label_dec', category=DeprecationWarning, module='.*')
warnings.filterwarnings('error', message='.*decorated_dummy.*', category=DeprecationWarning, module='.*')
@@ -60,384 +60,384 @@ warnings.filterwarnings('error', message='.*disable_gui.*', category=Deprecation
warnings.filterwarnings('error', message='.*ExceptionColors global is deprecated.*', category=DeprecationWarning, module='.*')
-if version_info < (6,):
- # nose.tools renames all things from `camelCase` to `snake_case` which raise an
- # warning with the runner they also import from standard import library. (as of Dec 2015)
- # Ignore, let's revisit that in a couple of years for IPython 6.
- warnings.filterwarnings('ignore', message='.*Please use assertEqual instead', category=Warning, module='IPython.*')
-
-
-# ------------------------------------------------------------------------------
-# 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 = {'matplotlib': test_for('matplotlib'),
- 'pygments': test_for('pygments'),
- 'sqlite3': test_for('sqlite3')}
-
-#-----------------------------------------------------------------------------
-# Test suite definitions
-#-----------------------------------------------------------------------------
-
-test_group_names = ['core',
- 'extensions', 'lib', 'terminal', 'testing', 'utils',
- ]
-
-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']
-sec.exclude('kernel')
-if not have['pygments']:
- sec.exclude('tests.test_lexers')
-# 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')
-# We also do this unconditionally, because wx can interfere with Unix signals.
-# There are currently no tests for it anyway.
-sec.exclude('inputhookwx')
-# 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']
-# 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')
-
-# don't run jupyter_console tests found via shim
-test_sections['terminal'].exclude('console')
-
-# extensions:
-sec = test_sections['extensions']
-# This is deprecated in favour of rpy2
-sec.exclude('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')
-
-
-#-----------------------------------------------------------------------------
-# Functions and classes
-#-----------------------------------------------------------------------------
-
-def check_exclusions_exist():
- from IPython.paths import get_ipython_package_dir
+if version_info < (6,):
+ # nose.tools renames all things from `camelCase` to `snake_case` which raise an
+ # warning with the runner they also import from standard import library. (as of Dec 2015)
+ # Ignore, let's revisit that in a couple of years for IPython 6.
+ warnings.filterwarnings('ignore', message='.*Please use assertEqual instead', category=Warning, module='IPython.*')
+
+
+# ------------------------------------------------------------------------------
+# 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 = {'matplotlib': test_for('matplotlib'),
+ 'pygments': test_for('pygments'),
+ 'sqlite3': test_for('sqlite3')}
+
+#-----------------------------------------------------------------------------
+# Test suite definitions
+#-----------------------------------------------------------------------------
+
+test_group_names = ['core',
+ 'extensions', 'lib', 'terminal', 'testing', 'utils',
+ ]
+
+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']
+sec.exclude('kernel')
+if not have['pygments']:
+ sec.exclude('tests.test_lexers')
+# 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')
+# We also do this unconditionally, because wx can interfere with Unix signals.
+# There are currently no tests for it anyway.
+sec.exclude('inputhookwx')
+# 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']
+# 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')
+
+# don't run jupyter_console tests found via shim
+test_sections['terminal'].exclude('console')
+
+# extensions:
+sec = test_sections['extensions']
+# This is deprecated in favour of rpy2
+sec.exclude('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')
+
+
+#-----------------------------------------------------------------------------
+# Functions and classes
+#-----------------------------------------------------------------------------
+
+def check_exclusions_exist():
+ from IPython.paths import get_ipython_package_dir
from warnings 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):
- daemon = True # Don't hang if main thread crashes
- started = False
- def __init__(self, echo=False):
- super(StreamCapturer, self).__init__()
- self.echo = echo
- self.streams = []
- self.buffer = BytesIO()
- self.readfd, self.writefd = os.pipe()
- self.buffer_lock = Lock()
- self.stop = Event()
-
- def run(self):
- self.started = True
-
- while not self.stop.is_set():
- chunk = os.read(self.readfd, 1024)
-
- with self.buffer_lock:
- self.buffer.write(chunk)
- if self.echo:
- sys.stdout.write(bytes_to_str(chunk))
-
- 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()
-
- def halt(self):
- """Safely stop the thread."""
- if not self.started:
- return
-
- self.stop.set()
- os.write(self.writefd, b'\0') # Ensure we're not locked in a read()
- self.join()
-
-class SubprocessStreamCapturePlugin(Plugin):
- name='subprocstreams'
- def __init__(self):
- Plugin.__init__(self)
- self.stream_capturer = StreamCapturer()
- self.destination = os.environ.get('IPTEST_SUBPROC_STREAMS', 'capture')
- # This is ugly, but distant parts of the test machinery need to be able
- # to redirect streams, so we make the object globally accessible.
- nose.iptest_stdstreams_fileno = self.get_write_fileno
-
- def get_write_fileno(self):
- if self.destination == 'capture':
- self.stream_capturer.ensure_started()
- return self.stream_capturer.writefd
- elif self.destination == 'discard':
- return os.open(os.devnull, os.O_WRONLY)
- else:
- return sys.__stdout__.fileno()
-
- def configure(self, options, config):
- Plugin.configure(self, options, config)
- # Override nose trying to disable plugin.
- if self.destination == 'capture':
- 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):
- self.stream_capturer.halt()
-
-
-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()
-
- 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
- # 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')
-
- plugins = [ ExclusionPlugin(section.excludes), KnownFailure(),
- SubprocessStreamCapturePlugin() ]
-
- # we still have some vestigial doctests in core
- if (section.name.startswith(('core', 'IPython.core'))):
- plugins.append(IPythonDoctest())
- argv.extend([
- '--with-ipdoctest',
- '--ipdoctest-tests',
- '--ipdoctest-extension=txt',
- ])
-
-
- # 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()
+ 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):
+ daemon = True # Don't hang if main thread crashes
+ started = False
+ def __init__(self, echo=False):
+ super(StreamCapturer, self).__init__()
+ self.echo = echo
+ self.streams = []
+ self.buffer = BytesIO()
+ self.readfd, self.writefd = os.pipe()
+ self.buffer_lock = Lock()
+ self.stop = Event()
+
+ def run(self):
+ self.started = True
+
+ while not self.stop.is_set():
+ chunk = os.read(self.readfd, 1024)
+
+ with self.buffer_lock:
+ self.buffer.write(chunk)
+ if self.echo:
+ sys.stdout.write(bytes_to_str(chunk))
+
+ 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()
+
+ def halt(self):
+ """Safely stop the thread."""
+ if not self.started:
+ return
+
+ self.stop.set()
+ os.write(self.writefd, b'\0') # Ensure we're not locked in a read()
+ self.join()
+
+class SubprocessStreamCapturePlugin(Plugin):
+ name='subprocstreams'
+ def __init__(self):
+ Plugin.__init__(self)
+ self.stream_capturer = StreamCapturer()
+ self.destination = os.environ.get('IPTEST_SUBPROC_STREAMS', 'capture')
+ # This is ugly, but distant parts of the test machinery need to be able
+ # to redirect streams, so we make the object globally accessible.
+ nose.iptest_stdstreams_fileno = self.get_write_fileno
+
+ def get_write_fileno(self):
+ if self.destination == 'capture':
+ self.stream_capturer.ensure_started()
+ return self.stream_capturer.writefd
+ elif self.destination == 'discard':
+ return os.open(os.devnull, os.O_WRONLY)
+ else:
+ return sys.__stdout__.fileno()
+
+ def configure(self, options, config):
+ Plugin.configure(self, options, config)
+ # Override nose trying to disable plugin.
+ if self.destination == 'capture':
+ 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):
+ self.stream_capturer.halt()
+
+
+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()
+
+ 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
+ # 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')
+
+ plugins = [ ExclusionPlugin(section.excludes), KnownFailure(),
+ SubprocessStreamCapturePlugin() ]
+
+ # we still have some vestigial doctests in core
+ if (section.name.startswith(('core', 'IPython.core'))):
+ plugins.append(IPythonDoctest())
+ argv.extend([
+ '--with-ipdoctest',
+ '--ipdoctest-tests',
+ '--ipdoctest-extension=txt',
+ ])
+
+
+ # 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()