Move strip_ansi function to utils

IPython/nbconvert/filters/ansi.py

@@ -1,178 +1,166 b''
 """Filters for processing ANSI colors within Jinja templates.
 """
 #-----------------------------------------------------------------------------
 # Copyright (c) 2013, the IPython Development Team.
 #
 # Distributed under the terms of the Modified BSD License.
 #
 # The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import re
 from IPython.utils import coloransi
+from IPython.utils.text import strip_ansi
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 __all__ = [
     'strip_ansi',
     'ansi2html',
     'single_ansi2latex',
     'ansi2latex'
 ]
 
-def strip_ansi(source):
-    """
-    Remove ansi from text
-    
-    Parameters
-    ----------
-    source : str
-        Source to remove the ansi from
-    """
-    
-    return re.sub(r'\033\[(\d|;)+?m', '', source)
-
-
 ansi_colormap = {
     '30': 'ansiblack',
     '31': 'ansired',
     '32': 'ansigreen',
     '33': 'ansiyellow',
     '34': 'ansiblue',
     '35': 'ansipurple',
     '36': 'ansicyan',
     '37': 'ansigrey',
     '01': 'ansibold',
 }
 
 html_escapes = {
     '<': '&lt;',
     '>': '&gt;',
     "'": '&apos;',
     '"': '&quot;',
     '`': '&#96;',
 }
 ansi_re = re.compile('\x1b' + r'\[([\dA-Fa-f;]*?)m')
 
 def ansi2html(text):
     """
     Convert ansi colors to html colors.
     
     Parameters
     ----------
     text : str
         Text containing ansi colors to convert to html
     """
 
     # do ampersand first
     text = text.replace('&', '&amp;')
     
     for c, escape in html_escapes.items():
         text = text.replace(c, escape)
 
     m = ansi_re.search(text)
     opened = False
     cmds = []
     opener = ''
     closer = ''
     while m:
         cmds = m.groups()[0].split(';')
         closer = '</span>' if opened else ''
         
         # True if there is there more than one element in cmds, *or*
         # if there is only one but it is not equal to a string of zeroes.
         opened = len(cmds) > 1 or cmds[0] != '0' * len(cmds[0])
         classes = []
         for cmd in cmds:
             if cmd in ansi_colormap:
                 classes.append(ansi_colormap[cmd])
 
         if classes:
             opener = '<span class="%s">' % (' '.join(classes))
         else:
             opener = ''
         text = re.sub(ansi_re, closer + opener, text, 1)
 
         m = ansi_re.search(text)
 
     if opened:
         text += '</span>'
     return text
 
 
 def single_ansi2latex(code):
     """Converts single ansi markup to latex format.
 
     Return latex code and number of open brackets.
 
     Accepts codes like '\x1b[1;32m' (bold, red) and the short form '\x1b[32m' (red)
 
     Colors are matched to those defined in coloransi, which defines colors
     using the 0, 1 (bold) and 5 (blinking) styles. Styles 1 and 5 are
     interpreted as bold. All other styles are mapped to 0. Note that in
     coloransi, a style of 1 does not just mean bold; for example, Brown is
     "0;33", but Yellow is "1;33". An empty string is returned for unrecognised
     codes and the "reset" code '\x1b[m'.
     """
     components = code.split(';')
     if len(components) > 1:
         # Style is digits after '['
         style = int(components[0].split('[')[-1])
         color = components[1][:-1]
     else:
         style = 0
         color = components[0][-3:-1]
         
     # If the style is not normal (0), bold (1) or blinking (5) then treat it as normal
     if style not in [0, 1, 5]:
         style = 0
 
     for name, tcode in coloransi.color_templates:
         tstyle, tcolor = tcode.split(';')
         tstyle = int(tstyle)
         if tstyle == style and tcolor == color:
             break
     else:
         return '', 0
 
     if style == 5:
         name = name[5:]                             # BlinkRed -> Red, etc
     name = name.lower()
 
     if style in [1, 5]:
         return r'\textbf{\color{'+name+'}', 1
     else:
         return r'{\color{'+name+'}', 1
 
 def ansi2latex(text):
     """Converts ansi formated text to latex version
 
     based on https://bitbucket.org/birkenfeld/sphinx-contrib/ansi.py
     """
     color_pattern = re.compile('\x1b\\[([^m]*)m')
     last_end = 0
     openbrack = 0
     outstring = ''
     for match in color_pattern.finditer(text):
         head = text[last_end:match.start()]
         outstring += head
         if openbrack:
             outstring += '}'*openbrack
             openbrack = 0
         code = match.group()
         if not (code == coloransi.TermColors.Normal or openbrack):
             texform, openbrack = single_ansi2latex(code)
             outstring += texform
         last_end = match.end()
     
     # Add the remainer of the string and THEN close any remaining color brackets.
     outstring += text[last_end:]
     if openbrack: 
         outstring += '}'*openbrack
     return outstring.strip()

IPython/testing/iptestcontroller.py

@@ -1,672 +1,672 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) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import argparse
 import json
 import multiprocessing.pool
 import os
 import shutil
 import signal
 import sys
 import subprocess
 import time
 import re
 
 from .iptest import have, test_group_names as py_test_group_names, test_sections, StreamCapturer
 from IPython.utils.path import compress_user
 from IPython.utils.py3compat import bytes_to_str
 from IPython.utils.sysinfo import get_sys_info
 from IPython.utils.tempdir import TemporaryDirectory
-from IPython.nbconvert.filters.ansi import strip_ansi
+from IPython.utils.text import strip_ansi
 
 try:
     # Python >= 3.3
     from subprocess import TimeoutExpired
     def popen_wait(p, timeout):
         return p.wait(timeout)
 except ImportError:
     class TimeoutExpired(Exception):
         pass
     def popen_wait(p, timeout):
         """backport of Popen.wait from Python 3"""
         for i in range(int(10 * timeout)):
             if p.poll() is not None:
                 return
             time.sleep(0.1)
         if p.poll() is None:
             raise TimeoutExpired
 
 NOTEBOOK_SHUTDOWN_TIMEOUT = 10
 
 class TestController(object):
     """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 = None
     #: 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.
         
         This is only called when we know the test group will be run. Things
         created here may be cleaned up by self.cleanup().
         """
         pass
 
     def launch(self, buffer_output=False):
         # print('*** ENV:', self.env)  # dbg
         # print('*** CMD:', self.cmd)  # dbg
         env = os.environ.copy()
         env.update(self.env)
         output = subprocess.PIPE if buffer_output else None
         stdout = subprocess.STDOUT if buffer_output else None
         self.process = subprocess.Popen(self.cmd, stdout=output,
                 stderr=stdout, env=env)
 
     def wait(self):
         self.stdout, _ = self.process.communicate()
         return self.process.returncode
 
     def print_extra_info(self):
         """Print extra information about this test run.
         
         If we're running in parallel and showing the concise view, this is only
         called if the test group fails. Otherwise, it's called before the test
         group is started.
         
         The base implementation does nothing, but it can be overridden by
         subclasses.
         """
         return
 
     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()
 
     __del__ = cleanup
 
 
 class PyTestController(TestController):
     """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
 
         # 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)
 
     @property
     def will_run(self):
         try:
             return test_sections[self.section].will_run
         except KeyError:
             return True
 
     def add_xunit(self):
         xunit_file = os.path.abspath(self.section + '.xunit.xml')
         self.cmd.extend(['--with-xunit', '--xunit-file', xunit_file])
 
     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=os.path.abspath('.coverage.'+self.section),
                                source="\n  ".join(sources))
         config_file = os.path.join(self.workingdir.name, '.coveragerc')
         with open(config_file, 'w') as f:
             f.write(coverage_rc)
 
         self.env['COVERAGE_PROCESS_START'] = config_file
         self.pycmd = "import coverage; coverage.process_startup(); " + self.pycmd
 
     def launch(self, buffer_output=False):
         self.cmd[2] = self.pycmd
         super(PyTestController, self).launch(buffer_output=buffer_output)
 
 
 js_prefix = 'js/'
 
 def get_js_test_dir():
     import IPython.html.tests as t
     return os.path.join(os.path.dirname(t.__file__), '')
 
 def all_js_groups():
     import glob
     test_dir = get_js_test_dir()
     all_subdirs = glob.glob(test_dir + '[!_]*/')
     return [js_prefix+os.path.relpath(x, test_dir) for x in all_subdirs]
 
 class JSController(TestController):
     """Run CasperJS tests """
     requirements =  ['zmq', 'tornado', 'jinja2', 'casperjs', 'sqlite3']
     display_slimer_output = False
 
     def __init__(self, section, enabled=True, engine='phantomjs'):
         """Create new test runner."""
         TestController.__init__(self)
         self.engine = engine
         self.section = section
         self.enabled = enabled
         self.slimer_failure = re.compile('^FAIL.*', flags=re.MULTILINE)
         js_test_dir = get_js_test_dir()
         includes = '--includes=' + os.path.join(js_test_dir,'util.js')
         test_cases = os.path.join(js_test_dir, self.section[len(js_prefix):])
         self.cmd = ['casperjs', 'test', includes, test_cases, '--engine=%s' % self.engine]
 
     def setup(self):
         self.ipydir = TemporaryDirectory()
         self.nbdir = TemporaryDirectory()
         self.dirs.append(self.ipydir)
         self.dirs.append(self.nbdir)
         os.makedirs(os.path.join(self.nbdir.name, os.path.join(u'sub ∂ir1', u'sub ∂ir 1a')))
         os.makedirs(os.path.join(self.nbdir.name, os.path.join(u'sub ∂ir2', u'sub ∂ir 1b')))
         
         # start the ipython notebook, so we get the port number
         self.server_port = 0
         self._init_server()
         if self.server_port:
             self.cmd.append("--port=%i" % self.server_port)
         else:
             # don't launch tests if the server didn't start
             self.cmd = [sys.executable, '-c', 'raise SystemExit(1)']
 
     def launch(self, buffer_output):
         # If the engine is SlimerJS, we need to buffer the output because
         # SlimerJS does not support exit codes, so CasperJS always returns 0.
         if self.engine == 'slimerjs' and not buffer_output:
             self.display_slimer_output = True
             return super(JSController, self).launch(buffer_output=True)
 
         else:
             return super(JSController, self).launch(buffer_output=buffer_output)
 
     def wait(self, *pargs, **kwargs):
         """Wait for the JSController to finish"""
         ret = super(JSController, self).wait(*pargs, **kwargs)
         # If this is a SlimerJS controller, check the captured stdout for
         # errors.  Otherwise, just return the return code.
         if self.engine == 'slimerjs':
             stdout = bytes_to_str(self.stdout)
             if self.display_slimer_output:
                 print(stdout)
             if ret != 0:
                 # This could still happen e.g. if it's stopped by SIGINT
                 return ret
             return bool(self.slimer_failure.search(strip_ansi(stdout)))
         else:
             return ret
 
     def print_extra_info(self):
         print("Running tests with notebook directory %r" % self.nbdir.name)
 
     @property
     def will_run(self):
         return self.enabled and all(have[a] for a in self.requirements + [self.engine])
 
     def _init_server(self):
         "Start the notebook server in a separate process"
         self.server_command = command = [sys.executable,
             '-m', 'IPython.html',
             '--no-browser',
             '--ipython-dir', self.ipydir.name,
             '--notebook-dir', self.nbdir.name,
         ]
         # ipc doesn't work on Windows, and darwin has crazy-long temp paths,
         # which run afoul of ipc's maximum path length.
         if sys.platform.startswith('linux'):
             command.append('--KernelManager.transport=ipc')
         self.stream_capturer = c = StreamCapturer()
         c.start()
         self.server = subprocess.Popen(command, stdout=c.writefd, stderr=subprocess.STDOUT)
         self.server_info_file = os.path.join(self.ipydir.name,
             'profile_default', 'security', 'nbserver-%i.json' % self.server.pid
         )
         self._wait_for_server()
     
     def _wait_for_server(self):
         """Wait 30 seconds for the notebook server to start"""
         for i in range(300):
             if self.server.poll() is not None:
                 return self._failed_to_start()
             if os.path.exists(self.server_info_file):
                 try:
                     self._load_server_info()
                 except ValueError:
                     # If the server is halfway through writing the file, we may
                     # get invalid JSON; it should be ready next iteration.
                     pass
                 else:
                     return
             time.sleep(0.1)
         print("Notebook server-info file never arrived: %s" % self.server_info_file,
             file=sys.stderr
         )
     
     def _failed_to_start(self):
         """Notebook server exited prematurely"""
         captured = self.stream_capturer.get_buffer().decode('utf-8', 'replace')
         print("Notebook failed to start: ", file=sys.stderr)
         print(self.server_command)
         print(captured, file=sys.stderr)
     
     def _load_server_info(self):
         """Notebook server started, load connection info from JSON"""
         with open(self.server_info_file) as f:
             info = json.load(f)
         self.server_port = info['port']
 
     def cleanup(self):
         try:
             self.server.terminate()
         except OSError:
             # already dead
             pass
         # wait 10s for the server to shutdown
         try:
             popen_wait(self.server, NOTEBOOK_SHUTDOWN_TIMEOUT)
         except TimeoutExpired:
             # server didn't terminate, kill it
             try:
                 print("Failed to terminate notebook server, killing it.",
                     file=sys.stderr
                 )
                 self.server.kill()
             except OSError:
                 # already dead
                 pass
         # wait another 10s
         try:
             popen_wait(self.server, NOTEBOOK_SHUTDOWN_TIMEOUT)
         except TimeoutExpired:
             print("Notebook server still running (%s)" % self.server_info_file,
                 file=sys.stderr
             )
             
         self.stream_capturer.halt()
         TestController.cleanup(self)
 
 
 def prepare_controllers(options):
     """Returns two lists of TestController instances, those to run, and those
     not to run."""
     testgroups = options.testgroups
     if testgroups:
         if 'js' in testgroups:
             js_testgroups = all_js_groups()
         else:
             js_testgroups = [g for g in testgroups if g.startswith(js_prefix)]
         
         py_testgroups = [g for g in testgroups if g not in ['js'] + js_testgroups]
     else:
         py_testgroups = py_test_group_names
         if not options.all:
             js_testgroups = []
             test_sections['parallel'].enabled = False
         else:
             js_testgroups = all_js_groups()
 
     engine = 'slimerjs' if options.slimerjs else 'phantomjs'
     c_js = [JSController(name, engine=engine) for name in js_testgroups]
     c_py = [PyTestController(name, options) for name in py_testgroups]
 
     controllers = c_py + c_js
     to_run = [c for c in controllers if c.will_run]
     not_run = [c for c in controllers if not c.will_run]
     return to_run, not_run
 
 def do_run(controller, buffer_output=True):
     """Setup and run a test controller.
     
     If buffer_output is True, no output is displayed, to avoid it appearing
     interleaved. In this case, the caller is responsible for displaying test
     output on failure.
     
     Returns
     -------
     controller : TestController
       The same controller as passed in, as a convenience for using map() type
       APIs.
     exitcode : int
       The exit code of the test subprocess. Non-zero indicates failure.
     """
     try:
         try:
             controller.setup()
             if not buffer_output:
                 controller.print_extra_info()
             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:
         controller.cleanup()
 
 def report():
     """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')
 
     return ''.join(out)
 
 def run_iptestall(options):
     """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
     ----------
 
     All parameters are passed as attributes of the options object.
 
     testgroups : list of str
       Run only these sections of the test suite. If empty, run all the available
       sections.
 
     fast : int or None
       Run the test suite in parallel, using n simultaneous processes. If None
       is passed, one process is used per CPU core. Default 1 (i.e. sequential)
 
     inc_slow : bool
       Include slow tests, like IPython.parallel. By default, these tests aren't
       run.
 
     slimerjs : bool
       Use slimerjs if it's installed instead of phantomjs for casperjs tests.
 
     xunit : bool
       Produce Xunit XML output. This is written to multiple foo.xunit.xml files.
 
     coverage : bool or str
       Measure code coverage from tests. True will store the raw coverage data,
       or pass 'html' or 'xml' to get reports.
 
     extra_args : list
       Extra arguments to pass to the test subprocesses, e.g. '-v'
     """
     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:
                     controller.print_extra_info()
                     print(bytes_to_str(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
         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._harvest_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':
             cov.xml_report(outfile='ipy_coverage.xml')
 
     if failed:
         # Ensure that our exit code indicates failure
         sys.exit(1)
 
 argparser = argparse.ArgumentParser(description='Run IPython test suite')
 argparser.add_argument('testgroups', nargs='*',
                     help='Run specified groups of tests. If omitted, run '
                     'all tests.')
 argparser.add_argument('--all', action='store_true',
                     help='Include slow tests not run by default.')
 argparser.add_argument('--slimerjs', action='store_true',
                     help="Use slimerjs if it's installed instead of phantomjs for casperjs tests.")
 argparser.add_argument('-j', '--fast', nargs='?', const=None, default=1, type=int,
                     help='Run test sections in parallel. This starts as many '
                     'processes as you have cores, or you can specify a number.')
 argparser.add_argument('--xunit', action='store_true',
                     help='Produce Xunit XML results')
 argparser.add_argument('--coverage', nargs='?', const=True, default=False,
                     help="Measure test coverage. Specify 'html' or "
                     "'xml' to get reports.")
 argparser.add_argument('--subproc-streams', default='capture',
                     help="What to do with stdout/stderr from subprocesses. "
                     "'capture' (default), 'show' and 'discard' are the options.")
 
 def default_options():
     """Get an argparse Namespace object with the default arguments, to pass to
     :func:`run_iptestall`.
     """
     options = argparser.parse_args([])
     options.extra_args = []
     return options
 
 def main():
     # iptest doesn't work correctly if the working directory is the
     # root of the IPython source tree. Tell the user to avoid
     # frustration.
     if os.path.exists(os.path.join(os.getcwd(),
                                    'IPython', 'testing', '__main__.py')):
         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
 
     run_iptestall(options)
 
 
 if __name__ == '__main__':
     main()

IPython/utils/text.py

@@ -1,773 +1,784 b''
 # encoding: utf-8
 """
 Utilities for working with strings and text.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.utils.text
    :parts: 3
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-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 re
 import sys
 import textwrap
 from string import Formatter
 
 from IPython.external.path import path
 from IPython.testing.skipdoctest import skip_doctest_py3, skip_doctest
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Declarations
 #-----------------------------------------------------------------------------
 
 # datetime.strftime date format for ipython
 if sys.platform == 'win32':
     date_format = "%B %d, %Y"
 else:
     date_format = "%B %-d, %Y"
 
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 class LSString(str):
     """String derivative with a special access attributes.
 
     These are normal strings, but with the special attributes:
 
         .l (or .list) : value as list (split on newlines).
         .n (or .nlstr): original value (the string itself).
         .s (or .spstr): value as whitespace-separated string.
         .p (or .paths): list of path objects
 
     Any values which require transformations are computed only once and
     cached.
 
     Such strings are very useful to efficiently interact with the shell, which
     typically only understands whitespace-separated options for commands."""
 
     def get_list(self):
         try:
             return self.__list
         except AttributeError:
             self.__list = self.split('\n')
             return self.__list
 
     l = list = property(get_list)
 
     def get_spstr(self):
         try:
             return self.__spstr
         except AttributeError:
             self.__spstr = self.replace('\n',' ')
             return self.__spstr
 
     s = spstr = property(get_spstr)
 
     def get_nlstr(self):
         return self
 
     n = nlstr = property(get_nlstr)
 
     def get_paths(self):
         try:
             return self.__paths
         except AttributeError:
             self.__paths = [path(p) for p in self.split('\n') if os.path.exists(p)]
             return self.__paths
 
     p = paths = property(get_paths)
 
 # FIXME: We need to reimplement type specific displayhook and then add this
 # back as a custom printer. This should also be moved outside utils into the
 # core.
 
 # def print_lsstring(arg):
 #     """ Prettier (non-repr-like) and more informative printer for LSString """
 #     print "LSString (.p, .n, .l, .s available). Value:"
 #     print arg
 #
 #
 # print_lsstring = result_display.when_type(LSString)(print_lsstring)
 
 
 class SList(list):
     """List derivative with a special access attributes.
 
     These are normal lists, but with the special attributes:
 
     * .l (or .list) : value as list (the list itself).
     * .n (or .nlstr): value as a string, joined on newlines.
     * .s (or .spstr): value as a string, joined on spaces.
     * .p (or .paths): list of path objects
 
     Any values which require transformations are computed only once and
     cached."""
 
     def get_list(self):
         return self
 
     l = list = property(get_list)
 
     def get_spstr(self):
         try:
             return self.__spstr
         except AttributeError:
             self.__spstr = ' '.join(self)
             return self.__spstr
 
     s = spstr = property(get_spstr)
 
     def get_nlstr(self):
         try:
             return self.__nlstr
         except AttributeError:
             self.__nlstr = '\n'.join(self)
             return self.__nlstr
 
     n = nlstr = property(get_nlstr)
 
     def get_paths(self):
         try:
             return self.__paths
         except AttributeError:
             self.__paths = [path(p) for p in self if os.path.exists(p)]
             return self.__paths
 
     p = paths = property(get_paths)
 
     def grep(self, pattern, prune = False, field = None):
         """ Return all strings matching 'pattern' (a regex or callable)
 
         This is case-insensitive. If prune is true, return all items
         NOT matching the pattern.
 
         If field is specified, the match must occur in the specified
         whitespace-separated field.
 
         Examples::
 
             a.grep( lambda x: x.startswith('C') )
             a.grep('Cha.*log', prune=1)
             a.grep('chm', field=-1)
         """
 
         def match_target(s):
             if field is None:
                 return s
             parts = s.split()
             try:
                 tgt = parts[field]
                 return tgt
             except IndexError:
                 return ""
 
         if isinstance(pattern, py3compat.string_types):
             pred = lambda x : re.search(pattern, x, re.IGNORECASE)
         else:
             pred = pattern
         if not prune:
             return SList([el for el in self if pred(match_target(el))])
         else:
             return SList([el for el in self if not pred(match_target(el))])
 
     def fields(self, *fields):
         """ Collect whitespace-separated fields from string list
 
         Allows quick awk-like usage of string lists.
 
         Example data (in var a, created by 'a = !ls -l')::
 
             -rwxrwxrwx  1 ville None      18 Dec 14  2006 ChangeLog
             drwxrwxrwx+ 6 ville None       0 Oct 24 18:05 IPython
 
         * ``a.fields(0)`` is ``['-rwxrwxrwx', 'drwxrwxrwx+']``
         * ``a.fields(1,0)`` is ``['1 -rwxrwxrwx', '6 drwxrwxrwx+']``
           (note the joining by space).
         * ``a.fields(-1)`` is ``['ChangeLog', 'IPython']``
 
         IndexErrors are ignored.
 
         Without args, fields() just split()'s the strings.
         """
         if len(fields) == 0:
             return [el.split() for el in self]
 
         res = SList()
         for el in [f.split() for f in self]:
             lineparts = []
 
             for fd in fields:
                 try:
                     lineparts.append(el[fd])
                 except IndexError:
                     pass
             if lineparts:
                 res.append(" ".join(lineparts))
 
         return res
 
     def sort(self,field= None,  nums = False):
         """ sort by specified fields (see fields())
 
         Example::
 
             a.sort(1, nums = True)
 
         Sorts a by second field, in numerical order (so that 21 > 3)
 
         """
 
         #decorate, sort, undecorate
         if field is not None:
             dsu = [[SList([line]).fields(field),  line] for line in self]
         else:
             dsu = [[line,  line] for line in self]
         if nums:
             for i in range(len(dsu)):
                 numstr = "".join([ch for ch in dsu[i][0] if ch.isdigit()])
                 try:
                     n = int(numstr)
                 except ValueError:
                     n = 0;
                 dsu[i][0] = n
 
 
         dsu.sort()
         return SList([t[1] for t in dsu])
 
 
 # FIXME: We need to reimplement type specific displayhook and then add this
 # back as a custom printer. This should also be moved outside utils into the
 # core.
 
 # def print_slist(arg):
 #     """ Prettier (non-repr-like) and more informative printer for SList """
 #     print "SList (.p, .n, .l, .s, .grep(), .fields(), sort() available):"
 #     if hasattr(arg,  'hideonce') and arg.hideonce:
 #         arg.hideonce = False
 #         return
 #
 #     nlprint(arg)   # This was a nested list printer, now removed.
 #
 # print_slist = result_display.when_type(SList)(print_slist)
 
 
 def indent(instr,nspaces=4, ntabs=0, flatten=False):
     """Indent a string a given number of spaces or tabstops.
 
     indent(str,nspaces=4,ntabs=0) -> indent str by ntabs+nspaces.
 
     Parameters
     ----------
 
     instr : basestring
         The string to be indented.
     nspaces : int (default: 4)
         The number of spaces to be indented.
     ntabs : int (default: 0)
         The number of tabs to be indented.
     flatten : bool (default: False)
         Whether to scrub existing indentation.  If True, all lines will be
         aligned to the same indentation.  If False, existing indentation will
         be strictly increased.
 
     Returns
     -------
 
     str|unicode : string indented by ntabs and nspaces.
 
     """
     if instr is None:
         return
     ind = '\t'*ntabs+' '*nspaces
     if flatten:
         pat = re.compile(r'^\s*', re.MULTILINE)
     else:
         pat = re.compile(r'^', re.MULTILINE)
     outstr = re.sub(pat, ind, instr)
     if outstr.endswith(os.linesep+ind):
         return outstr[:-len(ind)]
     else:
         return outstr
 
 
 def list_strings(arg):
     """Always return a list of strings, given a string or list of strings
     as input.
 
     Examples
     --------
     ::
 
         In [7]: list_strings('A single string')
         Out[7]: ['A single string']
 
         In [8]: list_strings(['A single string in a list'])
         Out[8]: ['A single string in a list']
 
         In [9]: list_strings(['A','list','of','strings'])
         Out[9]: ['A', 'list', 'of', 'strings']
     """
 
     if isinstance(arg, py3compat.string_types): return [arg]
     else: return arg
 
 
 def marquee(txt='',width=78,mark='*'):
     """Return the input string centered in a 'marquee'.
 
     Examples
     --------
     ::
 
         In [16]: marquee('A test',40)
         Out[16]: '**************** A test ****************'
 
         In [17]: marquee('A test',40,'-')
         Out[17]: '---------------- A test ----------------'
 
         In [18]: marquee('A test',40,' ')
         Out[18]: '                 A test                 '
 
     """
     if not txt:
         return (mark*width)[:width]
     nmark = (width-len(txt)-2)//len(mark)//2
     if nmark < 0: nmark =0
     marks = mark*nmark
     return '%s %s %s' % (marks,txt,marks)
 
 
 ini_spaces_re = re.compile(r'^(\s+)')
 
 def num_ini_spaces(strng):
     """Return the number of initial spaces in a string"""
 
     ini_spaces = ini_spaces_re.match(strng)
     if ini_spaces:
         return ini_spaces.end()
     else:
         return 0
 
 
 def format_screen(strng):
     """Format a string for screen printing.
 
     This removes some latex-type format codes."""
     # Paragraph continue
     par_re = re.compile(r'\\$',re.MULTILINE)
     strng = par_re.sub('',strng)
     return strng
 
 
 def dedent(text):
     """Equivalent of textwrap.dedent that ignores unindented first line.
 
     This means it will still dedent strings like:
     '''foo
     is a bar
     '''
 
     For use in wrap_paragraphs.
     """
 
     if text.startswith('\n'):
         # text starts with blank line, don't ignore the first line
         return textwrap.dedent(text)
 
     # split first line
     splits = text.split('\n',1)
     if len(splits) == 1:
         # only one line
         return textwrap.dedent(text)
 
     first, rest = splits
     # dedent everything but the first line
     rest = textwrap.dedent(rest)
     return '\n'.join([first, rest])
 
 
 def wrap_paragraphs(text, ncols=80):
     """Wrap multiple paragraphs to fit a specified width.
 
     This is equivalent to textwrap.wrap, but with support for multiple
     paragraphs, as separated by empty lines.
 
     Returns
     -------
 
     list of complete paragraphs, wrapped to fill `ncols` columns.
     """
     paragraph_re = re.compile(r'\n(\s*\n)+', re.MULTILINE)
     text = dedent(text).strip()
     paragraphs = paragraph_re.split(text)[::2] # every other entry is space
     out_ps = []
     indent_re = re.compile(r'\n\s+', re.MULTILINE)
     for p in paragraphs:
         # presume indentation that survives dedent is meaningful formatting,
         # so don't fill unless text is flush.
         if indent_re.search(p) is None:
             # wrap paragraph
             p = textwrap.fill(p, ncols)
         out_ps.append(p)
     return out_ps
 
 
 def long_substr(data):
     """Return the longest common substring in a list of strings.
     
     Credit: http://stackoverflow.com/questions/2892931/longest-common-substring-from-more-than-two-strings-python
     """
     substr = ''
     if len(data) > 1 and len(data[0]) > 0:
         for i in range(len(data[0])):
             for j in range(len(data[0])-i+1):
                 if j > len(substr) and all(data[0][i:i+j] in x for x in data):
                     substr = data[0][i:i+j]
     elif len(data) == 1:
         substr = data[0]
     return substr
 
 
 def strip_email_quotes(text):
     """Strip leading email quotation characters ('>').
 
     Removes any combination of leading '>' interspersed with whitespace that
     appears *identically* in all lines of the input text.
 
     Parameters
     ----------
     text : str
 
     Examples
     --------
 
     Simple uses::
 
         In [2]: strip_email_quotes('> > text')
         Out[2]: 'text'
 
         In [3]: strip_email_quotes('> > text\\n> > more')
         Out[3]: 'text\\nmore'
 
     Note how only the common prefix that appears in all lines is stripped::
 
         In [4]: strip_email_quotes('> > text\\n> > more\\n> more...')
         Out[4]: '> text\\n> more\\nmore...'
 
     So if any line has no quote marks ('>') , then none are stripped from any
     of them ::
     
         In [5]: strip_email_quotes('> > text\\n> > more\\nlast different')
         Out[5]: '> > text\\n> > more\\nlast different'
     """
     lines = text.splitlines()
     matches = set()
     for line in lines:
         prefix = re.match(r'^(\s*>[ >]*)', line)
         if prefix:
             matches.add(prefix.group(1))
         else:
             break
     else:
         prefix = long_substr(list(matches))
         if prefix:
             strip = len(prefix)
             text = '\n'.join([ ln[strip:] for ln in lines])
     return text
 
+def strip_ansi(source):
+    """
+    Remove ansi escape codes from text.
+    
+    Parameters
+    ----------
+    source : str
+        Source to remove the ansi from
+    """
+    return re.sub(r'\033\[(\d|;)+?m', '', source)
+
 
 class EvalFormatter(Formatter):
     """A String Formatter that allows evaluation of simple expressions.
     
     Note that this version interprets a : as specifying a format string (as per
     standard string formatting), so if slicing is required, you must explicitly
     create a slice.
     
     This is to be used in templating cases, such as the parallel batch
     script templates, where simple arithmetic on arguments is useful.
 
     Examples
     --------
     ::
 
         In [1]: f = EvalFormatter()
         In [2]: f.format('{n//4}', n=8)
         Out[2]: '2'
 
         In [3]: f.format("{greeting[slice(2,4)]}", greeting="Hello")
         Out[3]: 'll'
     """
     def get_field(self, name, args, kwargs):
         v = eval(name, kwargs)
         return v, name
 
 #XXX: As of Python 3.4, the format string parsing no longer splits on a colon
 # inside [], so EvalFormatter can handle slicing. Once we only support 3.4 and
 # above, it should be possible to remove FullEvalFormatter.
 
 @skip_doctest_py3
 class FullEvalFormatter(Formatter):
     """A String Formatter that allows evaluation of simple expressions.
     
     Any time a format key is not found in the kwargs,
     it will be tried as an expression in the kwargs namespace.
     
     Note that this version allows slicing using [1:2], so you cannot specify
     a format string. Use :class:`EvalFormatter` to permit format strings.
     
     Examples
     --------
     ::
 
         In [1]: f = FullEvalFormatter()
         In [2]: f.format('{n//4}', n=8)
         Out[2]: u'2'
 
         In [3]: f.format('{list(range(5))[2:4]}')
         Out[3]: u'[2, 3]'
 
         In [4]: f.format('{3*2}')
         Out[4]: u'6'
     """
     # copied from Formatter._vformat with minor changes to allow eval
     # and replace the format_spec code with slicing
     def _vformat(self, format_string, args, kwargs, used_args, recursion_depth):
         if recursion_depth < 0:
             raise ValueError('Max string recursion exceeded')
         result = []
         for literal_text, field_name, format_spec, conversion in \
                 self.parse(format_string):
 
             # output the literal text
             if literal_text:
                 result.append(literal_text)
 
             # if there's a field, output it
             if field_name is not None:
                 # this is some markup, find the object and do
                 # the formatting
 
                 if format_spec:
                     # override format spec, to allow slicing:
                     field_name = ':'.join([field_name, format_spec])
 
                 # eval the contents of the field for the object
                 # to be formatted
                 obj = eval(field_name, kwargs)
 
                 # do any conversion on the resulting object
                 obj = self.convert_field(obj, conversion)
 
                 # format the object and append to the result
                 result.append(self.format_field(obj, ''))
 
         return u''.join(py3compat.cast_unicode(s) for s in result)
 
 
 @skip_doctest_py3
 class DollarFormatter(FullEvalFormatter):
     """Formatter allowing Itpl style $foo replacement, for names and attribute
     access only. Standard {foo} replacement also works, and allows full
     evaluation of its arguments.
 
     Examples
     --------
     ::
 
         In [1]: f = DollarFormatter()
         In [2]: f.format('{n//4}', n=8)
         Out[2]: u'2'
 
         In [3]: f.format('23 * 76 is $result', result=23*76)
         Out[3]: u'23 * 76 is 1748'
 
         In [4]: f.format('$a or {b}', a=1, b=2)
         Out[4]: u'1 or 2'
     """
     _dollar_pattern = re.compile("(.*?)\$(\$?[\w\.]+)")
     def parse(self, fmt_string):
         for literal_txt, field_name, format_spec, conversion \
                     in Formatter.parse(self, fmt_string):
             
             # Find $foo patterns in the literal text.
             continue_from = 0
             txt = ""
             for m in self._dollar_pattern.finditer(literal_txt):
                 new_txt, new_field = m.group(1,2)
                 # $$foo --> $foo
                 if new_field.startswith("$"):
                     txt += new_txt + new_field
                 else:
                     yield (txt + new_txt, new_field, "", None)
                     txt = ""
                 continue_from = m.end()
             
             # Re-yield the {foo} style pattern
             yield (txt + literal_txt[continue_from:], field_name, format_spec, conversion)
 
 #-----------------------------------------------------------------------------
 # Utils to columnize a list of string
 #-----------------------------------------------------------------------------
 
 def _chunks(l, n):
     """Yield successive n-sized chunks from l."""
     for i in py3compat.xrange(0, len(l), n):
         yield l[i:i+n]
 
 
 def _find_optimal(rlist , separator_size=2 , displaywidth=80):
     """Calculate optimal info to columnize a list of string"""
     for nrow in range(1, len(rlist)+1) :
         chk = list(map(max,_chunks(rlist, nrow)))
         sumlength = sum(chk)
         ncols = len(chk)
         if sumlength+separator_size*(ncols-1) <= displaywidth :
             break;
     return {'columns_numbers' : ncols,
             'optimal_separator_width':(displaywidth - sumlength)/(ncols-1) if (ncols -1) else 0,
             'rows_numbers' : nrow,
             'columns_width' : chk
            }
 
 
 def _get_or_default(mylist, i, default=None):
     """return list item number, or default if don't exist"""
     if i >= len(mylist):
         return default
     else :
         return mylist[i]
 
 
 @skip_doctest
 def compute_item_matrix(items, empty=None, *args, **kwargs) :
     """Returns a nested list, and info to columnize items
 
     Parameters
     ----------
 
     items
         list of strings to columize
     empty : (default None)
         default value to fill list if needed
     separator_size : int (default=2)
         How much caracters will be used as a separation between each columns.
     displaywidth : int (default=80)
         The width of the area onto wich the columns should enter
 
     Returns
     -------
 
     strings_matrix
 
         nested list of string, the outer most list contains as many list as
         rows, the innermost lists have each as many element as colums. If the
         total number of elements in `items` does not equal the product of
         rows*columns, the last element of some lists are filled with `None`.
 
     dict_info
         some info to make columnize easier:
 
         columns_numbers
           number of columns
         rows_numbers
           number of rows
         columns_width
           list of with of each columns
         optimal_separator_width
           best separator width between columns
 
     Examples
     --------
     ::
 
         In [1]: l = ['aaa','b','cc','d','eeeee','f','g','h','i','j','k','l']
            ...: compute_item_matrix(l,displaywidth=12)
         Out[1]:
             ([['aaa', 'f', 'k'],
             ['b', 'g', 'l'],
             ['cc', 'h', None],
             ['d', 'i', None],
             ['eeeee', 'j', None]],
             {'columns_numbers': 3,
             'columns_width': [5, 1, 1],
             'optimal_separator_width': 2,
             'rows_numbers': 5})
     """
     info = _find_optimal(list(map(len, items)), *args, **kwargs)
     nrow, ncol = info['rows_numbers'], info['columns_numbers']
     return ([[ _get_or_default(items, c*nrow+i, default=empty) for c in range(ncol) ] for i in range(nrow) ], info)
 
 
 def columnize(items, separator='  ', displaywidth=80):
     """ Transform a list of strings into a single string with columns.
 
     Parameters
     ----------
     items : sequence of strings
         The strings to process.
 
     separator : str, optional [default is two spaces]
         The string that separates columns.
 
     displaywidth : int, optional [default is 80]
         Width of the display in number of characters.
 
     Returns
     -------
     The formatted string.
     """
     if not items :
         return '\n'
     matrix, info = compute_item_matrix(items, separator_size=len(separator), displaywidth=displaywidth)
     fmatrix = [filter(None, x) for x in matrix]
     sjoin = lambda x : separator.join([ y.ljust(w, ' ') for y, w in zip(x, info['columns_width'])])
     return '\n'.join(map(sjoin, fmatrix))+'\n'
 
 
 def get_text_list(list_, last_sep=' and ', sep=", ", wrap_item_with=""):
     """
     Return a string with a natural enumeration of items
 
     >>> get_text_list(['a', 'b', 'c', 'd'])
     'a, b, c and d'
     >>> get_text_list(['a', 'b', 'c'], ' or ')
     'a, b or c'
     >>> get_text_list(['a', 'b', 'c'], ', ')
     'a, b, c'
     >>> get_text_list(['a', 'b'], ' or ')
     'a or b'
     >>> get_text_list(['a'])
     'a'
     >>> get_text_list([])
     ''
     >>> get_text_list(['a', 'b'], wrap_item_with="`")
     '`a` and `b`'
     >>> get_text_list(['a', 'b', 'c', 'd'], " = ", sep=" + ")
     'a + b + c = d'
     """
     if len(list_) == 0:
         return ''
     if wrap_item_with:
         list_ = ['%s%s%s' % (wrap_item_with, item, wrap_item_with) for
                  item in list_]
     if len(list_) == 1:
         return list_[0]
     return '%s%s%s' % (
         sep.join(i for i in list_[:-1]),
         last_sep, list_[-1])
  No newline at end of file