Use StringIO.StringIO on Python 2.

io.StringIO is strict about unicode, StringIO.StringIO isn't.

IPython/core/formatters.py

@@ -1,654 +1,657
 # -*- coding: utf-8 -*-
 """Display formatters.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.core.formatters
    :parts: 3
 
 Authors:
 
 * Robert Kern
 * Brian Granger
 """
 #-----------------------------------------------------------------------------
 # Copyright (C) 2010-2011, 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
 #-----------------------------------------------------------------------------
 
 # Stdlib imports
 import abc
 import sys
 import warnings
-# We must use StringIO, as cStringIO doesn't handle unicode properly.
-from io import StringIO
 
 # Our own imports
 from IPython.config.configurable import Configurable
 from IPython.lib import pretty
 from IPython.utils.traitlets import (
     Bool, Dict, Integer, Unicode, CUnicode, ObjectName, List,
 )
-from IPython.utils.py3compat import unicode_to_str, with_metaclass
+from IPython.utils.py3compat import unicode_to_str, with_metaclass, PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 
 #-----------------------------------------------------------------------------
 # The main DisplayFormatter class
 #-----------------------------------------------------------------------------
 
 
 class DisplayFormatter(Configurable):
 
     # When set to true only the default plain text formatter will be used.
     plain_text_only = Bool(False, config=True)
     def _plain_text_only_changed(self, name, old, new):
         warnings.warn("""DisplayFormatter.plain_text_only is deprecated.
         
         Use DisplayFormatter.active_types = ['text/plain']
         for the same effect.
         """, DeprecationWarning)
         if new:
             self.active_types = ['text/plain']
         else:
             self.active_types = self.format_types
     
     active_types = List(Unicode, config=True,
         help="""List of currently active mime-types to display.
         You can use this to set a white-list for formats to display.
         
         Most users will not need to change this value.
         """)
     def _active_types_default(self):
         return self.format_types
     
     def _active_types_changed(self, name, old, new):
         for key, formatter in self.formatters.items():
             if key in new:
                 formatter.enabled = True
             else:
                 formatter.enabled = False
     
     # A dict of formatter whose keys are format types (MIME types) and whose
     # values are subclasses of BaseFormatter.
     formatters = Dict()
     def _formatters_default(self):
         """Activate the default formatters."""
         formatter_classes = [
             PlainTextFormatter,
             HTMLFormatter,
             SVGFormatter,
             PNGFormatter,
             JPEGFormatter,
             LatexFormatter,
             JSONFormatter,
             JavascriptFormatter
         ]
         d = {}
         for cls in formatter_classes:
             f = cls(parent=self)
             d[f.format_type] = f
         return d
 
     def format(self, obj, include=None, exclude=None):
         """Return a format data dict for an object.
 
         By default all format types will be computed.
 
         The following MIME types are currently implemented:
 
         * text/plain
         * text/html
         * text/latex
         * application/json
         * application/javascript
         * image/png
         * image/jpeg
         * image/svg+xml
 
         Parameters
         ----------
         obj : object
             The Python object whose format data will be computed.
         include : list or tuple, optional
             A list of format type strings (MIME types) to include in the
             format data dict. If this is set *only* the format types included
             in this list will be computed.
         exclude : list or tuple, optional
             A list of format type string (MIME types) to exclude in the format
             data dict. If this is set all format types will be computed,
             except for those included in this argument.
 
         Returns
         -------
         (format_dict, metadata_dict) : tuple of two dicts
         
             format_dict is a dictionary of key/value pairs, one of each format that was
             generated for the object. The keys are the format types, which
             will usually be MIME type strings and the values and JSON'able
             data structure containing the raw data for the representation in
             that format.
             
             metadata_dict is a dictionary of metadata about each mime-type output.
             Its keys will be a strict subset of the keys in format_dict.
         """
         format_dict = {}
         md_dict = {}
 
         for format_type, formatter in self.formatters.items():
             if include and format_type not in include:
                 continue
             if exclude and format_type in exclude:
                 continue
             
             md = None
             try:
                 data = formatter(obj)
             except:
                 # FIXME: log the exception
                 raise
             
             # formatters can return raw data or (data, metadata)
             if isinstance(data, tuple) and len(data) == 2:
                 data, md = data
             
             if data is not None:
                 format_dict[format_type] = data
             if md is not None:
                 md_dict[format_type] = md
             
         return format_dict, md_dict
 
     @property
     def format_types(self):
         """Return the format types (MIME types) of the active formatters."""
         return list(self.formatters.keys())
 
 
 #-----------------------------------------------------------------------------
 # Formatters for specific format types (text, html, svg, etc.)
 #-----------------------------------------------------------------------------
 
 
 class FormatterABC(with_metaclass(abc.ABCMeta, object)):
     """ Abstract base class for Formatters.
 
     A formatter is a callable class that is responsible for computing the
     raw format data for a particular format type (MIME type). For example,
     an HTML formatter would have a format type of `text/html` and would return
     the HTML representation of the object when called.
     """
 
     # The format type of the data returned, usually a MIME type.
     format_type = 'text/plain'
 
     # Is the formatter enabled...
     enabled = True
 
     @abc.abstractmethod
     def __call__(self, obj):
         """Return a JSON'able representation of the object.
 
         If the object cannot be formatted by this formatter, then return None
         """
         try:
             return repr(obj)
         except TypeError:
             return None
 
 
 class BaseFormatter(Configurable):
     """A base formatter class that is configurable.
 
     This formatter should usually be used as the base class of all formatters.
     It is a traited :class:`Configurable` class and includes an extensible
     API for users to determine how their objects are formatted. The following
     logic is used to find a function to format an given object.
 
     1. The object is introspected to see if it has a method with the name
        :attr:`print_method`. If is does, that object is passed to that method
        for formatting.
     2. If no print method is found, three internal dictionaries are consulted
        to find print method: :attr:`singleton_printers`, :attr:`type_printers`
        and :attr:`deferred_printers`.
 
     Users should use these dictionaries to register functions that will be
     used to compute the format data for their objects (if those objects don't
     have the special print methods). The easiest way of using these
     dictionaries is through the :meth:`for_type` and :meth:`for_type_by_name`
     methods.
 
     If no function/callable is found to compute the format data, ``None`` is
     returned and this format type is not used.
     """
 
     format_type = Unicode('text/plain')
 
     enabled = Bool(True, config=True)
 
     print_method = ObjectName('__repr__')
 
     # The singleton printers.
     # Maps the IDs of the builtin singleton objects to the format functions.
     singleton_printers = Dict(config=True)
     def _singleton_printers_default(self):
         return {}
 
     # The type-specific printers.
     # Map type objects to the format functions.
     type_printers = Dict(config=True)
     def _type_printers_default(self):
         return {}
 
     # The deferred-import type-specific printers.
     # Map (modulename, classname) pairs to the format functions.
     deferred_printers = Dict(config=True)
     def _deferred_printers_default(self):
         return {}
 
     def __call__(self, obj):
         """Compute the format for an object."""
         if self.enabled:
             obj_id = id(obj)
             try:
                 obj_class = getattr(obj, '__class__', None) or type(obj)
                 # First try to find registered singleton printers for the type.
                 try:
                     printer = self.singleton_printers[obj_id]
                 except (TypeError, KeyError):
                     pass
                 else:
                     return printer(obj)
                 # Next look for type_printers.
                 for cls in pretty._get_mro(obj_class):
                     if cls in self.type_printers:
                         return self.type_printers[cls](obj)
                     else:
                         printer = self._in_deferred_types(cls)
                         if printer is not None:
                             return printer(obj)
                 # Finally look for special method names.
                 if hasattr(obj_class, self.print_method):
                     printer = getattr(obj_class, self.print_method)
                     return printer(obj)
                 return None
             except Exception:
                 pass
         else:
             return None
 
     def for_type(self, typ, func):
         """Add a format function for a given type.
 
         Parameters
         -----------
         typ : class
             The class of the object that will be formatted using `func`.
         func : callable
             The callable that will be called to compute the format data. The
             call signature of this function is simple, it must take the
             object to be formatted and return the raw data for the given
             format. Subclasses may use a different call signature for the
             `func` argument.
         """
         oldfunc = self.type_printers.get(typ, None)
         if func is not None:
             # To support easy restoration of old printers, we need to ignore
             # Nones.
             self.type_printers[typ] = func
         return oldfunc
 
     def for_type_by_name(self, type_module, type_name, func):
         """Add a format function for a type specified by the full dotted
         module and name of the type, rather than the type of the object.
 
         Parameters
         ----------
         type_module : str
             The full dotted name of the module the type is defined in, like
             ``numpy``.
         type_name : str
             The name of the type (the class name), like ``dtype``
         func : callable
             The callable that will be called to compute the format data. The
             call signature of this function is simple, it must take the
             object to be formatted and return the raw data for the given
             format. Subclasses may use a different call signature for the
             `func` argument.
         """
         key = (type_module, type_name)
         oldfunc = self.deferred_printers.get(key, None)
         if func is not None:
             # To support easy restoration of old printers, we need to ignore
             # Nones.
             self.deferred_printers[key] = func
         return oldfunc
 
     def _in_deferred_types(self, cls):
         """
         Check if the given class is specified in the deferred type registry.
 
         Returns the printer from the registry if it exists, and None if the
         class is not in the registry. Successful matches will be moved to the
         regular type registry for future use.
         """
         mod = getattr(cls, '__module__', None)
         name = getattr(cls, '__name__', None)
         key = (mod, name)
         printer = None
         if key in self.deferred_printers:
             # Move the printer over to the regular registry.
             printer = self.deferred_printers.pop(key)
             self.type_printers[cls] = printer
         return printer
 
 
 class PlainTextFormatter(BaseFormatter):
     """The default pretty-printer.
 
     This uses :mod:`IPython.lib.pretty` to compute the format data of
     the object. If the object cannot be pretty printed, :func:`repr` is used.
     See the documentation of :mod:`IPython.lib.pretty` for details on
     how to write pretty printers.  Here is a simple example::
 
         def dtype_pprinter(obj, p, cycle):
             if cycle:
                 return p.text('dtype(...)')
             if hasattr(obj, 'fields'):
                 if obj.fields is None:
                     p.text(repr(obj))
                 else:
                     p.begin_group(7, 'dtype([')
                     for i, field in enumerate(obj.descr):
                         if i > 0:
                             p.text(',')
                             p.breakable()
                         p.pretty(field)
                     p.end_group(7, '])')
     """
 
     # The format type of data returned.
     format_type = Unicode('text/plain')
 
     # This subclass ignores this attribute as it always need to return
     # something.
     enabled = Bool(True, config=False)
 
     # Look for a _repr_pretty_ methods to use for pretty printing.
     print_method = ObjectName('_repr_pretty_')
 
     # Whether to pretty-print or not.
     pprint = Bool(True, config=True)
 
     # Whether to be verbose or not.
     verbose = Bool(False, config=True)
 
     # The maximum width.
     max_width = Integer(79, config=True)
 
     # The newline character.
     newline = Unicode('\n', config=True)
 
     # format-string for pprinting floats
     float_format = Unicode('%r')
     # setter for float precision, either int or direct format-string
     float_precision = CUnicode('', config=True)
 
     def _float_precision_changed(self, name, old, new):
         """float_precision changed, set float_format accordingly.
 
         float_precision can be set by int or str.
         This will set float_format, after interpreting input.
         If numpy has been imported, numpy print precision will also be set.
 
         integer `n` sets format to '%.nf', otherwise, format set directly.
 
         An empty string returns to defaults (repr for float, 8 for numpy).
 
         This parameter can be set via the '%precision' magic.
         """
 
         if '%' in new:
             # got explicit format string
             fmt = new
             try:
                 fmt%3.14159
             except Exception:
                 raise ValueError("Precision must be int or format string, not %r"%new)
         elif new:
             # otherwise, should be an int
             try:
                 i = int(new)
                 assert i >= 0
             except ValueError:
                 raise ValueError("Precision must be int or format string, not %r"%new)
             except AssertionError:
                 raise ValueError("int precision must be non-negative, not %r"%i)
 
             fmt = '%%.%if'%i
             if 'numpy' in sys.modules:
                 # set numpy precision if it has been imported
                 import numpy
                 numpy.set_printoptions(precision=i)
         else:
             # default back to repr
             fmt = '%r'
             if 'numpy' in sys.modules:
                 import numpy
                 # numpy default is 8
                 numpy.set_printoptions(precision=8)
         self.float_format = fmt
 
     # Use the default pretty printers from IPython.lib.pretty.
     def _singleton_printers_default(self):
         return pretty._singleton_pprinters.copy()
 
     def _type_printers_default(self):
         d = pretty._type_pprinters.copy()
         d[float] = lambda obj,p,cycle: p.text(self.float_format%obj)
         return d
 
     def _deferred_printers_default(self):
         return pretty._deferred_type_pprinters.copy()
 
     #### FormatterABC interface ####
 
     def __call__(self, obj):
         """Compute the pretty representation of the object."""
         if not self.pprint:
             try:
                 return repr(obj)
             except TypeError:
                 return ''
         else:
             # This uses use StringIO, as cStringIO doesn't handle unicode.
             stream = StringIO()
             # self.newline.encode() is a quick fix for issue gh-597. We need to
             # ensure that stream does not get a mix of unicode and bytestrings,
             # or it will cause trouble.
             printer = pretty.RepresentationPrinter(stream, self.verbose,
                 self.max_width, unicode_to_str(self.newline),
                 singleton_pprinters=self.singleton_printers,
                 type_pprinters=self.type_printers,
                 deferred_pprinters=self.deferred_printers)
             printer.pretty(obj)
             printer.flush()
             return stream.getvalue()
 
 
 class HTMLFormatter(BaseFormatter):
     """An HTML formatter.
 
     To define the callables that compute the HTML representation of your
     objects, define a :meth:`_repr_html_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid HTML snippet that
     could be injected into an existing DOM. It should *not* include the
     ```<html>`` or ```<body>`` tags.
     """
     format_type = Unicode('text/html')
 
     print_method = ObjectName('_repr_html_')
 
 
 class SVGFormatter(BaseFormatter):
     """An SVG formatter.
 
     To define the callables that compute the SVG representation of your
     objects, define a :meth:`_repr_svg_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be valid SVG enclosed in
     ```<svg>``` tags, that could be injected into an existing DOM. It should
     *not* include the ```<html>`` or ```<body>`` tags.
     """
     format_type = Unicode('image/svg+xml')
 
     print_method = ObjectName('_repr_svg_')
 
 
 class PNGFormatter(BaseFormatter):
     """A PNG formatter.
 
     To define the callables that compute the PNG representation of your
     objects, define a :meth:`_repr_png_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be raw PNG data, *not*
     base64 encoded.
     """
     format_type = Unicode('image/png')
 
     print_method = ObjectName('_repr_png_')
 
 
 class JPEGFormatter(BaseFormatter):
     """A JPEG formatter.
 
     To define the callables that compute the JPEG representation of your
     objects, define a :meth:`_repr_jpeg_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be raw JPEG data, *not*
     base64 encoded.
     """
     format_type = Unicode('image/jpeg')
 
     print_method = ObjectName('_repr_jpeg_')
 
 
 class LatexFormatter(BaseFormatter):
     """A LaTeX formatter.
 
     To define the callables that compute the LaTeX representation of your
     objects, define a :meth:`_repr_latex_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid LaTeX equation,
     enclosed in either ```$```, ```$$``` or another LaTeX equation
     environment.
     """
     format_type = Unicode('text/latex')
 
     print_method = ObjectName('_repr_latex_')
 
 
 class JSONFormatter(BaseFormatter):
     """A JSON string formatter.
 
     To define the callables that compute the JSON string representation of
     your objects, define a :meth:`_repr_json_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid JSON string.
     """
     format_type = Unicode('application/json')
 
     print_method = ObjectName('_repr_json_')
 
 
 class JavascriptFormatter(BaseFormatter):
     """A Javascript formatter.
 
     To define the callables that compute the Javascript representation of
     your objects, define a :meth:`_repr_javascript_` method or use the
     :meth:`for_type` or :meth:`for_type_by_name` methods to register functions
     that handle this.
 
     The return value of this formatter should be valid Javascript code and
     should *not* be enclosed in ```<script>``` tags.
     """
     format_type = Unicode('application/javascript')
 
     print_method = ObjectName('_repr_javascript_')
 
 FormatterABC.register(BaseFormatter)
 FormatterABC.register(PlainTextFormatter)
 FormatterABC.register(HTMLFormatter)
 FormatterABC.register(SVGFormatter)
 FormatterABC.register(PNGFormatter)
 FormatterABC.register(JPEGFormatter)
 FormatterABC.register(LatexFormatter)
 FormatterABC.register(JSONFormatter)
 FormatterABC.register(JavascriptFormatter)
 
 
 def format_display_data(obj, include=None, exclude=None):
     """Return a format data dict for an object.
 
     By default all format types will be computed.
 
     The following MIME types are currently implemented:
 
     * text/plain
     * text/html
     * text/latex
     * application/json
     * application/javascript
     * image/png
     * image/jpeg
     * image/svg+xml
 
     Parameters
     ----------
     obj : object
         The Python object whose format data will be computed.
 
     Returns
     -------
     format_dict : dict
         A dictionary of key/value pairs, one or each format that was
         generated for the object. The keys are the format types, which
         will usually be MIME type strings and the values and JSON'able
         data structure containing the raw data for the representation in
         that format.
     include : list or tuple, optional
         A list of format type strings (MIME types) to include in the
         format data dict. If this is set *only* the format types included
         in this list will be computed.
     exclude : list or tuple, optional
         A list of format type string (MIME types) to exclue in the format
         data dict. If this is set all format types will be computed,
         except for those included in this argument.
     """
     from IPython.core.interactiveshell import InteractiveShell
 
     InteractiveShell.instance().display_formatter.format(
         obj,
         include,
         exclude
     )
 

IPython/core/inputtransformer.py

@@ -1,531 +1,535
 import abc
 import functools
 import re
-from io import StringIO
 
 from IPython.core.splitinput import LineInfo
 from IPython.utils import tokenize2
 from IPython.utils.openpy import cookie_comment_re
-from IPython.utils.py3compat import with_metaclass
+from IPython.utils.py3compat import with_metaclass, PY3
 from IPython.utils.tokenize2 import generate_tokens, untokenize, TokenError
 
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # The escape sequences that define the syntax transformations IPython will
 # apply to user input.  These can NOT be just changed here: many regular
 # expressions and other parts of the code may use their hardcoded values, and
 # for all intents and purposes they constitute the 'IPython syntax', so they
 # should be considered fixed.
 
 ESC_SHELL  = '!'     # Send line to underlying system shell
 ESC_SH_CAP = '!!'    # Send line to system shell and capture output
 ESC_HELP   = '?'     # Find information about object
 ESC_HELP2  = '??'    # Find extra-detailed information about object
 ESC_MAGIC  = '%'     # Call magic function
 ESC_MAGIC2 = '%%'    # Call cell-magic function
 ESC_QUOTE  = ','     # Split args on whitespace, quote each as string and call
 ESC_QUOTE2 = ';'     # Quote all args as a single string, call
 ESC_PAREN  = '/'     # Call first argument with rest of line as arguments
 
 ESC_SEQUENCES = [ESC_SHELL, ESC_SH_CAP, ESC_HELP ,\
                  ESC_HELP2, ESC_MAGIC, ESC_MAGIC2,\
                  ESC_QUOTE, ESC_QUOTE2, ESC_PAREN ]
 
 
 class InputTransformer(with_metaclass(abc.ABCMeta, object)):
     """Abstract base class for line-based input transformers."""
     
     @abc.abstractmethod
     def push(self, line):
         """Send a line of input to the transformer, returning the transformed
         input or None if the transformer is waiting for more input.
         
         Must be overridden by subclasses.
         """
         pass
     
     @abc.abstractmethod
     def reset(self):
         """Return, transformed any lines that the transformer has accumulated,
         and reset its internal state.
         
         Must be overridden by subclasses.
         """
         pass
     
     @classmethod
     def wrap(cls, func):
         """Can be used by subclasses as a decorator, to return a factory that
         will allow instantiation with the decorated object.
         """
         @functools.wraps(func)
         def transformer_factory(**kwargs):
             return cls(func, **kwargs)
         
         return transformer_factory
 
 class StatelessInputTransformer(InputTransformer):
     """Wrapper for a stateless input transformer implemented as a function."""
     def __init__(self, func):
         self.func = func
     
     def __repr__(self):
         return "StatelessInputTransformer(func={0!r})".format(self.func)
     
     def push(self, line):
         """Send a line of input to the transformer, returning the
         transformed input."""
         return self.func(line)
     
     def reset(self):
         """No-op - exists for compatibility."""
         pass
 
 class CoroutineInputTransformer(InputTransformer):
     """Wrapper for an input transformer implemented as a coroutine."""
     def __init__(self, coro, **kwargs):
         # Prime it
         self.coro = coro(**kwargs)
         next(self.coro)
     
     def __repr__(self):
         return "CoroutineInputTransformer(coro={0!r})".format(self.coro)
     
     def push(self, line):
         """Send a line of input to the transformer, returning the
         transformed input or None if the transformer is waiting for more
         input.
         """
         return self.coro.send(line)
     
     def reset(self):
         """Return, transformed any lines that the transformer has
         accumulated, and reset its internal state.
         """
         return self.coro.send(None)
 
 class TokenInputTransformer(InputTransformer):
     """Wrapper for a token-based input transformer.
     
     func should accept a list of tokens (5-tuples, see tokenize docs), and
     return an iterable which can be passed to tokenize.untokenize().
     """
     def __init__(self, func):
         self.func = func
         self.current_line = ""
         self.line_used = False
         self.reset_tokenizer()
     
     def reset_tokenizer(self):
         self.tokenizer = generate_tokens(self.get_line)
     
     def get_line(self):
         if self.line_used:
             raise TokenError
         self.line_used = True
         return self.current_line
     
     def push(self, line):
         self.current_line += line + "\n"
         if self.current_line.isspace():
             return self.reset()
         
         self.line_used = False
         tokens = []
         stop_at_NL = False
         try:
             for intok in self.tokenizer:
                 tokens.append(intok)
                 t = intok[0]
                 if t == tokenize2.NEWLINE or (stop_at_NL and t == tokenize2.NL):
                     # Stop before we try to pull a line we don't have yet
                     break
                 elif t == tokenize2.ERRORTOKEN:
                     stop_at_NL = True
         except TokenError:
             # Multi-line statement - stop and try again with the next line
             self.reset_tokenizer()
             return None
         
         return self.output(tokens)
     
     def output(self, tokens):
         self.current_line = ""
         self.reset_tokenizer()
         return untokenize(self.func(tokens)).rstrip('\n')
     
     def reset(self):
         l = self.current_line
         self.current_line = ""
         self.reset_tokenizer()
         if l:
             return l.rstrip('\n')
 
 class assemble_python_lines(TokenInputTransformer):
     def __init__(self):
         super(assemble_python_lines, self).__init__(None)
     
     def output(self, tokens):
         return self.reset()
 
 @CoroutineInputTransformer.wrap
 def assemble_logical_lines():
     """Join lines following explicit line continuations (\)"""
     line = ''
     while True:
         line = (yield line)
         if not line or line.isspace():
             continue
         
         parts = []
         while line is not None:
             if line.endswith('\\') and (not has_comment(line)):
                 parts.append(line[:-1])
                 line = (yield None) # Get another line
             else:
                 parts.append(line)
                 break
         
         # Output
         line = ''.join(parts)
 
 # Utilities
 def _make_help_call(target, esc, lspace, next_input=None):
     """Prepares a pinfo(2)/psearch call from a target name and the escape
     (i.e. ? or ??)"""
     method  = 'pinfo2' if esc == '??' \
                 else 'psearch' if '*' in target \
                 else 'pinfo'
     arg = " ".join([method, target])
     if next_input is None:
         return '%sget_ipython().magic(%r)' % (lspace, arg)
     else:
         return '%sget_ipython().set_next_input(%r);get_ipython().magic(%r)' % \
            (lspace, next_input, arg)
     
 # These define the transformations for the different escape characters.
 def _tr_system(line_info):
     "Translate lines escaped with: !"
     cmd = line_info.line.lstrip().lstrip(ESC_SHELL)
     return '%sget_ipython().system(%r)' % (line_info.pre, cmd)
 
 def _tr_system2(line_info):
     "Translate lines escaped with: !!"
     cmd = line_info.line.lstrip()[2:]
     return '%sget_ipython().getoutput(%r)' % (line_info.pre, cmd)
 
 def _tr_help(line_info):
     "Translate lines escaped with: ?/??"
     # A naked help line should just fire the intro help screen
     if not line_info.line[1:]:
         return 'get_ipython().show_usage()'
 
     return _make_help_call(line_info.ifun, line_info.esc, line_info.pre)
 
 def _tr_magic(line_info):
     "Translate lines escaped with: %"
     tpl = '%sget_ipython().magic(%r)'
     if line_info.line.startswith(ESC_MAGIC2):
         return line_info.line
     cmd = ' '.join([line_info.ifun, line_info.the_rest]).strip()
     return tpl % (line_info.pre, cmd)
 
 def _tr_quote(line_info):
     "Translate lines escaped with: ,"
     return '%s%s("%s")' % (line_info.pre, line_info.ifun,
                          '", "'.join(line_info.the_rest.split()) )
 
 def _tr_quote2(line_info):
     "Translate lines escaped with: ;"
     return '%s%s("%s")' % (line_info.pre, line_info.ifun,
                            line_info.the_rest)
 
 def _tr_paren(line_info):
     "Translate lines escaped with: /"
     return '%s%s(%s)' % (line_info.pre, line_info.ifun,
                          ", ".join(line_info.the_rest.split()))
 
 tr = { ESC_SHELL  : _tr_system,
        ESC_SH_CAP : _tr_system2,
        ESC_HELP   : _tr_help,
        ESC_HELP2  : _tr_help,
        ESC_MAGIC  : _tr_magic,
        ESC_QUOTE  : _tr_quote,
        ESC_QUOTE2 : _tr_quote2,
        ESC_PAREN  : _tr_paren }
 
 @StatelessInputTransformer.wrap
 def escaped_commands(line):
     """Transform escaped commands - %magic, !system, ?help + various autocalls.
     """
     if not line or line.isspace():
         return line
     lineinf = LineInfo(line)
     if lineinf.esc not in tr:
         return line
     
     return tr[lineinf.esc](lineinf)
 
 _initial_space_re = re.compile(r'\s*')
 
 _help_end_re = re.compile(r"""(%{0,2}
                               [a-zA-Z_*][\w*]*        # Variable name
                               (\.[a-zA-Z_*][\w*]*)*   # .etc.etc
                               )
                               (\?\??)$                # ? or ??
                               """,
                               re.VERBOSE)
 
 # Extra pseudotokens for multiline strings and data structures
 _MULTILINE_STRING = object()
 _MULTILINE_STRUCTURE = object()
 
 def _line_tokens(line):
     """Helper for has_comment and ends_in_comment_or_string."""
     readline = StringIO(line).readline
     toktypes = set()
     try:
         for t in generate_tokens(readline):
             toktypes.add(t[0])
     except TokenError as e:
         # There are only two cases where a TokenError is raised.
         if 'multi-line string' in e.args[0]:
             toktypes.add(_MULTILINE_STRING)
         else:
             toktypes.add(_MULTILINE_STRUCTURE)
     return toktypes
 
 def has_comment(src):
     """Indicate whether an input line has (i.e. ends in, or is) a comment.
 
     This uses tokenize, so it can distinguish comments from # inside strings.
 
     Parameters
     ----------
     src : string
       A single line input string.
 
     Returns
     -------
     comment : bool
         True if source has a comment.
     """
     return (tokenize2.COMMENT in _line_tokens(src))
 
 def ends_in_comment_or_string(src):
     """Indicates whether or not an input line ends in a comment or within
     a multiline string.
     
     Parameters
     ----------
     src : string
       A single line input string.
 
     Returns
     -------
     comment : bool
         True if source ends in a comment or multiline string.
     """
     toktypes = _line_tokens(src)
     return (tokenize2.COMMENT in toktypes) or (_MULTILINE_STRING in toktypes)
         
 
 @StatelessInputTransformer.wrap
 def help_end(line):
     """Translate lines with ?/?? at the end"""
     m = _help_end_re.search(line)
     if m is None or ends_in_comment_or_string(line):
         return line
     target = m.group(1)
     esc = m.group(3)
     lspace = _initial_space_re.match(line).group(0)
 
     # If we're mid-command, put it back on the next prompt for the user.
     next_input = line.rstrip('?') if line.strip() != m.group(0) else None
 
     return _make_help_call(target, esc, lspace, next_input)
 
 
 @CoroutineInputTransformer.wrap
 def cellmagic(end_on_blank_line=False):
     """Captures & transforms cell magics.
     
     After a cell magic is started, this stores up any lines it gets until it is
     reset (sent None).
     """
     tpl = 'get_ipython().run_cell_magic(%r, %r, %r)'
     cellmagic_help_re = re.compile('%%\w+\?')
     line = ''
     while True:
         line = (yield line)
         # consume leading empty lines
         while not line:
             line = (yield line)
         
         if not line.startswith(ESC_MAGIC2):
             # This isn't a cell magic, idle waiting for reset then start over
             while line is not None:
                 line = (yield line)
             continue
         
         if cellmagic_help_re.match(line):
             # This case will be handled by help_end
             continue
         
         first = line
         body = []
         line = (yield None)
         while (line is not None) and \
                                 ((line.strip() != '') or not end_on_blank_line):
             body.append(line)
             line = (yield None)
         
         # Output
         magic_name, _, first = first.partition(' ')
         magic_name = magic_name.lstrip(ESC_MAGIC2)
         line = tpl % (magic_name, first, u'\n'.join(body))
 
 
 def _strip_prompts(prompt_re, initial_re=None):
     """Remove matching input prompts from a block of input.
     
     Parameters
     ----------
     prompt_re : regular expression
         A regular expression matching any input prompt (including continuation)
     initial_re : regular expression, optional
         A regular expression matching only the initial prompt, but not continuation.
         If no initial expression is given, prompt_re will be used everywhere.
         Used mainly for plain Python prompts, where the continuation prompt
         ``...`` is a valid Python expression in Python 3, so shouldn't be stripped.
     
     If initial_re and prompt_re differ,
     only initial_re will be tested against the first line.
     If any prompt is found on the first two lines,
     prompts will be stripped from the rest of the block.
     """
     if initial_re is None:
         initial_re = prompt_re
     line = ''
     while True:
         line = (yield line)
         
         # First line of cell
         if line is None:
             continue
         out, n1 = initial_re.subn('', line, count=1)
         line = (yield out)
         
         if line is None:
             continue
         # check for any prompt on the second line of the cell,
         # because people often copy from just after the first prompt,
         # so we might not see it in the first line.
         out, n2 = prompt_re.subn('', line, count=1)
         line = (yield out)
         
         if n1 or n2:
             # Found a prompt in the first two lines - check for it in
             # the rest of the cell as well.
             while line is not None:
                 line = (yield prompt_re.sub('', line, count=1))
         
         else:
             # Prompts not in input - wait for reset
             while line is not None:
                 line = (yield line)
 
 @CoroutineInputTransformer.wrap
 def classic_prompt():
     """Strip the >>>/... prompts of the Python interactive shell."""
     # FIXME: non-capturing version (?:...) usable?
     prompt_re = re.compile(r'^(>>> ?|\.\.\. ?)')
     initial_re = re.compile(r'^(>>> ?)')
     return _strip_prompts(prompt_re, initial_re)
 
 @CoroutineInputTransformer.wrap
 def ipy_prompt():
     """Strip IPython's In [1]:/...: prompts."""
     # FIXME: non-capturing version (?:...) usable?
     # FIXME: r'^(In \[\d+\]: | {3}\.{3,}: )' clearer?
     prompt_re = re.compile(r'^(In \[\d+\]: |\ \ \ \.\.\.+: )')
     return _strip_prompts(prompt_re)
 
 
 @CoroutineInputTransformer.wrap
 def leading_indent():
     """Remove leading indentation.
     
     If the first line starts with a spaces or tabs, the same whitespace will be
     removed from each following line until it is reset.
     """
     space_re = re.compile(r'^[ \t]+')
     line = ''
     while True:
         line = (yield line)
         
         if line is None:
             continue
         
         m = space_re.match(line)
         if m:
             space = m.group(0)
             while line is not None:
                 if line.startswith(space):
                     line = line[len(space):]
                 line = (yield line)
         else:
             # No leading spaces - wait for reset
             while line is not None:
                 line = (yield line)
 
 
 @CoroutineInputTransformer.wrap
 def strip_encoding_cookie():
     """Remove encoding comment if found in first two lines
     
     If the first or second line has the `# coding: utf-8` comment,
     it will be removed.
     """
     line = ''
     while True:
         line = (yield line)
         # check comment on first two lines
         for i in range(2):
             if line is None:
                 break
             if cookie_comment_re.match(line):
                 line = (yield "")
             else:
                 line = (yield line)
         
         # no-op on the rest of the cell
         while line is not None:
             line = (yield line)
 
 
 assign_system_re = re.compile(r'(?P<lhs>(\s*)([\w\.]+)((\s*,\s*[\w\.]+)*))'
                               r'\s*=\s*!\s*(?P<cmd>.*)')
 assign_system_template = '%s = get_ipython().getoutput(%r)'
 @StatelessInputTransformer.wrap
 def assign_from_system(line):
     """Transform assignment from system commands (e.g. files = !ls)"""
     m = assign_system_re.match(line)
     if m is None:
         return line
     
     return assign_system_template % m.group('lhs', 'cmd')
 
 assign_magic_re = re.compile(r'(?P<lhs>(\s*)([\w\.]+)((\s*,\s*[\w\.]+)*))'
                              r'\s*=\s*%\s*(?P<cmd>.*)')
 assign_magic_template = '%s = get_ipython().magic(%r)'
 @StatelessInputTransformer.wrap
 def assign_from_magic(line):
     """Transform assignment from magic commands (e.g. a = %who_ls)"""
     m = assign_magic_re.match(line)
     if m is None:
         return line
     
     return assign_magic_template % m.group('lhs', 'cmd')

IPython/core/magics/execution.py

@@ -1,1291 +1,1294
 # -*- coding: utf-8 -*-
 """Implementation of execution-related magic functions.
 """
 from __future__ import print_function
 #-----------------------------------------------------------------------------
 #  Copyright (c) 2012 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
 #-----------------------------------------------------------------------------
 
 # Stdlib
 import ast
 import bdb
 import os
 import sys
 import time
-from io import StringIO
 
 # cProfile was added in Python2.5
 try:
     import cProfile as profile
     import pstats
 except ImportError:
     # profile isn't bundled by default in Debian for license reasons
     try:
         import profile, pstats
     except ImportError:
         profile = pstats = None
 
 # Our own packages
 from IPython.core import debugger, oinspect
 from IPython.core import magic_arguments
 from IPython.core import page
 from IPython.core.error import UsageError
 from IPython.core.macro import Macro
 from IPython.core.magic import (Magics, magics_class, line_magic, cell_magic,
                                 line_cell_magic, on_off, needs_local_scope)
 from IPython.testing.skipdoctest import skip_doctest
 from IPython.utils import py3compat
-from IPython.utils.py3compat import builtin_mod, iteritems
+from IPython.utils.py3compat import builtin_mod, iteritems, PY3
 from IPython.utils.contexts import preserve_keys
 from IPython.utils.io import capture_output
 from IPython.utils.ipstruct import Struct
 from IPython.utils.module_paths import find_mod
 from IPython.utils.path import get_py_filename, unquote_filename, shellglob
 from IPython.utils.timing import clock, clock2
 from IPython.utils.warn import warn, error
 
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Magic implementation classes
 #-----------------------------------------------------------------------------
 
 
 class TimeitResult(object):
     """
     Object returned by the timeit magic with info about the run.
 
     Contain the following attributes :
 
     loops: (int) number of loop done per measurement
     repeat: (int) number of time the mesurement has been repeated
     best: (float) best execusion time / number
     all_runs: (list of float) execusion time of each run (in s)
     compile_time: (float) time of statement compilation (s)
 
     """
 
     def __init__(self, loops, repeat, best, all_runs, compile_time, precision):
         self.loops = loops
         self.repeat = repeat
         self.best = best
         self.all_runs = all_runs
         self.compile_time = compile_time
         self._precision = precision
 
     def _repr_pretty_(self, p , cycle):
          unic =  u"%d loops, best of %d: %s per loop" % (self.loops, self.repeat,
                                             _format_time(self.best, self._precision))
          p.text(u'<TimeitResult : '+unic+u'>')
 
 
 class TimeitTemplateFiller(ast.NodeTransformer):
     "This is quite tightly tied to the template definition above."
     def __init__(self, ast_setup, ast_stmt):
         self.ast_setup = ast_setup
         self.ast_stmt = ast_stmt
 
     def visit_FunctionDef(self, node):
         "Fill in the setup statement"
         self.generic_visit(node)
         if node.name == "inner":
             node.body[:1] = self.ast_setup.body
 
         return node
 
     def visit_For(self, node):
         "Fill in the statement to be timed"
         if getattr(getattr(node.body[0], 'value', None), 'id', None) == 'stmt':
             node.body = self.ast_stmt.body
         return node
 
 
 @magics_class
 class ExecutionMagics(Magics):
     """Magics related to code execution, debugging, profiling, etc.
 
     """
 
     def __init__(self, shell):
         super(ExecutionMagics, self).__init__(shell)
         if profile is None:
             self.prun = self.profile_missing_notice
         # Default execution function used to actually run user code.
         self.default_runner = None
 
     def profile_missing_notice(self, *args, **kwargs):
         error("""\
 The profile module could not be found. It has been removed from the standard
 python packages because of its non-free license. To use profiling, install the
 python-profiler package from non-free.""")
 
     @skip_doctest
     @line_cell_magic
     def prun(self, parameter_s='', cell=None):
 
         """Run a statement through the python code profiler.
 
         Usage, in line mode:
           %prun [options] statement
 
         Usage, in cell mode:
           %%prun [options] [statement]
           code...
           code...
 
         In cell mode, the additional code lines are appended to the (possibly
         empty) statement in the first line.  Cell mode allows you to easily
         profile multiline blocks without having to put them in a separate
         function.
 
         The given statement (which doesn't require quote marks) is run via the
         python profiler in a manner similar to the profile.run() function.
         Namespaces are internally managed to work correctly; profile.run
         cannot be used in IPython because it makes certain assumptions about
         namespaces which do not hold under IPython.
 
         Options:
 
         -l <limit>
           you can place restrictions on what or how much of the
           profile gets printed. The limit value can be:
 
              * A string: only information for function names containing this string
                is printed.
 
              * An integer: only these many lines are printed.
 
              * A float (between 0 and 1): this fraction of the report is printed
                (for example, use a limit of 0.4 to see the topmost 40% only).
 
           You can combine several limits with repeated use of the option. For
           example, ``-l __init__ -l 5`` will print only the topmost 5 lines of
           information about class constructors.
 
         -r
           return the pstats.Stats object generated by the profiling. This
           object has all the information about the profile in it, and you can
           later use it for further analysis or in other functions.
 
         -s <key>
           sort profile by given key. You can provide more than one key
           by using the option several times: '-s key1 -s key2 -s key3...'. The
           default sorting key is 'time'.
 
           The following is copied verbatim from the profile documentation
           referenced below:
 
           When more than one key is provided, additional keys are used as
           secondary criteria when the there is equality in all keys selected
           before them.
 
           Abbreviations can be used for any key names, as long as the
           abbreviation is unambiguous.  The following are the keys currently
           defined:
 
           ============  =====================
           Valid Arg     Meaning
           ============  =====================
           "calls"       call count
           "cumulative"  cumulative time
           "file"        file name
           "module"      file name
           "pcalls"      primitive call count
           "line"        line number
           "name"        function name
           "nfl"         name/file/line
           "stdname"     standard name
           "time"        internal time
           ============  =====================
 
           Note that all sorts on statistics are in descending order (placing
           most time consuming items first), where as name, file, and line number
           searches are in ascending order (i.e., alphabetical). The subtle
           distinction between "nfl" and "stdname" is that the standard name is a
           sort of the name as printed, which means that the embedded line
           numbers get compared in an odd way.  For example, lines 3, 20, and 40
           would (if the file names were the same) appear in the string order
           "20" "3" and "40".  In contrast, "nfl" does a numeric compare of the
           line numbers.  In fact, sort_stats("nfl") is the same as
           sort_stats("name", "file", "line").
 
         -T <filename>
           save profile results as shown on screen to a text
           file. The profile is still shown on screen.
 
         -D <filename>
           save (via dump_stats) profile statistics to given
           filename. This data is in a format understood by the pstats module, and
           is generated by a call to the dump_stats() method of profile
           objects. The profile is still shown on screen.
 
         -q
           suppress output to the pager.  Best used with -T and/or -D above.
 
         If you want to run complete programs under the profiler's control, use
         ``%run -p [prof_opts] filename.py [args to program]`` where prof_opts
         contains profiler specific options as described here.
 
         You can read the complete documentation for the profile module with::
 
           In [1]: import profile; profile.help()
         """
         opts, arg_str = self.parse_options(parameter_s, 'D:l:rs:T:q',
                                            list_all=True, posix=False)
         if cell is not None:
             arg_str += '\n' + cell
         arg_str = self.shell.input_splitter.transform_cell(arg_str)
         return self._run_with_profiler(arg_str, opts, self.shell.user_ns)
 
     def _run_with_profiler(self, code, opts, namespace):
         """
         Run `code` with profiler.  Used by ``%prun`` and ``%run -p``.
 
         Parameters
         ----------
         code : str
             Code to be executed.
         opts : Struct
             Options parsed by `self.parse_options`.
         namespace : dict
             A dictionary for Python namespace (e.g., `self.shell.user_ns`).
 
         """
 
         # Fill default values for unspecified options:
         opts.merge(Struct(D=[''], l=[], s=['time'], T=['']))
 
         prof = profile.Profile()
         try:
             prof = prof.runctx(code, namespace, namespace)
             sys_exit = ''
         except SystemExit:
             sys_exit = """*** SystemExit exception caught in code being profiled."""
 
         stats = pstats.Stats(prof).strip_dirs().sort_stats(*opts.s)
 
         lims = opts.l
         if lims:
             lims = []  # rebuild lims with ints/floats/strings
             for lim in opts.l:
                 try:
                     lims.append(int(lim))
                 except ValueError:
                     try:
                         lims.append(float(lim))
                     except ValueError:
                         lims.append(lim)
 
         # Trap output.
         stdout_trap = StringIO()
         stats_stream = stats.stream
         try:
             stats.stream = stdout_trap
             stats.print_stats(*lims)
         finally:
             stats.stream = stats_stream
 
         output = stdout_trap.getvalue()
         output = output.rstrip()
 
         if 'q' not in opts:
             page.page(output)
         print(sys_exit, end=' ')
 
         dump_file = opts.D[0]
         text_file = opts.T[0]
         if dump_file:
             dump_file = unquote_filename(dump_file)
             prof.dump_stats(dump_file)
             print('\n*** Profile stats marshalled to file',\
                   repr(dump_file)+'.',sys_exit)
         if text_file:
             text_file = unquote_filename(text_file)
             pfile = open(text_file,'w')
             pfile.write(output)
             pfile.close()
             print('\n*** Profile printout saved to text file',\
                   repr(text_file)+'.',sys_exit)
 
         if 'r' in opts:
             return stats
         else:
             return None
 
     @line_magic
     def pdb(self, parameter_s=''):
         """Control the automatic calling of the pdb interactive debugger.
 
         Call as '%pdb on', '%pdb 1', '%pdb off' or '%pdb 0'. If called without
         argument it works as a toggle.
 
         When an exception is triggered, IPython can optionally call the
         interactive pdb debugger after the traceback printout. %pdb toggles
         this feature on and off.
 
         The initial state of this feature is set in your configuration
         file (the option is ``InteractiveShell.pdb``).
 
         If you want to just activate the debugger AFTER an exception has fired,
         without having to type '%pdb on' and rerunning your code, you can use
         the %debug magic."""
 
         par = parameter_s.strip().lower()
 
         if par:
             try:
                 new_pdb = {'off':0,'0':0,'on':1,'1':1}[par]
             except KeyError:
                 print ('Incorrect argument. Use on/1, off/0, '
                        'or nothing for a toggle.')
                 return
         else:
             # toggle
             new_pdb = not self.shell.call_pdb
 
         # set on the shell
         self.shell.call_pdb = new_pdb
         print('Automatic pdb calling has been turned',on_off(new_pdb))
 
     @skip_doctest
     @magic_arguments.magic_arguments()
     @magic_arguments.argument('--breakpoint', '-b', metavar='FILE:LINE',
         help="""
         Set break point at LINE in FILE.
         """
     )
     @magic_arguments.argument('statement', nargs='*',
         help="""
         Code to run in debugger.
         You can omit this in cell magic mode.
         """
     )
     @line_cell_magic
     def debug(self, line='', cell=None):
         """Activate the interactive debugger.
 
         This magic command support two ways of activating debugger.
         One is to activate debugger before executing code.  This way, you
         can set a break point, to step through the code from the point.
         You can use this mode by giving statements to execute and optionally
         a breakpoint.
 
         The other one is to activate debugger in post-mortem mode.  You can
         activate this mode simply running %debug without any argument.
         If an exception has just occurred, this lets you inspect its stack
         frames interactively.  Note that this will always work only on the last
         traceback that occurred, so you must call this quickly after an
         exception that you wish to inspect has fired, because if another one
         occurs, it clobbers the previous one.
 
         If you want IPython to automatically do this on every exception, see
         the %pdb magic for more details.
         """
         args = magic_arguments.parse_argstring(self.debug, line)
 
         if not (args.breakpoint or args.statement or cell):
             self._debug_post_mortem()
         else:
             code = "\n".join(args.statement)
             if cell:
                 code += "\n" + cell
             self._debug_exec(code, args.breakpoint)
 
     def _debug_post_mortem(self):
         self.shell.debugger(force=True)
 
     def _debug_exec(self, code, breakpoint):
         if breakpoint:
             (filename, bp_line) = breakpoint.split(':', 1)
             bp_line = int(bp_line)
         else:
             (filename, bp_line) = (None, None)
         self._run_with_debugger(code, self.shell.user_ns, filename, bp_line)
 
     @line_magic
     def tb(self, s):
         """Print the last traceback with the currently active exception mode.
 
         See %xmode for changing exception reporting modes."""
         self.shell.showtraceback()
 
     @skip_doctest
     @line_magic
     def run(self, parameter_s='', runner=None,
                   file_finder=get_py_filename):
         """Run the named file inside IPython as a program.
 
         Usage::
         
           %run [-n -i -e -G]
                [( -t [-N<N>] | -d [-b<N>] | -p [profile options] )]
                ( -m mod | file ) [args]
 
         Parameters after the filename are passed as command-line arguments to
         the program (put in sys.argv). Then, control returns to IPython's
         prompt.
 
         This is similar to running at a system prompt ``python file args``,
         but with the advantage of giving you IPython's tracebacks, and of
         loading all variables into your interactive namespace for further use
         (unless -p is used, see below).
 
         The file is executed in a namespace initially consisting only of
         ``__name__=='__main__'`` and sys.argv constructed as indicated. It thus
         sees its environment as if it were being run as a stand-alone program
         (except for sharing global objects such as previously imported
         modules). But after execution, the IPython interactive namespace gets
         updated with all variables defined in the program (except for __name__
         and sys.argv). This allows for very convenient loading of code for
         interactive work, while giving each program a 'clean sheet' to run in.
 
         Arguments are expanded using shell-like glob match.  Patterns
         '*', '?', '[seq]' and '[!seq]' can be used.  Additionally,
         tilde '~' will be expanded into user's home directory.  Unlike
         real shells, quotation does not suppress expansions.  Use
         *two* back slashes (e.g. ``\\\\*``) to suppress expansions.
         To completely disable these expansions, you can use -G flag.
 
         Options:
 
         -n
           __name__ is NOT set to '__main__', but to the running file's name
           without extension (as python does under import).  This allows running
           scripts and reloading the definitions in them without calling code
           protected by an ``if __name__ == "__main__"`` clause.
 
         -i
           run the file in IPython's namespace instead of an empty one. This
           is useful if you are experimenting with code written in a text editor
           which depends on variables defined interactively.
 
         -e
           ignore sys.exit() calls or SystemExit exceptions in the script
           being run.  This is particularly useful if IPython is being used to
           run unittests, which always exit with a sys.exit() call.  In such
           cases you are interested in the output of the test results, not in
           seeing a traceback of the unittest module.
 
         -t
           print timing information at the end of the run.  IPython will give
           you an estimated CPU time consumption for your script, which under
           Unix uses the resource module to avoid the wraparound problems of
           time.clock().  Under Unix, an estimate of time spent on system tasks
           is also given (for Windows platforms this is reported as 0.0).
 
         If -t is given, an additional ``-N<N>`` option can be given, where <N>
         must be an integer indicating how many times you want the script to
         run.  The final timing report will include total and per run results.
 
         For example (testing the script uniq_stable.py)::
 
             In [1]: run -t uniq_stable
 
             IPython CPU timings (estimated):
               User  :    0.19597 s.
               System:        0.0 s.
 
             In [2]: run -t -N5 uniq_stable
 
             IPython CPU timings (estimated):
             Total runs performed: 5
               Times :      Total       Per run
               User  :   0.910862 s,  0.1821724 s.
               System:        0.0 s,        0.0 s.
 
         -d
           run your program under the control of pdb, the Python debugger.
           This allows you to execute your program step by step, watch variables,
           etc.  Internally, what IPython does is similar to calling::
 
               pdb.run('execfile("YOURFILENAME")')
 
           with a breakpoint set on line 1 of your file.  You can change the line
           number for this automatic breakpoint to be <N> by using the -bN option
           (where N must be an integer). For example::
 
               %run -d -b40 myscript
 
           will set the first breakpoint at line 40 in myscript.py.  Note that
           the first breakpoint must be set on a line which actually does
           something (not a comment or docstring) for it to stop execution.
 
           Or you can specify a breakpoint in a different file::
 
               %run -d -b myotherfile.py:20 myscript
 
           When the pdb debugger starts, you will see a (Pdb) prompt.  You must
           first enter 'c' (without quotes) to start execution up to the first
           breakpoint.
 
           Entering 'help' gives information about the use of the debugger.  You
           can easily see pdb's full documentation with "import pdb;pdb.help()"
           at a prompt.
 
         -p
           run program under the control of the Python profiler module (which
           prints a detailed report of execution times, function calls, etc).
 
           You can pass other options after -p which affect the behavior of the
           profiler itself. See the docs for %prun for details.
 
           In this mode, the program's variables do NOT propagate back to the
           IPython interactive namespace (because they remain in the namespace
           where the profiler executes them).
 
           Internally this triggers a call to %prun, see its documentation for
           details on the options available specifically for profiling.
 
         There is one special usage for which the text above doesn't apply:
         if the filename ends with .ipy, the file is run as ipython script,
         just as if the commands were written on IPython prompt.
 
         -m
           specify module name to load instead of script path. Similar to
           the -m option for the python interpreter. Use this option last if you
           want to combine with other %run options. Unlike the python interpreter
           only source modules are allowed no .pyc or .pyo files.
           For example::
 
               %run -m example
 
           will run the example module.
 
         -G
           disable shell-like glob expansion of arguments.
 
         """
 
         # get arguments and set sys.argv for program to be run.
         opts, arg_lst = self.parse_options(parameter_s,
                                            'nidtN:b:pD:l:rs:T:em:G',
                                            mode='list', list_all=1)
         if "m" in opts:
             modulename = opts["m"][0]
             modpath = find_mod(modulename)
             if modpath is None:
                 warn('%r is not a valid modulename on sys.path'%modulename)
                 return
             arg_lst = [modpath] + arg_lst
         try:
             filename = file_finder(arg_lst[0])
         except IndexError:
             warn('you must provide at least a filename.')
             print('\n%run:\n', oinspect.getdoc(self.run))
             return
         except IOError as e:
             try:
                 msg = str(e)
             except UnicodeError:
                 msg = e.message
             error(msg)
             return
 
         if filename.lower().endswith('.ipy'):
             with preserve_keys(self.shell.user_ns, '__file__'):
                 self.shell.user_ns['__file__'] = filename
                 self.shell.safe_execfile_ipy(filename)
             return
 
         # Control the response to exit() calls made by the script being run
         exit_ignore = 'e' in opts
 
         # Make sure that the running script gets a proper sys.argv as if it
         # were run from a system shell.
         save_argv = sys.argv # save it for later restoring
 
         if 'G' in opts:
             args = arg_lst[1:]
         else:
             # tilde and glob expansion
             args = shellglob(map(os.path.expanduser,  arg_lst[1:]))
 
         sys.argv = [filename] + args  # put in the proper filename
         # protect sys.argv from potential unicode strings on Python 2:
         if not py3compat.PY3:
             sys.argv = [ py3compat.cast_bytes(a) for a in sys.argv ]
 
         if 'i' in opts:
             # Run in user's interactive namespace
             prog_ns = self.shell.user_ns
             __name__save = self.shell.user_ns['__name__']
             prog_ns['__name__'] = '__main__'
             main_mod = self.shell.user_module
             
             # Since '%run foo' emulates 'python foo.py' at the cmd line, we must
             # set the __file__ global in the script's namespace
             # TK: Is this necessary in interactive mode?
             prog_ns['__file__'] = filename
         else:
             # Run in a fresh, empty namespace
             if 'n' in opts:
                 name = os.path.splitext(os.path.basename(filename))[0]
             else:
                 name = '__main__'
 
             # The shell MUST hold a reference to prog_ns so after %run
             # exits, the python deletion mechanism doesn't zero it out
             # (leaving dangling references). See interactiveshell for details
             main_mod = self.shell.new_main_mod(filename, name)
             prog_ns = main_mod.__dict__
 
         # pickle fix.  See interactiveshell for an explanation.  But we need to
         # make sure that, if we overwrite __main__, we replace it at the end
         main_mod_name = prog_ns['__name__']
 
         if main_mod_name == '__main__':
             restore_main = sys.modules['__main__']
         else:
             restore_main = False
 
         # This needs to be undone at the end to prevent holding references to
         # every single object ever created.
         sys.modules[main_mod_name] = main_mod
 
         if 'p' in opts or 'd' in opts:
             if 'm' in opts:
                 code = 'run_module(modulename, prog_ns)'
                 code_ns = {
                     'run_module': self.shell.safe_run_module,
                     'prog_ns': prog_ns,
                     'modulename': modulename,
                 }
             else:
                 code = 'execfile(filename, prog_ns)'
                 code_ns = {
                     'execfile': self.shell.safe_execfile,
                     'prog_ns': prog_ns,
                     'filename': get_py_filename(filename),
                 }
 
         try:
             stats = None
             with self.shell.readline_no_record:
                 if 'p' in opts:
                     stats = self._run_with_profiler(code, opts, code_ns)
                 else:
                     if 'd' in opts:
                         bp_file, bp_line = parse_breakpoint(
                             opts.get('b', ['1'])[0], filename)
                         self._run_with_debugger(
                             code, code_ns, filename, bp_line, bp_file)
                     else:
                         if 'm' in opts:
                             def run():
                                 self.shell.safe_run_module(modulename, prog_ns)
                         else:
                             if runner is None:
                                 runner = self.default_runner
                             if runner is None:
                                 runner = self.shell.safe_execfile
 
                             def run():
                                 runner(filename, prog_ns, prog_ns,
                                        exit_ignore=exit_ignore)
 
                         if 't' in opts:
                             # timed execution
                             try:
                                 nruns = int(opts['N'][0])
                                 if nruns < 1:
                                     error('Number of runs must be >=1')
                                     return
                             except (KeyError):
                                 nruns = 1
                             self._run_with_timing(run, nruns)
                         else:
                             # regular execution
                             run()
 
                 if 'i' in opts:
                     self.shell.user_ns['__name__'] = __name__save
                 else:
                     # update IPython interactive namespace
 
                     # Some forms of read errors on the file may mean the
                     # __name__ key was never set; using pop we don't have to
                     # worry about a possible KeyError.
                     prog_ns.pop('__name__', None)
 
                     with preserve_keys(self.shell.user_ns, '__file__'):
                         self.shell.user_ns.update(prog_ns)
         finally:
             # It's a bit of a mystery why, but __builtins__ can change from
             # being a module to becoming a dict missing some key data after
             # %run.  As best I can see, this is NOT something IPython is doing
             # at all, and similar problems have been reported before:
             # http://coding.derkeiler.com/Archive/Python/comp.lang.python/2004-10/0188.html
             # Since this seems to be done by the interpreter itself, the best
             # we can do is to at least restore __builtins__ for the user on
             # exit.
             self.shell.user_ns['__builtins__'] = builtin_mod
 
             # Ensure key global structures are restored
             sys.argv = save_argv
             if restore_main:
                 sys.modules['__main__'] = restore_main
             else:
                 # Remove from sys.modules the reference to main_mod we'd
                 # added.  Otherwise it will trap references to objects
                 # contained therein.
                 del sys.modules[main_mod_name]
 
         return stats
 
     def _run_with_debugger(self, code, code_ns, filename=None,
                            bp_line=None, bp_file=None):
         """
         Run `code` in debugger with a break point.
 
         Parameters
         ----------
         code : str
             Code to execute.
         code_ns : dict
             A namespace in which `code` is executed.
         filename : str
             `code` is ran as if it is in `filename`.
         bp_line : int, optional
             Line number of the break point.
         bp_file : str, optional
             Path to the file in which break point is specified.
             `filename` is used if not given.
 
         Raises
         ------
         UsageError
             If the break point given by `bp_line` is not valid.
 
         """
         deb = debugger.Pdb(self.shell.colors)
         # reset Breakpoint state, which is moronically kept
         # in a class
         bdb.Breakpoint.next = 1
         bdb.Breakpoint.bplist = {}
         bdb.Breakpoint.bpbynumber = [None]
         if bp_line is not None:
             # Set an initial breakpoint to stop execution
             maxtries = 10
             bp_file = bp_file or filename
             checkline = deb.checkline(bp_file, bp_line)
             if not checkline:
                 for bp in range(bp_line + 1, bp_line + maxtries + 1):
                     if deb.checkline(bp_file, bp):
                         break
                 else:
                     msg = ("\nI failed to find a valid line to set "
                            "a breakpoint\n"
                            "after trying up to line: %s.\n"
                            "Please set a valid breakpoint manually "
                            "with the -b option." % bp)
                     raise UsageError(msg)
             # if we find a good linenumber, set the breakpoint
             deb.do_break('%s:%s' % (bp_file, bp_line))
 
         if filename:
             # Mimic Pdb._runscript(...)
             deb._wait_for_mainpyfile = True
             deb.mainpyfile = deb.canonic(filename)
 
         # Start file run
         print("NOTE: Enter 'c' at the %s prompt to continue execution." % deb.prompt)
         try:
             if filename:
                 # save filename so it can be used by methods on the deb object
                 deb._exec_filename = filename
             deb.run(code, code_ns)
 
         except:
             etype, value, tb = sys.exc_info()
             # Skip three frames in the traceback: the %run one,
             # one inside bdb.py, and the command-line typed by the
             # user (run by exec in pdb itself).
             self.shell.InteractiveTB(etype, value, tb, tb_offset=3)
 
     @staticmethod
     def _run_with_timing(run, nruns):
         """
         Run function `run` and print timing information.
 
         Parameters
         ----------
         run : callable
             Any callable object which takes no argument.
         nruns : int
             Number of times to execute `run`.
 
         """
         twall0 = time.time()
         if nruns == 1:
             t0 = clock2()
             run()
             t1 = clock2()
             t_usr = t1[0] - t0[0]
             t_sys = t1[1] - t0[1]
             print("\nIPython CPU timings (estimated):")
             print("  User   : %10.2f s." % t_usr)
             print("  System : %10.2f s." % t_sys)
         else:
             runs = range(nruns)
             t0 = clock2()
             for nr in runs:
                 run()
             t1 = clock2()
             t_usr = t1[0] - t0[0]
             t_sys = t1[1] - t0[1]
             print("\nIPython CPU timings (estimated):")
             print("Total runs performed:", nruns)
             print("  Times  : %10s   %10s" % ('Total', 'Per run'))
             print("  User   : %10.2f s, %10.2f s." % (t_usr, t_usr / nruns))
             print("  System : %10.2f s, %10.2f s." % (t_sys, t_sys / nruns))
         twall1 = time.time()
         print("Wall time: %10.2f s." % (twall1 - twall0))
 
     @skip_doctest
     @line_cell_magic
     def timeit(self, line='', cell=None):
         """Time execution of a Python statement or expression
 
         Usage, in line mode:
           %timeit [-n<N> -r<R> [-t|-c] -q -p<P> -o] statement
         or in cell mode:
           %%timeit [-n<N> -r<R> [-t|-c] -q -p<P> -o] setup_code
           code
           code...
 
         Time execution of a Python statement or expression using the timeit
         module.  This function can be used both as a line and cell magic:
 
         - In line mode you can time a single-line statement (though multiple
           ones can be chained with using semicolons).
 
         - In cell mode, the statement in the first line is used as setup code
           (executed but not timed) and the body of the cell is timed.  The cell
           body has access to any variables created in the setup code.
 
         Options:
         -n<N>: execute the given statement <N> times in a loop. If this value
         is not given, a fitting value is chosen.
 
         -r<R>: repeat the loop iteration <R> times and take the best result.
         Default: 3
 
         -t: use time.time to measure the time, which is the default on Unix.
         This function measures wall time.
 
         -c: use time.clock to measure the time, which is the default on
         Windows and measures wall time. On Unix, resource.getrusage is used
         instead and returns the CPU user time.
 
         -p<P>: use a precision of <P> digits to display the timing result.
         Default: 3
 
         -q: Quiet, do not print result.
 
         -o: return a TimeitResult that can be stored in a variable to inspect
             the result in more details.
 
 
         Examples
         --------
         ::
 
           In [1]: %timeit pass
           10000000 loops, best of 3: 53.3 ns per loop
 
           In [2]: u = None
 
           In [3]: %timeit u is None
           10000000 loops, best of 3: 184 ns per loop
 
           In [4]: %timeit -r 4 u == None
           1000000 loops, best of 4: 242 ns per loop
 
           In [5]: import time
 
           In [6]: %timeit -n1 time.sleep(2)
           1 loops, best of 3: 2 s per loop
 
 
         The times reported by %timeit will be slightly higher than those
         reported by the timeit.py script when variables are accessed. This is
         due to the fact that %timeit executes the statement in the namespace
         of the shell, compared with timeit.py, which uses a single setup
         statement to import function or create variables. Generally, the bias
         does not matter as long as results from timeit.py are not mixed with
         those from %timeit."""
 
         import timeit
 
         opts, stmt = self.parse_options(line,'n:r:tcp:qo',
                                         posix=False, strict=False)
         if stmt == "" and cell is None:
             return
         
         timefunc = timeit.default_timer
         number = int(getattr(opts, "n", 0))
         repeat = int(getattr(opts, "r", timeit.default_repeat))
         precision = int(getattr(opts, "p", 3))
         quiet = 'q' in opts
         return_result = 'o' in opts
         if hasattr(opts, "t"):
             timefunc = time.time
         if hasattr(opts, "c"):
             timefunc = clock
 
         timer = timeit.Timer(timer=timefunc)
         # this code has tight coupling to the inner workings of timeit.Timer,
         # but is there a better way to achieve that the code stmt has access
         # to the shell namespace?
         transform  = self.shell.input_splitter.transform_cell
 
         if cell is None:
             # called as line magic
             ast_setup = ast.parse("pass")
             ast_stmt = ast.parse(transform(stmt))
         else:
             ast_setup = ast.parse(transform(stmt))
             ast_stmt = ast.parse(transform(cell))
 
         ast_setup = self.shell.transform_ast(ast_setup)
         ast_stmt = self.shell.transform_ast(ast_stmt)
 
         # This codestring is taken from timeit.template - we fill it in as an
         # AST, so that we can apply our AST transformations to the user code
         # without affecting the timing code.
         timeit_ast_template = ast.parse('def inner(_it, _timer):\n'
                                         '    setup\n'
                                         '    _t0 = _timer()\n'
                                         '    for _i in _it:\n'
                                         '        stmt\n'
                                         '    _t1 = _timer()\n'
                                         '    return _t1 - _t0\n')
 
         timeit_ast = TimeitTemplateFiller(ast_setup, ast_stmt).visit(timeit_ast_template)
         timeit_ast = ast.fix_missing_locations(timeit_ast)
 
         # Track compilation time so it can be reported if too long
         # Minimum time above which compilation time will be reported
         tc_min = 0.1
 
         t0 = clock()
         code = compile(timeit_ast, "<magic-timeit>", "exec")
         tc = clock()-t0
 
         ns = {}
         exec(code, self.shell.user_ns, ns)
         timer.inner = ns["inner"]
 
         if number == 0:
             # determine number so that 0.2 <= total time < 2.0
             number = 1
             for _ in range(1, 10):
                 if timer.timeit(number) >= 0.2:
                     break
                 number *= 10
         all_runs = timer.repeat(repeat, number)
         best = min(all_runs) / number
         if not quiet :
             print(u"%d loops, best of %d: %s per loop" % (number, repeat,
                                                               _format_time(best, precision)))
             if tc > tc_min:
                 print("Compiler time: %.2f s" % tc)
         if return_result:
             return TimeitResult(number, repeat, best, all_runs, tc, precision)
 
     @skip_doctest
     @needs_local_scope
     @line_cell_magic
     def time(self,line='', cell=None, local_ns=None):
         """Time execution of a Python statement or expression.
 
         The CPU and wall clock times are printed, and the value of the
         expression (if any) is returned.  Note that under Win32, system time
         is always reported as 0, since it can not be measured.
         
         This function can be used both as a line and cell magic:
 
         - In line mode you can time a single-line statement (though multiple
           ones can be chained with using semicolons).
 
         - In cell mode, you can time the cell body (a directly 
           following statement raises an error).
 
         This function provides very basic timing functionality.  Use the timeit 
         magic for more controll over the measurement.
 
         Examples
         --------
         ::
 
           In [1]: %time 2**128
           CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
           Wall time: 0.00
           Out[1]: 340282366920938463463374607431768211456L
 
           In [2]: n = 1000000
 
           In [3]: %time sum(range(n))
           CPU times: user 1.20 s, sys: 0.05 s, total: 1.25 s
           Wall time: 1.37
           Out[3]: 499999500000L
 
           In [4]: %time print 'hello world'
           hello world
           CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
           Wall time: 0.00
 
           Note that the time needed by Python to compile the given expression
           will be reported if it is more than 0.1s.  In this example, the
           actual exponentiation is done by Python at compilation time, so while
           the expression can take a noticeable amount of time to compute, that
           time is purely due to the compilation:
 
           In [5]: %time 3**9999;
           CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
           Wall time: 0.00 s
 
           In [6]: %time 3**999999;
           CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
           Wall time: 0.00 s
           Compiler : 0.78 s
           """
 
         # fail immediately if the given expression can't be compiled
         
         if line and cell:
             raise UsageError("Can't use statement directly after '%%time'!")
         
         if cell:
             expr = self.shell.input_transformer_manager.transform_cell(cell)
         else:
             expr = self.shell.input_transformer_manager.transform_cell(line)
 
         # Minimum time above which parse time will be reported
         tp_min = 0.1
 
         t0 = clock()
         expr_ast = ast.parse(expr)
         tp = clock()-t0
 
         # Apply AST transformations
         expr_ast = self.shell.transform_ast(expr_ast)
 
         # Minimum time above which compilation time will be reported
         tc_min = 0.1
 
         if len(expr_ast.body)==1 and isinstance(expr_ast.body[0], ast.Expr):
             mode = 'eval'
             source = '<timed eval>'
             expr_ast = ast.Expression(expr_ast.body[0].value)
         else:
             mode = 'exec'
             source = '<timed exec>'
         t0 = clock()
         code = compile(expr_ast, source, mode)
         tc = clock()-t0
 
         # skew measurement as little as possible
         glob = self.shell.user_ns
         wtime = time.time
         # time execution
         wall_st = wtime()
         if mode=='eval':
             st = clock2()
             out = eval(code, glob, local_ns)
             end = clock2()
         else:
             st = clock2()
             exec(code, glob, local_ns)
             end = clock2()
             out = None
         wall_end = wtime()
         # Compute actual times and report
         wall_time = wall_end-wall_st
         cpu_user = end[0]-st[0]
         cpu_sys = end[1]-st[1]
         cpu_tot = cpu_user+cpu_sys
         # On windows cpu_sys is always zero, so no new information to the next print 
         if sys.platform != 'win32':
             print("CPU times: user %s, sys: %s, total: %s" % \
                 (_format_time(cpu_user),_format_time(cpu_sys),_format_time(cpu_tot)))
         print("Wall time: %s" % _format_time(wall_time))
         if tc > tc_min:
             print("Compiler : %s" % _format_time(tc))
         if tp > tp_min:
             print("Parser   : %s" % _format_time(tp))
         return out
 
     @skip_doctest
     @line_magic
     def macro(self, parameter_s=''):
         """Define a macro for future re-execution. It accepts ranges of history,
         filenames or string objects.
 
         Usage:\\
           %macro [options] name n1-n2 n3-n4 ... n5 .. n6 ...
 
         Options:
 
           -r: use 'raw' input.  By default, the 'processed' history is used,
           so that magics are loaded in their transformed version to valid
           Python.  If this option is given, the raw input as typed at the
           command line is used instead.
           
           -q: quiet macro definition.  By default, a tag line is printed 
           to indicate the macro has been created, and then the contents of 
           the macro are printed.  If this option is given, then no printout
           is produced once the macro is created.
 
         This will define a global variable called `name` which is a string
         made of joining the slices and lines you specify (n1,n2,... numbers
         above) from your input history into a single string. This variable
         acts like an automatic function which re-executes those lines as if
         you had typed them. You just type 'name' at the prompt and the code
         executes.
 
         The syntax for indicating input ranges is described in %history.
 
         Note: as a 'hidden' feature, you can also use traditional python slice
         notation, where N:M means numbers N through M-1.
 
         For example, if your history contains (print using %hist -n )::
 
           44: x=1
           45: y=3
           46: z=x+y
           47: print x
           48: a=5
           49: print 'x',x,'y',y
 
         you can create a macro with lines 44 through 47 (included) and line 49
         called my_macro with::
 
           In [55]: %macro my_macro 44-47 49
 
         Now, typing `my_macro` (without quotes) will re-execute all this code
         in one pass.
 
         You don't need to give the line-numbers in order, and any given line
         number can appear multiple times. You can assemble macros with any
         lines from your input history in any order.
 
         The macro is a simple object which holds its value in an attribute,
         but IPython's display system checks for macros and executes them as
         code instead of printing them when you type their name.
 
         You can view a macro's contents by explicitly printing it with::
 
           print macro_name
 
         """
         opts,args = self.parse_options(parameter_s,'rq',mode='list')
         if not args:   # List existing macros
             return sorted(k for k,v in iteritems(self.shell.user_ns) if\
                                                         isinstance(v, Macro))
         if len(args) == 1:
             raise UsageError(
                 "%macro insufficient args; usage '%macro name n1-n2 n3-4...")
         name, codefrom = args[0], " ".join(args[1:])
 
         #print 'rng',ranges  # dbg
         try:
             lines = self.shell.find_user_code(codefrom, 'r' in opts)
         except (ValueError, TypeError) as e:
             print(e.args[0])
             return
         macro = Macro(lines)
         self.shell.define_macro(name, macro)
         if not ( 'q' in opts) : 
             print('Macro `%s` created. To execute, type its name (without quotes).' % name)
             print('=== Macro contents: ===')
             print(macro, end=' ')
 
     @magic_arguments.magic_arguments()
     @magic_arguments.argument('output', type=str, default='', nargs='?',
         help="""The name of the variable in which to store output.
         This is a utils.io.CapturedIO object with stdout/err attributes
         for the text of the captured output.
 
         CapturedOutput also has a show() method for displaying the output,
         and __call__ as well, so you can use that to quickly display the
         output.
 
         If unspecified, captured output is discarded.
         """
     )
     @magic_arguments.argument('--no-stderr', action="store_true",
         help="""Don't capture stderr."""
     )
     @magic_arguments.argument('--no-stdout', action="store_true",
         help="""Don't capture stdout."""
     )
     @magic_arguments.argument('--no-display', action="store_true",
         help="""Don't capture IPython's rich display."""
     )
     @cell_magic
     def capture(self, line, cell):
         """run the cell, capturing stdout, stderr, and IPython's rich display() calls."""
         args = magic_arguments.parse_argstring(self.capture, line)
         out = not args.no_stdout
         err = not args.no_stderr
         disp = not args.no_display
         with capture_output(out, err, disp) as io:
             self.shell.run_cell(cell)
         if args.output:
             self.shell.user_ns[args.output] = io
 
 def parse_breakpoint(text, current_file):
     '''Returns (file, line) for file:line and (current_file, line) for line'''
     colon = text.find(':')
     if colon == -1:
         return current_file, int(text)
     else:
         return text[:colon], int(text[colon+1:])
     
 def _format_time(timespan, precision=3):
     """Formats the timespan in a human readable form"""
     import math
     
     if timespan >= 60.0:
         # we have more than a minute, format that in a human readable form
         # Idea from http://snipplr.com/view/5713/
         parts = [("d", 60*60*24),("h", 60*60),("min", 60), ("s", 1)]
         time = []
         leftover = timespan
         for suffix, length in parts:
             value = int(leftover / length)
             if value > 0:
                 leftover = leftover % length
                 time.append(u'%s%s' % (str(value), suffix))
             if leftover < 1:
                 break
         return " ".join(time)
 
     
     # Unfortunately the unicode 'micro' symbol can cause problems in
     # certain terminals.  
     # See bug: https://bugs.launchpad.net/ipython/+bug/348466
     # Try to prevent crashes by being more secure than it needs to
     # E.g. eclipse is able to print a µ, but has no sys.stdout.encoding set.
     units = [u"s", u"ms",u'us',"ns"] # the save value   
     if hasattr(sys.stdout, 'encoding') and sys.stdout.encoding:
         try:
             u'\xb5'.encode(sys.stdout.encoding)
             units = [u"s", u"ms",u'\xb5s',"ns"]
         except:
             pass
     scaling = [1, 1e3, 1e6, 1e9]
         
     if timespan > 0.0:
         order = min(-int(math.floor(math.log10(timespan)) // 3), 3)
     else:
         order = 3
     return u"%.*g %s" % (precision, timespan * scaling[order], units[order])

IPython/core/tests/test_interactiveshell.py

@@ -1,672 +1,676
 # -*- coding: utf-8 -*-
 """Tests for the key interactiveshell module.
 
 Historically the main classes in interactiveshell have been under-tested.  This
 module should grow as many single-method tests as possible to trap many of the
 recurring bugs we seem to encounter with high-level interaction.
 
 Authors
 -------
 * Fernando Perez
 """
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 # stdlib
 import ast
 import os
 import signal
 import shutil
 import sys
 import tempfile
 import unittest
 from os.path import join
-from io import StringIO
 
 # third-party
 import nose.tools as nt
 
 # Our own
 from IPython.testing.decorators import skipif, skip_win32, onlyif_unicode_paths
 from IPython.testing import tools as tt
 from IPython.utils import io
-from IPython.utils.py3compat import unicode_type
+from IPython.utils.py3compat import unicode_type, PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 # This is used by every single test, no point repeating it ad nauseam
 ip = get_ipython()
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 class InteractiveShellTestCase(unittest.TestCase):
     def test_naked_string_cells(self):
         """Test that cells with only naked strings are fully executed"""
         # First, single-line inputs
         ip.run_cell('"a"\n')
         self.assertEqual(ip.user_ns['_'], 'a')
         # And also multi-line cells
         ip.run_cell('"""a\nb"""\n')
         self.assertEqual(ip.user_ns['_'], 'a\nb')
 
     def test_run_empty_cell(self):
         """Just make sure we don't get a horrible error with a blank
         cell of input. Yes, I did overlook that."""
         old_xc = ip.execution_count
         ip.run_cell('')
         self.assertEqual(ip.execution_count, old_xc)
 
     def test_run_cell_multiline(self):
         """Multi-block, multi-line cells must execute correctly.
         """
         src = '\n'.join(["x=1",
                          "y=2",
                          "if 1:",
                          "    x += 1",
                          "    y += 1",])
         ip.run_cell(src)
         self.assertEqual(ip.user_ns['x'], 2)
         self.assertEqual(ip.user_ns['y'], 3)
 
     def test_multiline_string_cells(self):
         "Code sprinkled with multiline strings should execute (GH-306)"
         ip.run_cell('tmp=0')
         self.assertEqual(ip.user_ns['tmp'], 0)
         ip.run_cell('tmp=1;"""a\nb"""\n')
         self.assertEqual(ip.user_ns['tmp'], 1)
 
     def test_dont_cache_with_semicolon(self):
         "Ending a line with semicolon should not cache the returned object (GH-307)"
         oldlen = len(ip.user_ns['Out'])
         a = ip.run_cell('1;', store_history=True)
         newlen = len(ip.user_ns['Out'])
         self.assertEqual(oldlen, newlen)
         #also test the default caching behavior
         ip.run_cell('1', store_history=True)
         newlen = len(ip.user_ns['Out'])
         self.assertEqual(oldlen+1, newlen)
 
     def test_In_variable(self):
         "Verify that In variable grows with user input (GH-284)"
         oldlen = len(ip.user_ns['In'])
         ip.run_cell('1;', store_history=True)
         newlen = len(ip.user_ns['In'])
         self.assertEqual(oldlen+1, newlen)
         self.assertEqual(ip.user_ns['In'][-1],'1;')
         
     def test_magic_names_in_string(self):
         ip.run_cell('a = """\n%exit\n"""')
         self.assertEqual(ip.user_ns['a'], '\n%exit\n')
     
     def test_trailing_newline(self):
         """test that running !(command) does not raise a SyntaxError"""
         ip.run_cell('!(true)\n', False)
         ip.run_cell('!(true)\n\n\n', False)
     
     def test_gh_597(self):
         """Pretty-printing lists of objects with non-ascii reprs may cause
         problems."""
         class Spam(object):
           def __repr__(self):
             return "\xe9"*50
         import IPython.core.formatters
         f = IPython.core.formatters.PlainTextFormatter()
         f([Spam(),Spam()])
     
 
     def test_future_flags(self):
         """Check that future flags are used for parsing code (gh-777)"""
         ip.run_cell('from __future__ import print_function')
         try:
             ip.run_cell('prfunc_return_val = print(1,2, sep=" ")')
             assert 'prfunc_return_val' in ip.user_ns
         finally:
             # Reset compiler flags so we don't mess up other tests.
             ip.compile.reset_compiler_flags()
 
     def test_future_unicode(self):
         """Check that unicode_literals is imported from __future__ (gh #786)"""
         try:
             ip.run_cell(u'byte_str = "a"')
             assert isinstance(ip.user_ns['byte_str'], str) # string literals are byte strings by default
             ip.run_cell('from __future__ import unicode_literals')
             ip.run_cell(u'unicode_str = "a"')
             assert isinstance(ip.user_ns['unicode_str'], unicode_type) # strings literals are now unicode
         finally:
             # Reset compiler flags so we don't mess up other tests.
             ip.compile.reset_compiler_flags()
     
     def test_can_pickle(self):
         "Can we pickle objects defined interactively (GH-29)"
         ip = get_ipython()
         ip.reset()
         ip.run_cell(("class Mylist(list):\n"
                      "    def __init__(self,x=[]):\n"
                      "        list.__init__(self,x)"))
         ip.run_cell("w=Mylist([1,2,3])")
         
         from pickle import dumps
         
         # We need to swap in our main module - this is only necessary
         # inside the test framework, because IPython puts the interactive module
         # in place (but the test framework undoes this).
         _main = sys.modules['__main__']
         sys.modules['__main__'] = ip.user_module
         try:
             res = dumps(ip.user_ns["w"])
         finally:
             sys.modules['__main__'] = _main
         self.assertTrue(isinstance(res, bytes))
         
     def test_global_ns(self):
         "Code in functions must be able to access variables outside them."
         ip = get_ipython()
         ip.run_cell("a = 10")
         ip.run_cell(("def f(x):\n"
                      "    return x + a"))
         ip.run_cell("b = f(12)")
         self.assertEqual(ip.user_ns["b"], 22)
 
     def test_bad_custom_tb(self):
         """Check that InteractiveShell is protected from bad custom exception handlers"""
         from IPython.utils import io
         save_stderr = io.stderr
         try:
             # capture stderr
             io.stderr = StringIO()
             ip.set_custom_exc((IOError,), lambda etype,value,tb: 1/0)
             self.assertEqual(ip.custom_exceptions, (IOError,))
             ip.run_cell(u'raise IOError("foo")')
             self.assertEqual(ip.custom_exceptions, ())
             self.assertTrue("Custom TB Handler failed" in io.stderr.getvalue())
         finally:
             io.stderr = save_stderr
 
     def test_bad_custom_tb_return(self):
         """Check that InteractiveShell is protected from bad return types in custom exception handlers"""
         from IPython.utils import io
         save_stderr = io.stderr
         try:
             # capture stderr
             io.stderr = StringIO()
             ip.set_custom_exc((NameError,),lambda etype,value,tb, tb_offset=None: 1)
             self.assertEqual(ip.custom_exceptions, (NameError,))
             ip.run_cell(u'a=abracadabra')
             self.assertEqual(ip.custom_exceptions, ())
             self.assertTrue("Custom TB Handler failed" in io.stderr.getvalue())
         finally:
             io.stderr = save_stderr
 
     def test_drop_by_id(self):
         myvars = {"a":object(), "b":object(), "c": object()}
         ip.push(myvars, interactive=False)
         for name in myvars:
             assert name in ip.user_ns, name
             assert name in ip.user_ns_hidden, name
         ip.user_ns['b'] = 12
         ip.drop_by_id(myvars)
         for name in ["a", "c"]:
             assert name not in ip.user_ns, name
             assert name not in ip.user_ns_hidden, name
         assert ip.user_ns['b'] == 12
         ip.reset()
 
     def test_var_expand(self):
         ip.user_ns['f'] = u'Ca\xf1o'
         self.assertEqual(ip.var_expand(u'echo $f'), u'echo Ca\xf1o')
         self.assertEqual(ip.var_expand(u'echo {f}'), u'echo Ca\xf1o')
         self.assertEqual(ip.var_expand(u'echo {f[:-1]}'), u'echo Ca\xf1')
         self.assertEqual(ip.var_expand(u'echo {1*2}'), u'echo 2')
 
         ip.user_ns['f'] = b'Ca\xc3\xb1o'
         # This should not raise any exception:
         ip.var_expand(u'echo $f')
     
     def test_var_expand_local(self):
         """Test local variable expansion in !system and %magic calls"""
         # !system
         ip.run_cell('def test():\n'
                     '    lvar = "ttt"\n'
                     '    ret = !echo {lvar}\n'
                     '    return ret[0]\n')
         res = ip.user_ns['test']()
         nt.assert_in('ttt', res)
         
         # %magic
         ip.run_cell('def makemacro():\n'
                     '    macroname = "macro_var_expand_locals"\n'
                     '    %macro {macroname} codestr\n')
         ip.user_ns['codestr'] = "str(12)"
         ip.run_cell('makemacro()')
         nt.assert_in('macro_var_expand_locals', ip.user_ns)
     
     def test_var_expand_self(self):
         """Test variable expansion with the name 'self', which was failing.
         
         See https://github.com/ipython/ipython/issues/1878#issuecomment-7698218
         """
         ip.run_cell('class cTest:\n'
                     '  classvar="see me"\n'
                     '  def test(self):\n'
                     '    res = !echo Variable: {self.classvar}\n'
                     '    return res[0]\n')
         nt.assert_in('see me', ip.user_ns['cTest']().test())
     
     def test_bad_var_expand(self):
         """var_expand on invalid formats shouldn't raise"""
         # SyntaxError
         self.assertEqual(ip.var_expand(u"{'a':5}"), u"{'a':5}")
         # NameError
         self.assertEqual(ip.var_expand(u"{asdf}"), u"{asdf}")
         # ZeroDivisionError
         self.assertEqual(ip.var_expand(u"{1/0}"), u"{1/0}")
     
     def test_silent_nopostexec(self):
         """run_cell(silent=True) doesn't invoke post-exec funcs"""
         d = dict(called=False)
         def set_called():
             d['called'] = True
         
         ip.register_post_execute(set_called)
         ip.run_cell("1", silent=True)
         self.assertFalse(d['called'])
         # double-check that non-silent exec did what we expected
         # silent to avoid
         ip.run_cell("1")
         self.assertTrue(d['called'])
         # remove post-exec
         ip._post_execute.pop(set_called)
     
     def test_silent_noadvance(self):
         """run_cell(silent=True) doesn't advance execution_count"""
         ec = ip.execution_count
         # silent should force store_history=False
         ip.run_cell("1", store_history=True, silent=True)
         
         self.assertEqual(ec, ip.execution_count)
         # double-check that non-silent exec did what we expected
         # silent to avoid
         ip.run_cell("1", store_history=True)
         self.assertEqual(ec+1, ip.execution_count)
     
     def test_silent_nodisplayhook(self):
         """run_cell(silent=True) doesn't trigger displayhook"""
         d = dict(called=False)
         
         trap = ip.display_trap
         save_hook = trap.hook
         
         def failing_hook(*args, **kwargs):
             d['called'] = True
         
         try:
             trap.hook = failing_hook
             ip.run_cell("1", silent=True)
             self.assertFalse(d['called'])
             # double-check that non-silent exec did what we expected
             # silent to avoid
             ip.run_cell("1")
             self.assertTrue(d['called'])
         finally:
             trap.hook = save_hook
 
     @skipif(sys.version_info[0] >= 3, "softspace removed in py3")
     def test_print_softspace(self):
         """Verify that softspace is handled correctly when executing multiple
         statements.
 
         In [1]: print 1; print 2
         1
         2
 
         In [2]: print 1,; print 2
         1 2
         """
         
     def test_ofind_line_magic(self):
         from IPython.core.magic import register_line_magic
         
         @register_line_magic
         def lmagic(line):
             "A line magic"
 
         # Get info on line magic
         lfind = ip._ofind('lmagic')
         info = dict(found=True, isalias=False, ismagic=True,
                     namespace = 'IPython internal', obj= lmagic.__wrapped__,
                     parent = None)
         nt.assert_equal(lfind, info)
         
     def test_ofind_cell_magic(self):
         from IPython.core.magic import register_cell_magic
         
         @register_cell_magic
         def cmagic(line, cell):
             "A cell magic"
 
         # Get info on cell magic
         find = ip._ofind('cmagic')
         info = dict(found=True, isalias=False, ismagic=True,
                     namespace = 'IPython internal', obj= cmagic.__wrapped__,
                     parent = None)
         nt.assert_equal(find, info)
     
     def test_custom_exception(self):
         called = []
         def my_handler(shell, etype, value, tb, tb_offset=None):
             called.append(etype)
             shell.showtraceback((etype, value, tb), tb_offset=tb_offset)
         
         ip.set_custom_exc((ValueError,), my_handler)
         try:
             ip.run_cell("raise ValueError('test')")
             # Check that this was called, and only once.
             self.assertEqual(called, [ValueError])
         finally:
             # Reset the custom exception hook
             ip.set_custom_exc((), None)
     
     @skipif(sys.version_info[0] >= 3, "no differences with __future__ in py3")
     def test_future_environment(self):
         "Can we run code with & without the shell's __future__ imports?"
         ip.run_cell("from __future__ import division")
         ip.run_cell("a = 1/2", shell_futures=True)
         self.assertEqual(ip.user_ns['a'], 0.5)
         ip.run_cell("b = 1/2", shell_futures=False)
         self.assertEqual(ip.user_ns['b'], 0)
         
         ip.compile.reset_compiler_flags()
         # This shouldn't leak to the shell's compiler
         ip.run_cell("from __future__ import division \nc=1/2", shell_futures=False)
         self.assertEqual(ip.user_ns['c'], 0.5)
         ip.run_cell("d = 1/2", shell_futures=True)
         self.assertEqual(ip.user_ns['d'], 0)
 
 
 class TestSafeExecfileNonAsciiPath(unittest.TestCase):
 
     @onlyif_unicode_paths
     def setUp(self):
         self.BASETESTDIR = tempfile.mkdtemp()
         self.TESTDIR = join(self.BASETESTDIR, u"åäö")
         os.mkdir(self.TESTDIR)
         with open(join(self.TESTDIR, u"åäötestscript.py"), "w") as sfile:
             sfile.write("pass\n")
         self.oldpath = os.getcwdu()
         os.chdir(self.TESTDIR)
         self.fname = u"åäötestscript.py"
 
     def tearDown(self):
         os.chdir(self.oldpath)
         shutil.rmtree(self.BASETESTDIR)
 
     @onlyif_unicode_paths
     def test_1(self):
         """Test safe_execfile with non-ascii path
         """
         ip.safe_execfile(self.fname, {}, raise_exceptions=True)
 
 class ExitCodeChecks(tt.TempFileMixin):
     def test_exit_code_ok(self):
         self.system('exit 0')
         self.assertEqual(ip.user_ns['_exit_code'], 0)
 
     def test_exit_code_error(self):
         self.system('exit 1')
         self.assertEqual(ip.user_ns['_exit_code'], 1)
 
     @skipif(not hasattr(signal, 'SIGALRM'))
     def test_exit_code_signal(self):
         self.mktmp("import signal, time\n"
                    "signal.setitimer(signal.ITIMER_REAL, 0.1)\n"
                    "time.sleep(1)\n")
         self.system("%s %s" % (sys.executable, self.fname))
         self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGALRM)
 
 class TestSystemRaw(unittest.TestCase, ExitCodeChecks):
     system = ip.system_raw
 
     @onlyif_unicode_paths
     def test_1(self):
         """Test system_raw with non-ascii cmd
         """
         cmd = u'''python -c "'åäö'"   '''
         ip.system_raw(cmd)
 
 # TODO: Exit codes are currently ignored on Windows.
 class TestSystemPipedExitCode(unittest.TestCase, ExitCodeChecks):
     system = ip.system_piped
 
     @skip_win32
     def test_exit_code_ok(self):
         ExitCodeChecks.test_exit_code_ok(self)
 
     @skip_win32
     def test_exit_code_error(self):
         ExitCodeChecks.test_exit_code_error(self)
 
     @skip_win32
     def test_exit_code_signal(self):
         ExitCodeChecks.test_exit_code_signal(self)
 
 class TestModules(unittest.TestCase, tt.TempFileMixin):
     def test_extraneous_loads(self):
         """Test we're not loading modules on startup that we shouldn't.
         """
         self.mktmp("import sys\n"
                    "print('numpy' in sys.modules)\n"
                    "print('IPython.parallel' in sys.modules)\n"
                    "print('IPython.kernel.zmq' in sys.modules)\n"
                    )
         out = "False\nFalse\nFalse\n"
         tt.ipexec_validate(self.fname, out)
 
 class Negator(ast.NodeTransformer):
     """Negates all number literals in an AST."""
     def visit_Num(self, node):
         node.n = -node.n
         return node
 
 class TestAstTransform(unittest.TestCase):
     def setUp(self):
         self.negator = Negator()
         ip.ast_transformers.append(self.negator)
     
     def tearDown(self):
         ip.ast_transformers.remove(self.negator)
     
     def test_run_cell(self):
         with tt.AssertPrints('-34'):
             ip.run_cell('print (12 + 22)')
         
         # A named reference to a number shouldn't be transformed.
         ip.user_ns['n'] = 55
         with tt.AssertNotPrints('-55'):
             ip.run_cell('print (n)')
     
     def test_timeit(self):
         called = set()
         def f(x):
             called.add(x)
         ip.push({'f':f})
         
         with tt.AssertPrints("best of "):
             ip.run_line_magic("timeit", "-n1 f(1)")
         self.assertEqual(called, set([-1]))
         called.clear()
         
         with tt.AssertPrints("best of "):
             ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
         self.assertEqual(called, set([-2, -3]))
     
     def test_time(self):
         called = []
         def f(x):
             called.append(x)
         ip.push({'f':f})
         
         # Test with an expression
         with tt.AssertPrints("Wall time: "):
             ip.run_line_magic("time", "f(5+9)")
         self.assertEqual(called, [-14])
         called[:] = []
         
         # Test with a statement (different code path)
         with tt.AssertPrints("Wall time: "):
             ip.run_line_magic("time", "a = f(-3 + -2)")
         self.assertEqual(called, [5])
     
     def test_macro(self):
         ip.push({'a':10})
         # The AST transformation makes this do a+=-1
         ip.define_macro("amacro", "a+=1\nprint(a)")
         
         with tt.AssertPrints("9"):
             ip.run_cell("amacro")
         with tt.AssertPrints("8"):
             ip.run_cell("amacro")
 
 class IntegerWrapper(ast.NodeTransformer):
     """Wraps all integers in a call to Integer()"""
     def visit_Num(self, node):
         if isinstance(node.n, int):
             return ast.Call(func=ast.Name(id='Integer', ctx=ast.Load()),
                             args=[node], keywords=[])
         return node
 
 class TestAstTransform2(unittest.TestCase):
     def setUp(self):
         self.intwrapper = IntegerWrapper()
         ip.ast_transformers.append(self.intwrapper)
         
         self.calls = []
         def Integer(*args):
             self.calls.append(args)
             return args
         ip.push({"Integer": Integer})
     
     def tearDown(self):
         ip.ast_transformers.remove(self.intwrapper)
         del ip.user_ns['Integer']
     
     def test_run_cell(self):
         ip.run_cell("n = 2")
         self.assertEqual(self.calls, [(2,)])
         
         # This shouldn't throw an error
         ip.run_cell("o = 2.0")
         self.assertEqual(ip.user_ns['o'], 2.0)
     
     def test_timeit(self):
         called = set()
         def f(x):
             called.add(x)
         ip.push({'f':f})
         
         with tt.AssertPrints("best of "):
             ip.run_line_magic("timeit", "-n1 f(1)")
         self.assertEqual(called, set([(1,)]))
         called.clear()
         
         with tt.AssertPrints("best of "):
             ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
         self.assertEqual(called, set([(2,), (3,)]))
 
 class ErrorTransformer(ast.NodeTransformer):
     """Throws an error when it sees a number."""
     def visit_Num(self):
         raise ValueError("test")
 
 class TestAstTransformError(unittest.TestCase):
     def test_unregistering(self):
         err_transformer = ErrorTransformer()
         ip.ast_transformers.append(err_transformer)
         
         with tt.AssertPrints("unregister", channel='stderr'):
             ip.run_cell("1 + 2")
         
         # This should have been removed.
         nt.assert_not_in(err_transformer, ip.ast_transformers)
 
 def test__IPYTHON__():
     # This shouldn't raise a NameError, that's all
     __IPYTHON__
 
 
 class DummyRepr(object):
     def __repr__(self):
         return "DummyRepr"
     
     def _repr_html_(self):
         return "<b>dummy</b>"
     
     def _repr_javascript_(self):
         return "console.log('hi');", {'key': 'value'}
     
 
 def test_user_variables():
     # enable all formatters
     ip.display_formatter.active_types = ip.display_formatter.format_types
     
     ip.user_ns['dummy'] = d = DummyRepr()
     keys = set(['dummy', 'doesnotexist'])
     r = ip.user_variables(keys)
 
     nt.assert_equal(keys, set(r.keys()))
     dummy = r['dummy']
     nt.assert_equal(set(['status', 'data', 'metadata']), set(dummy.keys()))
     nt.assert_equal(dummy['status'], 'ok')
     data = dummy['data']
     metadata = dummy['metadata']
     nt.assert_equal(data.get('text/html'), d._repr_html_())
     js, jsmd = d._repr_javascript_()
     nt.assert_equal(data.get('application/javascript'), js)
     nt.assert_equal(metadata.get('application/javascript'), jsmd)
     
     dne = r['doesnotexist']
     nt.assert_equal(dne['status'], 'error')
     nt.assert_equal(dne['ename'], 'KeyError')
     
     # back to text only
     ip.display_formatter.active_types = ['text/plain']
     
 def test_user_expression():
     # enable all formatters
     ip.display_formatter.active_types = ip.display_formatter.format_types
     query = {
         'a' : '1 + 2',
         'b' : '1/0',
     }
     r = ip.user_expressions(query)
     import pprint
     pprint.pprint(r)
     nt.assert_equal(r.keys(), query.keys())
     a = r['a']
     nt.assert_equal(set(['status', 'data', 'metadata']), set(a.keys()))
     nt.assert_equal(a['status'], 'ok')
     data = a['data']
     metadata = a['metadata']
     nt.assert_equal(data.get('text/plain'), '3')
     
     b = r['b']
     nt.assert_equal(b['status'], 'error')
     nt.assert_equal(b['ename'], 'ZeroDivisionError')
     
     # back to text only
     ip.display_formatter.active_types = ['text/plain']
     
 
 
 
 
 

IPython/core/tests/test_magic.py

@@ -1,940 +1,944
 # -*- coding: utf-8 -*-
 """Tests for various magic functions.
 
 Needs to be run by nose (to make ipython session available).
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import io
 import os
 import sys
-from io import StringIO
 from unittest import TestCase
 
 try:
     from importlib import invalidate_caches   # Required from Python 3.3
 except ImportError:
     def invalidate_caches():
         pass
 
 import nose.tools as nt
 
 from IPython.core import magic
 from IPython.core.magic import (Magics, magics_class, line_magic,
                                 cell_magic, line_cell_magic,
                                 register_line_magic, register_cell_magic,
                                 register_line_cell_magic)
 from IPython.core.magics import execution, script, code
 from IPython.nbformat.v3.tests.nbexamples import nb0
 from IPython.nbformat import current
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 from IPython.utils.io import capture_output
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.utils.process import find_cmd
 
+if py3compat.PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 
 @magic.magics_class
 class DummyMagics(magic.Magics): pass
 
 def test_extract_code_ranges():
     instr = "1 3 5-6 7-9 10:15 17: :10 10- -13 :"
     expected = [(0, 1), 
                 (2, 3),
                 (4, 6),
                 (6, 9),
                 (9, 14),
                 (16, None),
                 (None, 9),
                 (9, None),
                 (None, 13),
                 (None, None)]
     actual = list(code.extract_code_ranges(instr))
     nt.assert_equal(actual, expected)
 
 def test_extract_symbols():
     source = """import foo\na = 10\ndef b():\n    return 42\n\n\nclass A: pass\n\n\n"""
     symbols_args = ["a", "b", "A", "A,b", "A,a", "z"]
     expected = [([], ['a']),
                 (["def b():\n    return 42\n"], []),
                 (["class A: pass\n"], []),
                 (["class A: pass\n", "def b():\n    return 42\n"], []),
                 (["class A: pass\n"], ['a']),
                 ([], ['z'])]
     for symbols, exp in zip(symbols_args, expected):
         nt.assert_equal(code.extract_symbols(source, symbols), exp)
 
 
 def test_extract_symbols_raises_exception_with_non_python_code():
     source = ("=begin A Ruby program :)=end\n"
               "def hello\n"
               "puts 'Hello world'\n"
               "end")
     with nt.assert_raises(SyntaxError):
         code.extract_symbols(source, "hello")
 
 def test_config():
     """ test that config magic does not raise
     can happen if Configurable init is moved too early into
     Magics.__init__ as then a Config object will be registerd as a
     magic.
     """
     ## should not raise.
     _ip.magic('config')
 
 def test_rehashx():
     # clear up everything
     _ip = get_ipython()
     _ip.alias_manager.clear_aliases()
     del _ip.db['syscmdlist']
     
     _ip.magic('rehashx')
     # Practically ALL ipython development systems will have more than 10 aliases
 
     nt.assert_true(len(_ip.alias_manager.aliases) > 10)
     for name, cmd in _ip.alias_manager.aliases:
         # we must strip dots from alias names
         nt.assert_not_in('.', name)
 
     # rehashx must fill up syscmdlist
     scoms = _ip.db['syscmdlist']
     nt.assert_true(len(scoms) > 10)
 
 
 def test_magic_parse_options():
     """Test that we don't mangle paths when parsing magic options."""
     ip = get_ipython()
     path = 'c:\\x'
     m = DummyMagics(ip)
     opts = m.parse_options('-f %s' % path,'f:')[0]
     # argv splitting is os-dependent
     if os.name == 'posix':
         expected = 'c:x'
     else:
         expected = path
     nt.assert_equal(opts['f'], expected)
 
 def test_magic_parse_long_options():
     """Magic.parse_options can handle --foo=bar long options"""
     ip = get_ipython()
     m = DummyMagics(ip)
     opts, _ = m.parse_options('--foo --bar=bubble', 'a', 'foo', 'bar=')
     nt.assert_in('foo', opts)
     nt.assert_in('bar', opts)
     nt.assert_equal(opts['bar'], "bubble")
 
 
 @dec.skip_without('sqlite3')
 def doctest_hist_f():
     """Test %hist -f with temporary filename.
 
     In [9]: import tempfile
 
     In [10]: tfile = tempfile.mktemp('.py','tmp-ipython-')
 
     In [11]: %hist -nl -f $tfile 3
 
     In [13]: import os; os.unlink(tfile)
     """
 
 
 @dec.skip_without('sqlite3')
 def doctest_hist_r():
     """Test %hist -r
 
     XXX - This test is not recording the output correctly.  For some reason, in
     testing mode the raw history isn't getting populated.  No idea why.
     Disabling the output checking for now, though at least we do run it.
 
     In [1]: 'hist' in _ip.lsmagic()
     Out[1]: True
 
     In [2]: x=1
 
     In [3]: %hist -rl 2
     x=1 # random
     %hist -r 2
     """
 
 
 @dec.skip_without('sqlite3')
 def doctest_hist_op():
     """Test %hist -op
 
     In [1]: class b(float):
        ...:     pass
        ...: 
 
     In [2]: class s(object):
        ...:     def __str__(self):
        ...:         return 's'
        ...: 
 
     In [3]: 
 
     In [4]: class r(b):
        ...:     def __repr__(self):
        ...:         return 'r'
        ...: 
 
     In [5]: class sr(s,r): pass
        ...: 
 
     In [6]: 
 
     In [7]: bb=b()
 
     In [8]: ss=s()
 
     In [9]: rr=r()
 
     In [10]: ssrr=sr()
 
     In [11]: 4.5
     Out[11]: 4.5
 
     In [12]: str(ss)
     Out[12]: 's'
 
     In [13]: 
 
     In [14]: %hist -op
     >>> class b:
     ...     pass
     ... 
     >>> class s(b):
     ...     def __str__(self):
     ...         return 's'
     ... 
     >>> 
     >>> class r(b):
     ...     def __repr__(self):
     ...         return 'r'
     ... 
     >>> class sr(s,r): pass
     >>> 
     >>> bb=b()
     >>> ss=s()
     >>> rr=r()
     >>> ssrr=sr()
     >>> 4.5
     4.5
     >>> str(ss)
     's'
     >>> 
     """
 
 
 @dec.skip_without('sqlite3')
 def test_macro():
     ip = get_ipython()
     ip.history_manager.reset()   # Clear any existing history.
     cmds = ["a=1", "def b():\n  return a**2", "print(a,b())"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
     ip.magic("macro test 1-3")
     nt.assert_equal(ip.user_ns["test"].value, "\n".join(cmds)+"\n")
     
     # List macros
     nt.assert_in("test", ip.magic("macro"))
 
 
 @dec.skip_without('sqlite3')
 def test_macro_run():
     """Test that we can run a multi-line macro successfully."""
     ip = get_ipython()
     ip.history_manager.reset()
     cmds = ["a=10", "a+=1", py3compat.doctest_refactor_print("print a"),
                                                             "%macro test 2-3"]
     for cmd in cmds:
         ip.run_cell(cmd, store_history=True)
     nt.assert_equal(ip.user_ns["test"].value,
                             py3compat.doctest_refactor_print("a+=1\nprint a\n"))
     with tt.AssertPrints("12"):
         ip.run_cell("test")
     with tt.AssertPrints("13"):
         ip.run_cell("test")
 
 
 def test_magic_magic():
     """Test %magic"""
     ip = get_ipython()
     with capture_output() as captured:
         ip.magic("magic")
     
     stdout = captured.stdout
     nt.assert_in('%magic', stdout)
     nt.assert_in('IPython', stdout)
     nt.assert_in('Available', stdout)
 
 
 @dec.skipif_not_numpy
 def test_numpy_reset_array_undec():
     "Test '%reset array' functionality"
     _ip.ex('import numpy as np')
     _ip.ex('a = np.empty(2)')
     nt.assert_in('a', _ip.user_ns)
     _ip.magic('reset -f array')
     nt.assert_not_in('a', _ip.user_ns)
 
 def test_reset_out():
     "Test '%reset out' magic"
     _ip.run_cell("parrot = 'dead'", store_history=True)
     # test '%reset -f out', make an Out prompt
     _ip.run_cell("parrot", store_history=True)
     nt.assert_true('dead' in [_ip.user_ns[x] for x in '_','__','___'])
     _ip.magic('reset -f out')
     nt.assert_false('dead' in [_ip.user_ns[x] for x in '_','__','___'])
     nt.assert_equal(len(_ip.user_ns['Out']), 0)
 
 def test_reset_in():
     "Test '%reset in' magic"
     # test '%reset -f in'
     _ip.run_cell("parrot", store_history=True)
     nt.assert_true('parrot' in [_ip.user_ns[x] for x in '_i','_ii','_iii'])
     _ip.magic('%reset -f in')
     nt.assert_false('parrot' in [_ip.user_ns[x] for x in '_i','_ii','_iii'])
     nt.assert_equal(len(set(_ip.user_ns['In'])), 1)
 
 def test_reset_dhist():
     "Test '%reset dhist' magic"
     _ip.run_cell("tmp = [d for d in _dh]") # copy before clearing
     _ip.magic('cd ' + os.path.dirname(nt.__file__))
     _ip.magic('cd -')
     nt.assert_true(len(_ip.user_ns['_dh']) > 0)
     _ip.magic('reset -f dhist')
     nt.assert_equal(len(_ip.user_ns['_dh']), 0)
     _ip.run_cell("_dh = [d for d in tmp]") #restore
 
 def test_reset_in_length():
     "Test that '%reset in' preserves In[] length"
     _ip.run_cell("print 'foo'")
     _ip.run_cell("reset -f in")
     nt.assert_equal(len(_ip.user_ns['In']), _ip.displayhook.prompt_count+1)
 
 def test_tb_syntaxerror():
     """test %tb after a SyntaxError"""
     ip = get_ipython()
     ip.run_cell("for")
     
     # trap and validate stdout
     save_stdout = sys.stdout
     try:
         sys.stdout = StringIO()
         ip.run_cell("%tb")
         out = sys.stdout.getvalue()
     finally:
         sys.stdout = save_stdout
     # trim output, and only check the last line
     last_line = out.rstrip().splitlines()[-1].strip()
     nt.assert_equal(last_line, "SyntaxError: invalid syntax")
 
 
 def test_time():
     ip = get_ipython()
     
     with tt.AssertPrints("Wall time: "):
         ip.run_cell("%time None")
     
     ip.run_cell("def f(kmjy):\n"
                 "    %time print (2*kmjy)")
     
     with tt.AssertPrints("Wall time: "):
         with tt.AssertPrints("hihi", suppress=False):
             ip.run_cell("f('hi')")
 
 
 @dec.skip_win32
 def test_time2():
     ip = get_ipython()
     
     with tt.AssertPrints("CPU times: user "):
         ip.run_cell("%time None")
 
 def test_time3():
     """Erroneous magic function calls, issue gh-3334"""
     ip = get_ipython()
     ip.user_ns.pop('run', None)
     
     with tt.AssertNotPrints("not found", channel='stderr'):
         ip.run_cell("%%time\n"
                     "run = 0\n"
                     "run += 1")
 
 def test_doctest_mode():
     "Toggle doctest_mode twice, it should be a no-op and run without error"
     _ip.magic('doctest_mode')
     _ip.magic('doctest_mode')
 
 
 def test_parse_options():
     """Tests for basic options parsing in magics."""
     # These are only the most minimal of tests, more should be added later.  At
     # the very least we check that basic text/unicode calls work OK.
     m = DummyMagics(_ip)
     nt.assert_equal(m.parse_options('foo', '')[1], 'foo')
     nt.assert_equal(m.parse_options(u'foo', '')[1], u'foo')
 
     
 def test_dirops():
     """Test various directory handling operations."""
     # curpath = lambda :os.path.splitdrive(os.getcwdu())[1].replace('\\','/')
     curpath = os.getcwdu
     startdir = os.getcwdu()
     ipdir = os.path.realpath(_ip.ipython_dir)
     try:
         _ip.magic('cd "%s"' % ipdir)
         nt.assert_equal(curpath(), ipdir)
         _ip.magic('cd -')
         nt.assert_equal(curpath(), startdir)
         _ip.magic('pushd "%s"' % ipdir)
         nt.assert_equal(curpath(), ipdir)
         _ip.magic('popd')
         nt.assert_equal(curpath(), startdir)
     finally:
         os.chdir(startdir)
 
 
 def test_xmode():
     # Calling xmode three times should be a no-op
     xmode = _ip.InteractiveTB.mode
     for i in range(3):
         _ip.magic("xmode")
     nt.assert_equal(_ip.InteractiveTB.mode, xmode)
     
 def test_reset_hard():
     monitor = []
     class A(object):
         def __del__(self):
             monitor.append(1)
         def __repr__(self):
             return "<A instance>"
             
     _ip.user_ns["a"] = A()
     _ip.run_cell("a")
     
     nt.assert_equal(monitor, [])
     _ip.magic("reset -f")
     nt.assert_equal(monitor, [1])
     
 class TestXdel(tt.TempFileMixin):
     def test_xdel(self):
         """Test that references from %run are cleared by xdel."""
         src = ("class A(object):\n"
                "    monitor = []\n"
                "    def __del__(self):\n"
                "        self.monitor.append(1)\n"
                "a = A()\n")
         self.mktmp(src)
         # %run creates some hidden references...
         _ip.magic("run %s" % self.fname)
         # ... as does the displayhook.
         _ip.run_cell("a")
         
         monitor = _ip.user_ns["A"].monitor
         nt.assert_equal(monitor, [])
         
         _ip.magic("xdel a")
         
         # Check that a's __del__ method has been called.
         nt.assert_equal(monitor, [1])
 
 def doctest_who():
     """doctest for %who
     
     In [1]: %reset -f
     
     In [2]: alpha = 123
     
     In [3]: beta = 'beta'
     
     In [4]: %who int
     alpha
     
     In [5]: %who str
     beta
     
     In [6]: %whos
     Variable   Type    Data/Info
     ----------------------------
     alpha      int     123
     beta       str     beta
     
     In [7]: %who_ls
     Out[7]: ['alpha', 'beta']
     """
 
 def test_whos():
     """Check that whos is protected against objects where repr() fails."""
     class A(object):
         def __repr__(self):
             raise Exception()
     _ip.user_ns['a'] = A()
     _ip.magic("whos")
 
 @py3compat.u_format
 def doctest_precision():
     """doctest for %precision
     
     In [1]: f = get_ipython().display_formatter.formatters['text/plain']
     
     In [2]: %precision 5
     Out[2]: {u}'%.5f'
     
     In [3]: f.float_format
     Out[3]: {u}'%.5f'
     
     In [4]: %precision %e
     Out[4]: {u}'%e'
     
     In [5]: f(3.1415927)
     Out[5]: {u}'3.141593e+00'
     """
 
 def test_psearch():
     with tt.AssertPrints("dict.fromkeys"):
         _ip.run_cell("dict.fr*?")
 
 def test_timeit_shlex():
     """test shlex issues with timeit (#1109)"""
     _ip.ex("def f(*a,**kw): pass")
     _ip.magic('timeit -n1 "this is a bug".count(" ")')
     _ip.magic('timeit -r1 -n1 f(" ", 1)')
     _ip.magic('timeit -r1 -n1 f(" ", 1, " ", 2, " ")')
     _ip.magic('timeit -r1 -n1 ("a " + "b")')
     _ip.magic('timeit -r1 -n1 f("a " + "b")')
     _ip.magic('timeit -r1 -n1 f("a " + "b ")')
 
 
 def test_timeit_arguments():
     "Test valid timeit arguments, should not cause SyntaxError (GH #1269)"
     _ip.magic("timeit ('#')")
 
 
 def test_timeit_special_syntax():
     "Test %%timeit with IPython special syntax"
     @register_line_magic
     def lmagic(line):
         ip = get_ipython()
         ip.user_ns['lmagic_out'] = line
 
     # line mode test
     _ip.run_line_magic('timeit', '-n1 -r1 %lmagic my line')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line')
     # cell mode test
     _ip.run_cell_magic('timeit', '-n1 -r1', '%lmagic my line2')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line2')
 
 def test_timeit_return():
     """
     test wether timeit -o return object
     """
 
     res = _ip.run_line_magic('timeit','-n10 -r10 -o 1')
     assert(res is not None)
 
 def test_timeit_quiet():
     """
     test quiet option of timeit magic
     """
     with tt.AssertNotPrints("loops"):
         _ip.run_cell("%timeit -n1 -r1 -q 1")
 
 @dec.skipif(execution.profile is None)
 def test_prun_special_syntax():
     "Test %%prun with IPython special syntax"
     @register_line_magic
     def lmagic(line):
         ip = get_ipython()
         ip.user_ns['lmagic_out'] = line
 
     # line mode test
     _ip.run_line_magic('prun', '-q %lmagic my line')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line')
     # cell mode test
     _ip.run_cell_magic('prun', '-q', '%lmagic my line2')
     nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line2')
 
 @dec.skipif(execution.profile is None)
 def test_prun_quotes():
     "Test that prun does not clobber string escapes (GH #1302)"
     _ip.magic(r"prun -q x = '\t'")
     nt.assert_equal(_ip.user_ns['x'], '\t')
 
 def test_extension():
     tmpdir = TemporaryDirectory()
     orig_ipython_dir = _ip.ipython_dir
     try:
         _ip.ipython_dir = tmpdir.name
         nt.assert_raises(ImportError, _ip.magic, "load_ext daft_extension")
         url = os.path.join(os.path.dirname(__file__), "daft_extension.py")
         _ip.magic("install_ext %s" % url)
         _ip.user_ns.pop('arq', None)
         invalidate_caches()   # Clear import caches
         _ip.magic("load_ext daft_extension")
         nt.assert_equal(_ip.user_ns['arq'], 185)
         _ip.magic("unload_ext daft_extension")
         assert 'arq' not in _ip.user_ns
     finally:
         _ip.ipython_dir = orig_ipython_dir
         tmpdir.cleanup()
         
 def test_notebook_export_json():
     with TemporaryDirectory() as td:
         outfile = os.path.join(td, "nb.ipynb")
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -e %s" % outfile)
 
 def test_notebook_export_py():
     with TemporaryDirectory() as td:
         outfile = os.path.join(td, "nb.py")
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -e %s" % outfile)
 
 def test_notebook_reformat_py():
     with TemporaryDirectory() as td:
         infile = os.path.join(td, "nb.ipynb")
         with io.open(infile, 'w', encoding='utf-8') as f:
             current.write(nb0, f, 'json')
             
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -f py %s" % infile)
 
 def test_notebook_reformat_json():
     with TemporaryDirectory() as td:
         infile = os.path.join(td, "nb.py")
         with io.open(infile, 'w', encoding='utf-8') as f:
             current.write(nb0, f, 'py')
             
         _ip.ex(py3compat.u_format(u"u = {u}'héllo'"))
         _ip.magic("notebook -f ipynb %s" % infile)
         _ip.magic("notebook -f json %s" % infile)
 
 def test_env():
     env = _ip.magic("env")
     assert isinstance(env, dict), type(env)
 
 
 class CellMagicTestCase(TestCase):
 
     def check_ident(self, magic):
         # Manually called, we get the result
         out = _ip.run_cell_magic(magic, 'a', 'b')
         nt.assert_equal(out, ('a','b'))
         # Via run_cell, it goes into the user's namespace via displayhook
         _ip.run_cell('%%' + magic +' c\nd')
         nt.assert_equal(_ip.user_ns['_'], ('c','d'))
 
     def test_cell_magic_func_deco(self):
         "Cell magic using simple decorator"
         @register_cell_magic
         def cellm(line, cell):
             return line, cell
 
         self.check_ident('cellm')
 
     def test_cell_magic_reg(self):
         "Cell magic manually registered"
         def cellm(line, cell):
             return line, cell
 
         _ip.register_magic_function(cellm, 'cell', 'cellm2')
         self.check_ident('cellm2')
 
     def test_cell_magic_class(self):
         "Cell magics declared via a class"
         @magics_class
         class MyMagics(Magics):
 
             @cell_magic
             def cellm3(self, line, cell):
                 return line, cell
 
         _ip.register_magics(MyMagics)
         self.check_ident('cellm3')
 
     def test_cell_magic_class2(self):
         "Cell magics declared via a class, #2"
         @magics_class
         class MyMagics2(Magics):
 
             @cell_magic('cellm4')
             def cellm33(self, line, cell):
                 return line, cell
             
         _ip.register_magics(MyMagics2)
         self.check_ident('cellm4')
         # Check that nothing is registered as 'cellm33'
         c33 = _ip.find_cell_magic('cellm33')
         nt.assert_equal(c33, None)
 
 def test_file():
     """Basic %%file"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.run_cell_magic("file", fname, u'\n'.join([
             'line1',
             'line2',
         ]))
         with open(fname) as f:
             s = f.read()
         nt.assert_in('line1\n', s)
         nt.assert_in('line2', s)
 
 def test_file_var_expand():
     """%%file $filename"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.user_ns['filename'] = fname
         ip.run_cell_magic("file", '$filename', u'\n'.join([
             'line1',
             'line2',
         ]))
         with open(fname) as f:
             s = f.read()
         nt.assert_in('line1\n', s)
         nt.assert_in('line2', s)
 
 def test_file_unicode():
     """%%file with unicode cell"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.run_cell_magic("file", fname, u'\n'.join([
             u'liné1',
             u'liné2',
         ]))
         with io.open(fname, encoding='utf-8') as f:
             s = f.read()
         nt.assert_in(u'liné1\n', s)
         nt.assert_in(u'liné2', s)
 
 def test_file_amend():
     """%%file -a amends files"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file2')
         ip.run_cell_magic("file", fname, u'\n'.join([
             'line1',
             'line2',
         ]))
         ip.run_cell_magic("file", "-a %s" % fname, u'\n'.join([
             'line3',
             'line4',
         ]))
         with open(fname) as f:
             s = f.read()
         nt.assert_in('line1\n', s)
         nt.assert_in('line3\n', s)
         
     
 def test_script_config():
     ip = get_ipython()
     ip.config.ScriptMagics.script_magics = ['whoda']
     sm = script.ScriptMagics(shell=ip)
     nt.assert_in('whoda', sm.magics['cell'])
 
 @dec.skip_win32
 def test_script_out():
     ip = get_ipython()
     ip.run_cell_magic("script", "--out output sh", "echo 'hi'")
     nt.assert_equal(ip.user_ns['output'], 'hi\n')
 
 @dec.skip_win32
 def test_script_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--err error sh", "echo 'hello' >&2")
     nt.assert_equal(ip.user_ns['error'], 'hello\n')
 
 @dec.skip_win32
 def test_script_out_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--out output --err error sh", "echo 'hi'\necho 'hello' >&2")
     nt.assert_equal(ip.user_ns['output'], 'hi\n')
     nt.assert_equal(ip.user_ns['error'], 'hello\n')
 
 @dec.skip_win32
 def test_script_bg_out():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --out output sh", "echo 'hi'")
     nt.assert_equal(ip.user_ns['output'].read(), b'hi\n')
 
 @dec.skip_win32
 def test_script_bg_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --err error sh", "echo 'hello' >&2")
     nt.assert_equal(ip.user_ns['error'].read(), b'hello\n')
 
 @dec.skip_win32
 def test_script_bg_out_err():
     ip = get_ipython()
     ip.run_cell_magic("script", "--bg --out output --err error sh", "echo 'hi'\necho 'hello' >&2")
     nt.assert_equal(ip.user_ns['output'].read(), b'hi\n')
     nt.assert_equal(ip.user_ns['error'].read(), b'hello\n')
 
 def test_script_defaults():
     ip = get_ipython()
     for cmd in ['sh', 'bash', 'perl', 'ruby']:
         try:
             find_cmd(cmd)
         except Exception:
             pass
         else:
             nt.assert_in(cmd, ip.magics_manager.magics['cell'])
 
 
 @magics_class
 class FooFoo(Magics):
     """class with both %foo and %%foo magics"""
     @line_magic('foo')
     def line_foo(self, line):
         "I am line foo"
         pass
 
     @cell_magic("foo")
     def cell_foo(self, line, cell):
         "I am cell foo, not line foo"
         pass
 
 def test_line_cell_info():
     """%%foo and %foo magics are distinguishable to inspect"""
     ip = get_ipython()
     ip.magics_manager.register(FooFoo)
     oinfo = ip.object_inspect('foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     
     oinfo = ip.object_inspect('%%foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     nt.assert_equal(oinfo['docstring'], FooFoo.cell_foo.__doc__)
 
     oinfo = ip.object_inspect('%foo')
     nt.assert_true(oinfo['found'])
     nt.assert_true(oinfo['ismagic'])
     nt.assert_equal(oinfo['docstring'], FooFoo.line_foo.__doc__)
 
 def test_multiple_magics():
     ip = get_ipython()
     foo1 = FooFoo(ip)
     foo2 = FooFoo(ip)
     mm = ip.magics_manager
     mm.register(foo1)
     nt.assert_true(mm.magics['line']['foo'].im_self is foo1)
     mm.register(foo2)
     nt.assert_true(mm.magics['line']['foo'].im_self is foo2)
 
 def test_alias_magic():
     """Test %alias_magic."""
     ip = get_ipython()
     mm = ip.magics_manager
 
     # Basic operation: both cell and line magics are created, if possible.
     ip.run_line_magic('alias_magic', 'timeit_alias timeit')
     nt.assert_in('timeit_alias', mm.magics['line'])
     nt.assert_in('timeit_alias', mm.magics['cell'])
 
     # --cell is specified, line magic not created.
     ip.run_line_magic('alias_magic', '--cell timeit_cell_alias timeit')
     nt.assert_not_in('timeit_cell_alias', mm.magics['line'])
     nt.assert_in('timeit_cell_alias', mm.magics['cell'])
 
     # Test that line alias is created successfully.
     ip.run_line_magic('alias_magic', '--line env_alias env')
     nt.assert_equal(ip.run_line_magic('env', ''),
                     ip.run_line_magic('env_alias', ''))
 
 def test_save():
     """Test %save."""
     ip = get_ipython()
     ip.history_manager.reset()   # Clear any existing history.
     cmds = [u"a=1", u"def b():\n  return a**2", u"print(a, b())"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
     with TemporaryDirectory() as tmpdir:
         file = os.path.join(tmpdir, "testsave.py")
         ip.run_line_magic("save", "%s 1-10" % file)
         with open(file) as f:
             content = f.read()
             nt.assert_equal(content.count(cmds[0]), 1)
             nt.assert_in('coding: utf-8', content)
         ip.run_line_magic("save", "-a %s 1-10" % file)
         with open(file) as f:
             content = f.read()
             nt.assert_equal(content.count(cmds[0]), 2)
             nt.assert_in('coding: utf-8', content)
 
 
 def test_store():
     """Test %store."""
     ip = get_ipython()
     ip.run_line_magic('load_ext', 'storemagic')
     
     # make sure the storage is empty
     ip.run_line_magic('store', '-z')
     ip.user_ns['var'] = 42
     ip.run_line_magic('store', 'var')
     ip.user_ns['var'] = 39
     ip.run_line_magic('store', '-r')
     nt.assert_equal(ip.user_ns['var'], 42)
 
     ip.run_line_magic('store', '-d var')
     ip.user_ns['var'] = 39
     ip.run_line_magic('store' , '-r')
     nt.assert_equal(ip.user_ns['var'], 39)
 
 
 def _run_edit_test(arg_s, exp_filename=None,
                         exp_lineno=-1,
                         exp_contents=None,
                         exp_is_temp=None):
     ip = get_ipython()
     M = code.CodeMagics(ip)
     last_call = ['','']
     opts,args = M.parse_options(arg_s,'prxn:')
     filename, lineno, is_temp = M._find_edit_target(ip, args, opts, last_call)
         
     if exp_filename is not None:
         nt.assert_equal(exp_filename, filename)
     if exp_contents is not None:
         with io.open(filename, 'r') as f:
             contents = f.read()
         nt.assert_equal(exp_contents, contents)
     if exp_lineno != -1:
         nt.assert_equal(exp_lineno, lineno)
     if exp_is_temp is not None:
         nt.assert_equal(exp_is_temp, is_temp)
 
 
 def test_edit_interactive():
     """%edit on interactively defined objects"""
     ip = get_ipython()
     n = ip.execution_count
     ip.run_cell(u"def foo(): return 1", store_history=True)
     
     try:
         _run_edit_test("foo")
     except code.InteractivelyDefined as e:
         nt.assert_equal(e.index, n)
     else:
         raise AssertionError("Should have raised InteractivelyDefined")
 
 
 def test_edit_cell():
     """%edit [cell id]"""
     ip = get_ipython()
     
     ip.run_cell(u"def foo(): return 1", store_history=True)
     
     # test
     _run_edit_test("1", exp_contents=ip.user_ns['In'][1], exp_is_temp=True)

IPython/extensions/tests/test_autoreload.py

@@ -1,317 +1,322
 """Tests for autoreload extension.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (c) 2012 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 os
 import sys
 import tempfile
 import shutil
 import random
 import time
-from io import StringIO
 
 import nose.tools as nt
 import IPython.testing.tools as tt
 
 from IPython.extensions.autoreload import AutoreloadMagics
 from IPython.core.hooks import TryNext
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Test fixture
 #-----------------------------------------------------------------------------
 
 noop = lambda *a, **kw: None
 
 class FakeShell(object):
 
     def __init__(self):
         self.ns = {}
         self.auto_magics = AutoreloadMagics(shell=self)
 
     register_magics = set_hook = noop
 
     def run_code(self, code):
         try:
             self.auto_magics.pre_run_code_hook(self)
         except TryNext:
             pass
         exec(code, self.ns)
 
     def push(self, items):
         self.ns.update(items)
 
     def magic_autoreload(self, parameter):
         self.auto_magics.autoreload(parameter)
 
     def magic_aimport(self, parameter, stream=None):
         self.auto_magics.aimport(parameter, stream=stream)
 
 
 class Fixture(object):
     """Fixture for creating test module files"""
 
     test_dir = None
     old_sys_path = None
     filename_chars = "abcdefghijklmopqrstuvwxyz0123456789"
 
     def setUp(self):
         self.test_dir = tempfile.mkdtemp()
         self.old_sys_path = list(sys.path)
         sys.path.insert(0, self.test_dir)
         self.shell = FakeShell()
 
     def tearDown(self):
         shutil.rmtree(self.test_dir)
         sys.path = self.old_sys_path
 
         self.test_dir = None
         self.old_sys_path = None
         self.shell = None
 
     def get_module(self):
         module_name = "tmpmod_" + "".join(random.sample(self.filename_chars,20))
         if module_name in sys.modules:
             del sys.modules[module_name]
         file_name = os.path.join(self.test_dir, module_name + ".py")
         return module_name, file_name
 
     def write_file(self, filename, content):
         """
         Write a file, and force a timestamp difference of at least one second
 
         Notes
         -----
         Python's .pyc files record the timestamp of their compilation
         with a time resolution of one second.
 
         Therefore, we need to force a timestamp difference between .py
         and .pyc, without having the .py file be timestamped in the
         future, and without changing the timestamp of the .pyc file
         (because that is stored in the file).  The only reliable way
         to achieve this seems to be to sleep.
         """
 
         # Sleep one second + eps
         time.sleep(1.05)
 
         # Write
         f = open(filename, 'w')
         try:
             f.write(content)
         finally:
             f.close()
 
     def new_module(self, code):
         mod_name, mod_fn = self.get_module()
         f = open(mod_fn, 'w')
         try:
             f.write(code)
         finally:
             f.close()
         return mod_name, mod_fn
 
 #-----------------------------------------------------------------------------
 # Test automatic reloading
 #-----------------------------------------------------------------------------
 
 class TestAutoreload(Fixture):
     def _check_smoketest(self, use_aimport=True):
         """
         Functional test for the automatic reloader using either
         '%autoreload 1' or '%autoreload 2'
         """
 
         mod_name, mod_fn = self.new_module("""
 x = 9
 
 z = 123  # this item will be deleted
 
 def foo(y):
     return y + 3
 
 class Baz(object):
     def __init__(self, x):
         self.x = x
     def bar(self, y):
         return self.x + y
     @property
     def quux(self):
         return 42
     def zzz(self):
         '''This method will be deleted below'''
         return 99
 
 class Bar:    # old-style class: weakref doesn't work for it on Python < 2.7
     def foo(self):
         return 1
 """)
 
         #
         # Import module, and mark for reloading
         #
         if use_aimport:
             self.shell.magic_autoreload("1")
             self.shell.magic_aimport(mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             nt.assert_true(("Modules to reload:\n%s" % mod_name) in
                            stream.getvalue())
 
             nt.assert_raises(
                 ImportError,
                 self.shell.magic_aimport, "tmpmod_as318989e89ds")
         else:
             self.shell.magic_autoreload("2")
             self.shell.run_code("import %s" % mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             nt.assert_true("Modules to reload:\nall-except-skipped" in
                            stream.getvalue())
         nt.assert_in(mod_name, self.shell.ns)
 
         mod = sys.modules[mod_name]
 
         #
         # Test module contents
         #
         old_foo = mod.foo
         old_obj = mod.Baz(9)
         old_obj2 = mod.Bar()
 
         def check_module_contents():
             nt.assert_equal(mod.x, 9)
             nt.assert_equal(mod.z, 123)
 
             nt.assert_equal(old_foo(0), 3)
             nt.assert_equal(mod.foo(0), 3)
 
             obj = mod.Baz(9)
             nt.assert_equal(old_obj.bar(1), 10)
             nt.assert_equal(obj.bar(1), 10)
             nt.assert_equal(obj.quux, 42)
             nt.assert_equal(obj.zzz(), 99)
 
             obj2 = mod.Bar()
             nt.assert_equal(old_obj2.foo(), 1)
             nt.assert_equal(obj2.foo(), 1)
 
         check_module_contents()
 
         #
         # Simulate a failed reload: no reload should occur and exactly
         # one error message should be printed
         #
         self.write_file(mod_fn, """
 a syntax error
 """)
 
         with tt.AssertPrints(('[autoreload of %s failed:' % mod_name), channel='stderr'):
             self.shell.run_code("pass") # trigger reload
         with tt.AssertNotPrints(('[autoreload of %s failed:' % mod_name), channel='stderr'):
             self.shell.run_code("pass") # trigger another reload
         check_module_contents()
 
         #
         # Rewrite module (this time reload should succeed)
         #
         self.write_file(mod_fn, """
 x = 10
 
 def foo(y):
     return y + 4
 
 class Baz(object):
     def __init__(self, x):
         self.x = x
     def bar(self, y):
         return self.x + y + 1
     @property
     def quux(self):
         return 43
 
 class Bar:    # old-style class
     def foo(self):
         return 2
 """)
 
         def check_module_contents():
             nt.assert_equal(mod.x, 10)
             nt.assert_false(hasattr(mod, 'z'))
 
             nt.assert_equal(old_foo(0), 4) # superreload magic!
             nt.assert_equal(mod.foo(0), 4)
 
             obj = mod.Baz(9)
             nt.assert_equal(old_obj.bar(1), 11) # superreload magic!
             nt.assert_equal(obj.bar(1), 11)
 
             nt.assert_equal(old_obj.quux, 43)
             nt.assert_equal(obj.quux, 43)
 
             nt.assert_false(hasattr(old_obj, 'zzz'))
             nt.assert_false(hasattr(obj, 'zzz'))
 
             obj2 = mod.Bar()
             nt.assert_equal(old_obj2.foo(), 2)
             nt.assert_equal(obj2.foo(), 2)
 
         self.shell.run_code("pass") # trigger reload
         check_module_contents()
 
         #
         # Another failure case: deleted file (shouldn't reload)
         #
         os.unlink(mod_fn)
 
         self.shell.run_code("pass") # trigger reload
         check_module_contents()
 
         #
         # Disable autoreload and rewrite module: no reload should occur
         #
         if use_aimport:
             self.shell.magic_aimport("-" + mod_name)
             stream = StringIO()
             self.shell.magic_aimport("", stream=stream)
             nt.assert_true(("Modules to skip:\n%s" % mod_name) in
                            stream.getvalue())
 
             # This should succeed, although no such module exists
             self.shell.magic_aimport("-tmpmod_as318989e89ds")
         else:
             self.shell.magic_autoreload("0")
 
         self.write_file(mod_fn, """
 x = -99
 """)
 
         self.shell.run_code("pass") # trigger reload
         self.shell.run_code("pass")
         check_module_contents()
 
         #
         # Re-enable autoreload: reload should now occur
         #
         if use_aimport:
             self.shell.magic_aimport(mod_name)
         else:
             self.shell.magic_autoreload("")
 
         self.shell.run_code("pass") # trigger reload
         nt.assert_equal(mod.x, -99)
 
     def test_smoketest_aimport(self):
         self._check_smoketest(use_aimport=True)
 
     def test_smoketest_autoreload(self):
         self._check_smoketest(use_aimport=False)

IPython/extensions/tests/test_rmagic.py

@@ -1,122 +1,126
-from io import StringIO
-
 import numpy as np
 from IPython.testing.decorators import skip_without
 from IPython.extensions import rmagic
+from IPython.utils.py3compat import PY3
 from rpy2 import rinterface
 import nose.tools as nt
 
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
 ip = get_ipython()
 ip.magic('load_ext rmagic')
 
 
 def test_push():
     rm = rmagic.RMagics(ip)
     ip.push({'X':np.arange(5), 'Y':np.array([3,5,4,6,7])})
     ip.run_line_magic('Rpush', 'X Y')
     np.testing.assert_almost_equal(np.asarray(rm.r('X')), ip.user_ns['X'])
     np.testing.assert_almost_equal(np.asarray(rm.r('Y')), ip.user_ns['Y'])
 
 def test_push_localscope():
     """Test that Rpush looks for variables in the local scope first."""
     ip.run_cell('''
 def rmagic_addone(u):
     %Rpush u
     %R result = u+1
     %Rpull result
     return result[0]
 u = 0
 result = rmagic_addone(12344)
     ''')
     result = ip.user_ns['result']
     np.testing.assert_equal(result, 12345)
 
 @skip_without('pandas')
 def test_push_dataframe():
     from pandas import DataFrame
     rm = rmagic.RMagics(ip)
     df = DataFrame([{'a': 1, 'b': 'bar'}, {'a': 5, 'b': 'foo', 'c': 20}])
     ip.push({'df':df})
     ip.run_line_magic('Rpush', 'df')
     
     # This is converted to factors, which are currently converted back to Python
     # as integers, so for now we test its representation in R.
     sio = StringIO()
     rinterface.set_writeconsole(sio.write)
     try:
         rm.r('print(df$b[1])')
         nt.assert_in('[1] bar', sio.getvalue())
     finally:
         rinterface.set_writeconsole(None)
     
     # Values come packaged in arrays, so we unbox them to test.
     nt.assert_equal(rm.r('df$a[2]')[0], 5)
     missing = rm.r('df$c[1]')[0]
     assert np.isnan(missing), missing
 
 def test_pull():
     rm = rmagic.RMagics(ip)
     rm.r('Z=c(11:20)')
     ip.run_line_magic('Rpull', 'Z')
     np.testing.assert_almost_equal(np.asarray(rm.r('Z')), ip.user_ns['Z'])
     np.testing.assert_almost_equal(ip.user_ns['Z'], np.arange(11,21))
 
 def test_Rconverter():
     datapy= np.array([(1, 2.9, 'a'), (2, 3.5, 'b'), (3, 2.1, 'c')], 
           dtype=[('x', '<i4'), ('y', '<f8'), ('z', '|S1')])
     ip.user_ns['datapy'] = datapy
     ip.run_line_magic('Rpush', 'datapy')
 
     # test to see if a copy is being made
     v = ip.run_line_magic('Rget', '-d datapy')
     w = ip.run_line_magic('Rget', '-d datapy')
     np.testing.assert_almost_equal(w['x'], v['x'])
     np.testing.assert_almost_equal(w['y'], v['y'])
     nt.assert_true(np.all(w['z'] == v['z']))
     np.testing.assert_equal(id(w.data), id(v.data))
     nt.assert_equal(w.dtype, v.dtype)
 
     ip.run_cell_magic('R', ' -d datar', 'datar=datapy')
 
     u = ip.run_line_magic('Rget', ' -d datar')
     np.testing.assert_almost_equal(u['x'], v['x'])
     np.testing.assert_almost_equal(u['y'], v['y'])
     nt.assert_true(np.all(u['z'] == v['z']))
     np.testing.assert_equal(id(u.data), id(v.data))
     nt.assert_equal(u.dtype, v.dtype)
 
 
 def test_cell_magic():
 
     ip.push({'x':np.arange(5), 'y':np.array([3,5,4,6,7])})
     snippet = '''
     print(summary(a))
     plot(x, y, pch=23, bg='orange', cex=2)
     plot(x, x)
     print(summary(x))
     r = resid(a)
     xc = coef(a)
     '''
     ip.run_cell_magic('R', '-i x,y -o r,xc -w 150 -u mm a=lm(y~x)', snippet)
     np.testing.assert_almost_equal(ip.user_ns['xc'], [3.2, 0.9])
     np.testing.assert_almost_equal(ip.user_ns['r'], np.array([-0.2,  0.9, -1. ,  0.1,  0.2]))
 
 
 def test_rmagic_localscope():
     ip.push({'x':0})
     ip.run_line_magic('R', '-i x -o result result <-x+1')
     result = ip.user_ns['result']
     nt.assert_equal(result[0], 1)
 
     ip.run_cell('''def rmagic_addone(u):
     %R -i u -o result result <- u+1
     return result[0]''')
     ip.run_cell('result = rmagic_addone(1)')
     result = ip.user_ns['result']
     nt.assert_equal(result, 2)
 
     nt.assert_raises(
         NameError,
         ip.run_line_magic,
         "R",
         "-i var_not_defined 1+1")

IPython/kernel/inprocess/tests/test_kernel.py

@@ -1,91 +1,95
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2012  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
 
 # Standard library imports
-from io import StringIO
 import sys
 import unittest
 
 # Local imports
 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
 from IPython.kernel.inprocess.manager import InProcessKernelManager
 from IPython.kernel.inprocess.ipkernel import InProcessKernel
 from IPython.testing.decorators import skipif_not_matplotlib
 from IPython.utils.io import capture_output
 from IPython.utils import py3compat
 
+if py3compat.PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
 #-----------------------------------------------------------------------------
 # Test case
 #-----------------------------------------------------------------------------
 
 class InProcessKernelTestCase(unittest.TestCase):
 
     def setUp(self):
         self.km = InProcessKernelManager()
         self.km.start_kernel()
         self.kc = BlockingInProcessKernelClient(kernel=self.km.kernel)
         self.kc.start_channels()
 
     @skipif_not_matplotlib
     def test_pylab(self):
         """ Does pylab work in the in-process kernel?
         """
         kc = self.kc
         kc.execute('%pylab')
         msg = get_stream_message(kc)
         self.assert_('matplotlib' in msg['content']['data'])
 
     def test_raw_input(self):
         """ Does the in-process kernel handle raw_input correctly?
         """
         io = StringIO('foobar\n')
         sys_stdin = sys.stdin
         sys.stdin = io
         try:
             if py3compat.PY3:
                 self.kc.execute('x = input()')
             else:
                 self.kc.execute('x = raw_input()')
         finally:
             sys.stdin = sys_stdin
         self.assertEqual(self.km.kernel.shell.user_ns.get('x'), 'foobar')
 
     def test_stdout(self):
         """ Does the in-process kernel correctly capture IO?
         """
         kernel = InProcessKernel()
 
         with capture_output() as io:
             kernel.shell.run_cell('print("foo")')
         self.assertEqual(io.stdout, 'foo\n')
 
         kc = BlockingInProcessKernelClient(kernel=kernel)
         kernel.frontends.append(kc)
         kc.shell_channel.execute('print("bar")')
         msg = get_stream_message(kc)
         self.assertEqual(msg['content']['data'], 'bar\n')
 
 #-----------------------------------------------------------------------------
 # Utility functions
 #-----------------------------------------------------------------------------
 
 def get_stream_message(kernel_client, timeout=5):
     """ Gets a single stream message synchronously from the sub channel.
     """
     while True:
         msg = kernel_client.get_iopub_msg(timeout=timeout)
         if msg['header']['msg_type'] == 'stream':
             return msg
 
 
 if __name__ == '__main__':
     unittest.main()

IPython/lib/pretty.py

@@ -1,791 +1,797
 # -*- coding: utf-8 -*-
 """
 Python advanced pretty printer.  This pretty printer is intended to
 replace the old `pprint` python module which does not allow developers
 to provide their own pretty print callbacks.
 
 This module is based on ruby's `prettyprint.rb` library by `Tanaka Akira`.
 
 
 Example Usage
 -------------
 
 To directly print the representation of an object use `pprint`::
 
     from pretty import pprint
     pprint(complex_object)
 
 To get a string of the output use `pretty`::
 
     from pretty import pretty
     string = pretty(complex_object)
 
 
 Extending
 ---------
 
 The pretty library allows developers to add pretty printing rules for their
 own objects.  This process is straightforward.  All you have to do is to
 add a `_repr_pretty_` method to your object and call the methods on the
 pretty printer passed::
 
     class MyObject(object):
 
         def _repr_pretty_(self, p, cycle):
             ...
 
 Depending on the python version you want to support you have two
 possibilities.  The following list shows the python 2.5 version and the
 compatibility one.
 
 
 Here the example implementation of a `_repr_pretty_` method for a list
 subclass for python 2.5 and higher (python 2.5 requires the with statement
 __future__ import)::
 
     class MyList(list):
 
         def _repr_pretty_(self, p, cycle):
             if cycle:
                 p.text('MyList(...)')
             else:
                 with p.group(8, 'MyList([', '])'):
                     for idx, item in enumerate(self):
                         if idx:
                             p.text(',')
                             p.breakable()
                         p.pretty(item)
 
 The `cycle` parameter is `True` if pretty detected a cycle.  You *have* to
 react to that or the result is an infinite loop.  `p.text()` just adds
 non breaking text to the output, `p.breakable()` either adds a whitespace
 or breaks here.  If you pass it an argument it's used instead of the
 default space.  `p.pretty` prettyprints another object using the pretty print
 method.
 
 The first parameter to the `group` function specifies the extra indentation
 of the next line.  In this example the next item will either be not
 breaked (if the items are short enough) or aligned with the right edge of
 the opening bracked of `MyList`.
 
 If you want to support python 2.4 and lower you can use this code::
 
     class MyList(list):
 
         def _repr_pretty_(self, p, cycle):
             if cycle:
                 p.text('MyList(...)')
             else:
                 p.begin_group(8, 'MyList([')
                 for idx, item in enumerate(self):
                     if idx:
                         p.text(',')
                         p.breakable()
                     p.pretty(item)
                 p.end_group(8, '])')
 
 If you just want to indent something you can use the group function
 without open / close parameters.  Under python 2.5 you can also use this
 code::
 
     with p.indent(2):
         ...
 
 Or under python2.4 you might want to modify ``p.indentation`` by hand but
 this is rather ugly.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.lib.pretty
    :parts: 3
 
 :copyright: 2007 by Armin Ronacher.
             Portions (c) 2009 by Robert Kern.
 :license: BSD License.
 """
 from __future__ import print_function
 from contextlib import contextmanager
 import sys
 import types
 import re
 import datetime
-from io import StringIO
 from collections import deque
 
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
+
 
 __all__ = ['pretty', 'pprint', 'PrettyPrinter', 'RepresentationPrinter',
     'for_type', 'for_type_by_name']
 
 
 _re_pattern_type = type(re.compile(''))
 
 
 def pretty(obj, verbose=False, max_width=79, newline='\n'):
     """
     Pretty print the object's representation.
     """
     stream = StringIO()
     printer = RepresentationPrinter(stream, verbose, max_width, newline)
     printer.pretty(obj)
     printer.flush()
     return stream.getvalue()
 
 
 def pprint(obj, verbose=False, max_width=79, newline='\n'):
     """
     Like `pretty` but print to stdout.
     """
     printer = RepresentationPrinter(sys.stdout, verbose, max_width, newline)
     printer.pretty(obj)
     printer.flush()
     sys.stdout.write(newline)
     sys.stdout.flush()
 
 class _PrettyPrinterBase(object):
 
     @contextmanager
     def indent(self, indent):
         """with statement support for indenting/dedenting."""
         self.indentation += indent
         try:
             yield
         finally:
             self.indentation -= indent
 
     @contextmanager
     def group(self, indent=0, open='', close=''):
         """like begin_group / end_group but for the with statement."""
         self.begin_group(indent, open)
         try:
             yield
         finally:
             self.end_group(indent, close)
 
 class PrettyPrinter(_PrettyPrinterBase):
     """
     Baseclass for the `RepresentationPrinter` prettyprinter that is used to
     generate pretty reprs of objects.  Contrary to the `RepresentationPrinter`
     this printer knows nothing about the default pprinters or the `_repr_pretty_`
     callback method.
     """
 
     def __init__(self, output, max_width=79, newline='\n'):
         self.output = output
         self.max_width = max_width
         self.newline = newline
         self.output_width = 0
         self.buffer_width = 0
         self.buffer = deque()
 
         root_group = Group(0)
         self.group_stack = [root_group]
         self.group_queue = GroupQueue(root_group)
         self.indentation = 0
 
     def _break_outer_groups(self):
         while self.max_width < self.output_width + self.buffer_width:
             group = self.group_queue.deq()
             if not group:
                 return
             while group.breakables:
                 x = self.buffer.popleft()
                 self.output_width = x.output(self.output, self.output_width)
                 self.buffer_width -= x.width
             while self.buffer and isinstance(self.buffer[0], Text):
                 x = self.buffer.popleft()
                 self.output_width = x.output(self.output, self.output_width)
                 self.buffer_width -= x.width
 
     def text(self, obj):
         """Add literal text to the output."""
         width = len(obj)
         if self.buffer:
             text = self.buffer[-1]
             if not isinstance(text, Text):
                 text = Text()
                 self.buffer.append(text)
             text.add(obj, width)
             self.buffer_width += width
             self._break_outer_groups()
         else:
             self.output.write(obj)
             self.output_width += width
 
     def breakable(self, sep=' '):
         """
         Add a breakable separator to the output.  This does not mean that it
         will automatically break here.  If no breaking on this position takes
         place the `sep` is inserted which default to one space.
         """
         width = len(sep)
         group = self.group_stack[-1]
         if group.want_break:
             self.flush()
             self.output.write(self.newline)
             self.output.write(' ' * self.indentation)
             self.output_width = self.indentation
             self.buffer_width = 0
         else:
             self.buffer.append(Breakable(sep, width, self))
             self.buffer_width += width
             self._break_outer_groups()
             
     def break_(self):
         """
         Explicitly insert a newline into the output, maintaining correct indentation.
         """
         self.flush()
         self.output.write(self.newline)
         self.output.write(' ' * self.indentation)
         self.output_width = self.indentation
         self.buffer_width = 0
         
 
     def begin_group(self, indent=0, open=''):
         """
         Begin a group.  If you want support for python < 2.5 which doesn't has
         the with statement this is the preferred way:
 
             p.begin_group(1, '{')
             ...
             p.end_group(1, '}')
 
         The python 2.5 expression would be this:
 
             with p.group(1, '{', '}'):
                 ...
 
         The first parameter specifies the indentation for the next line (usually
         the width of the opening text), the second the opening text.  All
         parameters are optional.
         """
         if open:
             self.text(open)
         group = Group(self.group_stack[-1].depth + 1)
         self.group_stack.append(group)
         self.group_queue.enq(group)
         self.indentation += indent
 
     def end_group(self, dedent=0, close=''):
         """End a group. See `begin_group` for more details."""
         self.indentation -= dedent
         group = self.group_stack.pop()
         if not group.breakables:
             self.group_queue.remove(group)
         if close:
             self.text(close)
 
     def flush(self):
         """Flush data that is left in the buffer."""
         for data in self.buffer:
             self.output_width += data.output(self.output, self.output_width)
         self.buffer.clear()
         self.buffer_width = 0
 
 
 def _get_mro(obj_class):
     """ Get a reasonable method resolution order of a class and its superclasses
     for both old-style and new-style classes.
     """
     if not hasattr(obj_class, '__mro__'):
         # Old-style class. Mix in object to make a fake new-style class.
         try:
             obj_class = type(obj_class.__name__, (obj_class, object), {})
         except TypeError:
             # Old-style extension type that does not descend from object.
             # FIXME: try to construct a more thorough MRO.
             mro = [obj_class]
         else:
             mro = obj_class.__mro__[1:-1]
     else:
         mro = obj_class.__mro__
     return mro
 
 
 class RepresentationPrinter(PrettyPrinter):
     """
     Special pretty printer that has a `pretty` method that calls the pretty
     printer for a python object.
 
     This class stores processing data on `self` so you must *never* use
     this class in a threaded environment.  Always lock it or reinstanciate
     it.
 
     Instances also have a verbose flag callbacks can access to control their
     output.  For example the default instance repr prints all attributes and
     methods that are not prefixed by an underscore if the printer is in
     verbose mode.
     """
 
     def __init__(self, output, verbose=False, max_width=79, newline='\n',
         singleton_pprinters=None, type_pprinters=None, deferred_pprinters=None):
 
         PrettyPrinter.__init__(self, output, max_width, newline)
         self.verbose = verbose
         self.stack = []
         if singleton_pprinters is None:
             singleton_pprinters = _singleton_pprinters.copy()
         self.singleton_pprinters = singleton_pprinters
         if type_pprinters is None:
             type_pprinters = _type_pprinters.copy()
         self.type_pprinters = type_pprinters
         if deferred_pprinters is None:
             deferred_pprinters = _deferred_type_pprinters.copy()
         self.deferred_pprinters = deferred_pprinters
 
     def pretty(self, obj):
         """Pretty print the given object."""
         obj_id = id(obj)
         cycle = obj_id in self.stack
         self.stack.append(obj_id)
         self.begin_group()
         try:
             obj_class = getattr(obj, '__class__', None) or type(obj)
             # First try to find registered singleton printers for the type.
             try:
                 printer = self.singleton_pprinters[obj_id]
             except (TypeError, KeyError):
                 pass
             else:
                 return printer(obj, self, cycle)
             # Next walk the mro and check for either:
             #   1) a registered printer
             #   2) a _repr_pretty_ method
             for cls in _get_mro(obj_class):
                 if cls in self.type_pprinters:
                     # printer registered in self.type_pprinters
                     return self.type_pprinters[cls](obj, self, cycle)
                 else:
                     # deferred printer
                     printer = self._in_deferred_types(cls)
                     if printer is not None:
                         return printer(obj, self, cycle)
                     else:
                         # Finally look for special method names.
                         # Some objects automatically create any requested
                         # attribute. Try to ignore most of them by checking for
                         # callability.
                         if '_repr_pretty_' in cls.__dict__:
                             meth = cls._repr_pretty_
                             if callable(meth):
                                 return meth(obj, self, cycle)
             return _default_pprint(obj, self, cycle)
         finally:
             self.end_group()
             self.stack.pop()
 
     def _in_deferred_types(self, cls):
         """
         Check if the given class is specified in the deferred type registry.
 
         Returns the printer from the registry if it exists, and None if the
         class is not in the registry. Successful matches will be moved to the
         regular type registry for future use.
         """
         mod = getattr(cls, '__module__', None)
         name = getattr(cls, '__name__', None)
         key = (mod, name)
         printer = None
         if key in self.deferred_pprinters:
             # Move the printer over to the regular registry.
             printer = self.deferred_pprinters.pop(key)
             self.type_pprinters[cls] = printer
         return printer
 
 
 class Printable(object):
 
     def output(self, stream, output_width):
         return output_width
 
 
 class Text(Printable):
 
     def __init__(self):
         self.objs = []
         self.width = 0
 
     def output(self, stream, output_width):
         for obj in self.objs:
             stream.write(obj)
         return output_width + self.width
 
     def add(self, obj, width):
         self.objs.append(obj)
         self.width += width
 
 
 class Breakable(Printable):
 
     def __init__(self, seq, width, pretty):
         self.obj = seq
         self.width = width
         self.pretty = pretty
         self.indentation = pretty.indentation
         self.group = pretty.group_stack[-1]
         self.group.breakables.append(self)
 
     def output(self, stream, output_width):
         self.group.breakables.popleft()
         if self.group.want_break:
             stream.write(self.pretty.newline)
             stream.write(' ' * self.indentation)
             return self.indentation
         if not self.group.breakables:
             self.pretty.group_queue.remove(self.group)
         stream.write(self.obj)
         return output_width + self.width
 
 
 class Group(Printable):
 
     def __init__(self, depth):
         self.depth = depth
         self.breakables = deque()
         self.want_break = False
 
 
 class GroupQueue(object):
 
     def __init__(self, *groups):
         self.queue = []
         for group in groups:
             self.enq(group)
 
     def enq(self, group):
         depth = group.depth
         while depth > len(self.queue) - 1:
             self.queue.append([])
         self.queue[depth].append(group)
 
     def deq(self):
         for stack in self.queue:
             for idx, group in enumerate(reversed(stack)):
                 if group.breakables:
                     del stack[idx]
                     group.want_break = True
                     return group
             for group in stack:
                 group.want_break = True
             del stack[:]
 
     def remove(self, group):
         try:
             self.queue[group.depth].remove(group)
         except ValueError:
             pass
 
 try:
     _baseclass_reprs = (object.__repr__, types.InstanceType.__repr__)
 except AttributeError:  # Python 3
     _baseclass_reprs = (object.__repr__,)
 
 
 def _default_pprint(obj, p, cycle):
     """
     The default print function.  Used if an object does not provide one and
     it's none of the builtin objects.
     """
     klass = getattr(obj, '__class__', None) or type(obj)
     if getattr(klass, '__repr__', None) not in _baseclass_reprs:
         # A user-provided repr. Find newlines and replace them with p.break_()
         output = repr(obj)
         for idx,output_line in enumerate(output.splitlines()):
             if idx:
                 p.break_()
             p.text(output_line)
         return
     p.begin_group(1, '<')
     p.pretty(klass)
     p.text(' at 0x%x' % id(obj))
     if cycle:
         p.text(' ...')
     elif p.verbose:
         first = True
         for key in dir(obj):
             if not key.startswith('_'):
                 try:
                     value = getattr(obj, key)
                 except AttributeError:
                     continue
                 if isinstance(value, types.MethodType):
                     continue
                 if not first:
                     p.text(',')
                 p.breakable()
                 p.text(key)
                 p.text('=')
                 step = len(key) + 1
                 p.indentation += step
                 p.pretty(value)
                 p.indentation -= step
                 first = False
     p.end_group(1, '>')
 
 
 def _seq_pprinter_factory(start, end, basetype):
     """
     Factory that returns a pprint function useful for sequences.  Used by
     the default pprint for tuples, dicts, and lists.
     """
     def inner(obj, p, cycle):
         typ = type(obj)
         if basetype is not None and typ is not basetype and typ.__repr__ != basetype.__repr__:
             # If the subclass provides its own repr, use it instead.
             return p.text(typ.__repr__(obj))
 
         if cycle:
             return p.text(start + '...' + end)
         step = len(start)
         p.begin_group(step, start)
         for idx, x in enumerate(obj):
             if idx:
                 p.text(',')
                 p.breakable()
             p.pretty(x)
         if len(obj) == 1 and type(obj) is tuple:
             # Special case for 1-item tuples.
             p.text(',')
         p.end_group(step, end)
     return inner
 
 
 def _set_pprinter_factory(start, end, basetype):
     """
     Factory that returns a pprint function useful for sets and frozensets.
     """
     def inner(obj, p, cycle):
         typ = type(obj)
         if basetype is not None and typ is not basetype and typ.__repr__ != basetype.__repr__:
             # If the subclass provides its own repr, use it instead.
             return p.text(typ.__repr__(obj))
 
         if cycle:
             return p.text(start + '...' + end)
         if len(obj) == 0:
             # Special case.
             p.text(basetype.__name__ + '()')
         else:
             step = len(start)
             p.begin_group(step, start)
             # Like dictionary keys, we will try to sort the items.
             items = list(obj)
             try:
                 items.sort()
             except Exception:
                 # Sometimes the items don't sort.
                 pass
             for idx, x in enumerate(items):
                 if idx:
                     p.text(',')
                     p.breakable()
                 p.pretty(x)
             p.end_group(step, end)
     return inner
 
 
 def _dict_pprinter_factory(start, end, basetype=None):
     """
     Factory that returns a pprint function used by the default pprint of
     dicts and dict proxies.
     """
     def inner(obj, p, cycle):
         typ = type(obj)
         if basetype is not None and typ is not basetype and typ.__repr__ != basetype.__repr__:
             # If the subclass provides its own repr, use it instead.
             return p.text(typ.__repr__(obj))
 
         if cycle:
             return p.text('{...}')
         p.begin_group(1, start)
         keys = obj.keys()
         try:
             keys.sort()
         except Exception as e:
             # Sometimes the keys don't sort.
             pass
         for idx, key in enumerate(keys):
             if idx:
                 p.text(',')
                 p.breakable()
             p.pretty(key)
             p.text(': ')
             p.pretty(obj[key])
         p.end_group(1, end)
     return inner
 
 
 def _super_pprint(obj, p, cycle):
     """The pprint for the super type."""
     p.begin_group(8, '<super: ')
     p.pretty(obj.__self_class__)
     p.text(',')
     p.breakable()
     p.pretty(obj.__self__)
     p.end_group(8, '>')
 
 
 def _re_pattern_pprint(obj, p, cycle):
     """The pprint function for regular expression patterns."""
     p.text('re.compile(')
     pattern = repr(obj.pattern)
     if pattern[:1] in 'uU':
         pattern = pattern[1:]
         prefix = 'ur'
     else:
         prefix = 'r'
     pattern = prefix + pattern.replace('\\\\', '\\')
     p.text(pattern)
     if obj.flags:
         p.text(',')
         p.breakable()
         done_one = False
         for flag in ('TEMPLATE', 'IGNORECASE', 'LOCALE', 'MULTILINE', 'DOTALL',
             'UNICODE', 'VERBOSE', 'DEBUG'):
             if obj.flags & getattr(re, flag):
                 if done_one:
                     p.text('|')
                 p.text('re.' + flag)
                 done_one = True
     p.text(')')
 
 
 def _type_pprint(obj, p, cycle):
     """The pprint for classes and types."""
     mod = getattr(obj, '__module__', None)
     if mod is None:
         # Heap allocated types might not have the module attribute,
         # and others may set it to None.
         return p.text(obj.__name__)
 
     if mod in ('__builtin__', 'builtins', 'exceptions'):
         name = obj.__name__
     else:
         name = mod + '.' + obj.__name__
     p.text(name)
 
 
 def _repr_pprint(obj, p, cycle):
     """A pprint that just redirects to the normal repr function."""
     p.text(repr(obj))
 
 
 def _function_pprint(obj, p, cycle):
     """Base pprint for all functions and builtin functions."""
     if obj.__module__ in ('__builtin__', 'builtins', 'exceptions') or not obj.__module__:
         name = obj.__name__
     else:
         name = obj.__module__ + '.' + obj.__name__
     p.text('<function %s>' % name)
 
 
 def _exception_pprint(obj, p, cycle):
     """Base pprint for all exceptions."""
     if obj.__class__.__module__ in ('exceptions', 'builtins'):
         name = obj.__class__.__name__
     else:
         name = '%s.%s' % (
             obj.__class__.__module__,
             obj.__class__.__name__
         )
     step = len(name) + 1
     p.begin_group(step, name + '(')
     for idx, arg in enumerate(getattr(obj, 'args', ())):
         if idx:
             p.text(',')
             p.breakable()
         p.pretty(arg)
     p.end_group(step, ')')
 
 
 #: the exception base
 try:
     _exception_base = BaseException
 except NameError:
     _exception_base = Exception
 
 
 #: printers for builtin types
 _type_pprinters = {
     int:                        _repr_pprint,
     float:                      _repr_pprint,
     str:                        _repr_pprint,
     tuple:                      _seq_pprinter_factory('(', ')', tuple),
     list:                       _seq_pprinter_factory('[', ']', list),
     dict:                       _dict_pprinter_factory('{', '}', dict),
     
     set:                        _set_pprinter_factory('{', '}', set),
     frozenset:                  _set_pprinter_factory('frozenset({', '})', frozenset),
     super:                      _super_pprint,
     _re_pattern_type:           _re_pattern_pprint,
     type:                       _type_pprint,
     types.FunctionType:         _function_pprint,
     types.BuiltinFunctionType:  _function_pprint,
     types.MethodType:           _repr_pprint,
     
     datetime.datetime:          _repr_pprint,
     datetime.timedelta:         _repr_pprint,
     _exception_base:            _exception_pprint
 }
 
 try:
     _type_pprinters[types.DictProxyType] = _dict_pprinter_factory('<dictproxy {', '}>')
     _type_pprinters[types.ClassType] = _type_pprint
     _type_pprinters[types.SliceType] = _repr_pprint
 except AttributeError: # Python 3
     _type_pprinters[slice] = _repr_pprint
     
 try:
     _type_pprinters[xrange] = _repr_pprint
     _type_pprinters[long] = _repr_pprint
     _type_pprinters[unicode] = _repr_pprint
 except NameError:
     _type_pprinters[range] = _repr_pprint
     _type_pprinters[bytes] = _repr_pprint
 
 #: printers for types specified by name
 _deferred_type_pprinters = {
 }
 
 def for_type(typ, func):
     """
     Add a pretty printer for a given type.
     """
     oldfunc = _type_pprinters.get(typ, None)
     if func is not None:
         # To support easy restoration of old pprinters, we need to ignore Nones.
         _type_pprinters[typ] = func
     return oldfunc
 
 def for_type_by_name(type_module, type_name, func):
     """
     Add a pretty printer for a type specified by the module and name of a type
     rather than the type object itself.
     """
     key = (type_module, type_name)
     oldfunc = _deferred_type_pprinters.get(key, None)
     if func is not None:
         # To support easy restoration of old pprinters, we need to ignore Nones.
         _deferred_type_pprinters[key] = func
     return oldfunc
 
 
 #: printers for the default singletons
 _singleton_pprinters = dict.fromkeys(map(id, [None, True, False, Ellipsis,
                                       NotImplemented]), _repr_pprint)
 
 
 if __name__ == '__main__':
     from random import randrange
     class Foo(object):
         def __init__(self):
             self.foo = 1
             self.bar = re.compile(r'\s+')
             self.blub = dict.fromkeys(range(30), randrange(1, 40))
             self.hehe = 23424.234234
             self.list = ["blub", "blah", self]
 
         def get_foo(self):
             print("foo")
 
     pprint(Foo(), verbose=True)

IPython/lib/tests/test_irunner.py

@@ -1,180 +1,184
 """Test suite for the irunner module.
 
 Not the most elegant or fine-grained, but it does cover at least the bulk
 functionality."""
 from __future__ import print_function
 
 # Global to make tests extra verbose and help debugging
 VERBOSE = True
 
 # stdlib imports
-import io
 import sys
 import unittest
 
 # IPython imports
 from IPython.lib import irunner
-from IPython.utils.py3compat import doctest_refactor_print
+from IPython.utils.py3compat import doctest_refactor_print, PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 # Testing code begins
 class RunnerTestCase(unittest.TestCase):
 
     def setUp(self):
-        self.out = io.StringIO()
+        self.out = StringIO()
         #self.out = sys.stdout
 
     def _test_runner(self,runner,source,output):
         """Test that a given runner's input/output match."""
         
         runner.run_source(source)
         out = self.out.getvalue()
         #out = ''
         # this output contains nasty \r\n lineends, and the initial ipython
         # banner.  clean it up for comparison, removing lines of whitespace
         output_l = [l for l in output.splitlines() if l and not l.isspace()]
         out_l = [l for l in out.splitlines() if l and not l.isspace()]
         mismatch  = 0
         if len(output_l) != len(out_l):
             message = ("Mismatch in number of lines\n\n"
                        "Expected:\n"
                        "~~~~~~~~~\n"
                        "%s\n\n"
                        "Got:\n"
                        "~~~~~~~~~\n"
                        "%s"
                        ) % ("\n".join(output_l), "\n".join(out_l))
             self.fail(message)
         for n in range(len(output_l)):
             # Do a line-by-line comparison
             ol1 = output_l[n].strip()
             ol2 = out_l[n].strip()
             if ol1 != ol2:
                 mismatch += 1
                 if VERBOSE:
                     print('<<< line %s does not match:' % n)
                     print(repr(ol1))
                     print(repr(ol2))
                     print('>>>')
         self.assertTrue(mismatch==0,'Number of mismatched lines: %s' %
                         mismatch)
 
     def testIPython(self):
         """Test the IPython runner."""
         source = doctest_refactor_print("""
 print 'hello, this is python'
 # some more code
 x=1;y=2
 x+y**2
 
 # An example of autocall functionality
 from math import *
 autocall 1
 cos pi
 autocall 0
 cos pi
 cos(pi)
 
 for i in range(5):
     print i
 
 print "that's all folks!"
 
 exit
 """)
         output = doctest_refactor_print("""\
 In [1]: print 'hello, this is python'
 hello, this is python
 
 
 # some more code
 In [2]: x=1;y=2
 
 In [3]: x+y**2
 Out[3]: 5
 
 
 # An example of autocall functionality
 In [4]: from math import *
 
 In [5]: autocall 1
 Automatic calling is: Smart
 
 In [6]: cos pi
 ------> cos(pi)
 Out[6]: -1.0
 
 In [7]: autocall 0
 Automatic calling is: OFF
 
 In [8]: cos pi
    File "<ipython-input-8-6bd7313dd9a9>", line 1
      cos pi
           ^
 SyntaxError: invalid syntax
 
 
 In [9]: cos(pi)
 Out[9]: -1.0
 
 
 In [10]: for i in range(5):
    ....:     print i
    ....:
 0
 1
 2
 3
 4
 
 In [11]: print "that's all folks!"
 that's all folks!
 
 
 In [12]: exit
 """)
         runner = irunner.IPythonRunner(out=self.out)
         self._test_runner(runner,source,output)
 
     def testPython(self):
         """Test the Python runner."""
         runner = irunner.PythonRunner(out=self.out, args=['-E'])
         source = doctest_refactor_print("""
 print 'hello, this is python'
 
 # some more code
 x=1;y=2
 x+y**2
 
 from math import *
 cos(pi)
 
 for i in range(5):
     print i
 
 print "that's all folks!"
         """)
         output = doctest_refactor_print("""\
 >>> print 'hello, this is python'
 hello, this is python
 
 # some more code
 >>> x=1;y=2
 >>> x+y**2
 5
 
 >>> from math import *
 >>> cos(pi)
 -1.0
 
 >>> for i in range(5):
 ...     print i
 ...
 0
 1
 2
 3
 4
 >>> print "that's all folks!"
 that's all folks!
 """)
         self._test_runner(runner,source,output)

IPython/lib/tests/test_irunner_pylab_magic.py

@@ -1,121 +1,126
 """Test suite for pylab_import_all magic
 Modified from the irunner module but using regex.
 """
 from __future__ import print_function
 
 # Global to make tests extra verbose and help debugging
 VERBOSE = True
 
 # stdlib imports
-import io
 import sys
 import unittest
 import re
 
 # IPython imports
 from IPython.lib import irunner
 from IPython.testing import decorators
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 def pylab_not_importable():
     """Test if importing pylab fails. (For example, when having no display)"""
     try:
         import pylab
         return False
     except:
         return True
 
 # Testing code begins
 class RunnerTestCase(unittest.TestCase):
 
     def setUp(self):
-        self.out = io.StringIO()
+        self.out = StringIO()
         #self.out = sys.stdout
 
     def _test_runner(self,runner,source,output):
         """Test that a given runner's input/output match."""
 
         runner.run_source(source)
         out = self.out.getvalue()
         #out = ''
         # this output contains nasty \r\n lineends, and the initial ipython
         # banner.  clean it up for comparison, removing lines of whitespace
         output_l = [l for l in output.splitlines() if l and not l.isspace()]
         out_l = [l for l in out.splitlines() if l and not l.isspace()]
         mismatch  = 0
         if len(output_l) != len(out_l):
             message = ("Mismatch in number of lines\n\n"
                        "Expected:\n"
                        "~~~~~~~~~\n"
                        "%s\n\n"
                        "Got:\n"
                        "~~~~~~~~~\n"
                        "%s"
                        ) % ("\n".join(output_l), "\n".join(out_l))
             self.fail(message)
         for n in range(len(output_l)):
             # Do a line-by-line comparison
             ol1 = output_l[n].strip()
             ol2 = out_l[n].strip()
             if not re.match(ol1,ol2):
                 mismatch += 1
                 if VERBOSE:
                     print('<<< line %s does not match:' % n)
                     print(repr(ol1))
                     print(repr(ol2))
                     print('>>>')
         self.assertTrue(mismatch==0,'Number of mismatched lines: %s' %
                         mismatch)
 
     @decorators.skip_if_no_x11
     @decorators.skipif_not_matplotlib
     @decorators.skipif(pylab_not_importable, "Likely a run without X.")
     def test_pylab_import_all_enabled(self):
         "Verify that plot is available when pylab_import_all = True"
         source = """
 from IPython.config.application import Application
 app = Application.instance()
 app.pylab_import_all = True
 pylab
 ip=get_ipython()
 'plot' in ip.user_ns
         """
         output = """
 In \[1\]: from IPython\.config\.application import Application
 In \[2\]: app = Application\.instance\(\)
 In \[3\]: app\.pylab_import_all = True
 In \[4\]: pylab
 ^Using matplotlib backend:
 Populating the interactive namespace from numpy and matplotlib
 In \[5\]: ip=get_ipython\(\)
 In \[6\]: \'plot\' in ip\.user_ns
 Out\[6\]: True
 """
         runner = irunner.IPythonRunner(out=self.out)
         self._test_runner(runner,source,output)
 
     @decorators.skip_if_no_x11
     @decorators.skipif_not_matplotlib
     @decorators.skipif(pylab_not_importable, "Likely a run without X.")
     def test_pylab_import_all_disabled(self):
         "Verify that plot is not available when pylab_import_all = False"
         source = """
 from IPython.config.application import Application
 app = Application.instance()
 app.pylab_import_all = False
 pylab
 ip=get_ipython()
 'plot' in ip.user_ns
         """
         output = """
 In \[1\]: from IPython\.config\.application import Application
 In \[2\]: app = Application\.instance\(\)
 In \[3\]: app\.pylab_import_all = False
 In \[4\]: pylab
 ^Using matplotlib backend:
 Populating the interactive namespace from numpy and matplotlib
 In \[5\]: ip=get_ipython\(\)
 In \[6\]: \'plot\' in ip\.user_ns
 Out\[6\]: False
 """
         runner = irunner.IPythonRunner(out=self.out)
         self._test_runner(runner,source,output)

IPython/nbconvert/writers/tests/test_debug.py

@@ -1,51 +1,56
 """
 Module with tests for debug
 """
 
 #-----------------------------------------------------------------------------
 # 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 sys
-from io import StringIO
 
 from ...tests.base import TestsBase
 from ..debug import DebugWriter
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 
 #-----------------------------------------------------------------------------
 # Class
 #-----------------------------------------------------------------------------
 
 class TestDebug(TestsBase):
     """Contains test functions for debug.py"""
 
     def test_output(self):
         """Test debug writer output."""
         
         # Capture the stdout.  Remember original.
         stdout = sys.stdout
         stream = StringIO()
         sys.stdout = stream
 
         # Create stdout writer, get output
         writer = DebugWriter()
         writer.write('aaa', {'outputs': {'bbb': 'ccc'}})
         output = stream.getvalue()
 
         # Check output.  Make sure resources dictionary is dumped, but nothing
         # else.
         assert 'aaa' not in output
         assert 'bbb' in output
         assert 'ccc' in output
 
         # Revert stdout
         sys.stdout = stdout
  No newline at end of file

IPython/nbconvert/writers/tests/test_files.py

@@ -1,159 +1,164
 """
 Module with tests for files
 """
 
 #-----------------------------------------------------------------------------
 # 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 sys
 import os
-from io import StringIO
 
 from ...tests.base import TestsBase
 from ..files import FilesWriter
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 
 #-----------------------------------------------------------------------------
 # Class
 #-----------------------------------------------------------------------------
 
 class Testfiles(TestsBase):
     """Contains test functions for files.py"""
 
     def test_basic_output(self):
         """Is FilesWriter basic output correct?"""
 
         # Work in a temporary directory.
         with self.create_temp_cwd():
 
             # Create the resoruces dictionary
             res = {}
 
             # Create files writer, test output
             writer = FilesWriter()
             writer.write(u'y', res, notebook_name="z")
 
             # Check the output of the file
             with open('z', 'r') as f:
                 output = f.read()
                 self.assertEqual(output, u'y')
 
     def test_ext(self):
         """Does the FilesWriter add the correct extension to the output?"""
 
         # Work in a temporary directory.
         with self.create_temp_cwd():
 
             # Create the resoruces dictionary
             res = {'output_extension': 'txt'}
 
             # Create files writer, test output
             writer = FilesWriter()
             writer.write(u'y', res, notebook_name="z")
 
             # Check the output of the file
             assert os.path.isfile('z.txt')
             with open('z.txt', 'r') as f:
                 output = f.read()
                 self.assertEqual(output, u'y')
 
 
     def test_extract(self):
         """Can FilesWriter write extracted figures correctly?"""
 
         # Work in a temporary directory.
         with self.create_temp_cwd():
 
             # Create the resoruces dictionary
             res = {'outputs': {os.path.join('z_files', 'a'): b'b'}}
 
             # Create files writer, test output
             writer = FilesWriter()
             writer.write(u'y', res, notebook_name="z")
 
             # Check the output of the file
             with open('z', 'r') as f:
                 output = f.read()
                 self.assertEqual(output, u'y')
 
             # Check the output of the extracted file
             extracted_file_dest = os.path.join('z_files', 'a')
             assert os.path.isfile(extracted_file_dest)
             with open(extracted_file_dest, 'r') as f:
                 output = f.read()
                 self.assertEqual(output, 'b')
 
 
     def test_builddir(self):
         """Can FilesWriter write to a build dir correctly?"""
 
         # Work in a temporary directory.
         with self.create_temp_cwd():
 
             # Create the resoruces dictionary
             res = {'outputs': {os.path.join('z_files', 'a'): b'b'}}
 
             # Create files writer, test output
             writer = FilesWriter()
             writer.build_directory = u'build'
             writer.write(u'y', res, notebook_name="z")
 
             # Check the output of the file
             assert os.path.isdir(writer.build_directory)
             dest = os.path.join(writer.build_directory, 'z')
             with open(dest, 'r') as f:
                 output = f.read()
                 self.assertEqual(output, u'y')
 
             # Check the output of the extracted file
             extracted_file_dest = os.path.join(writer.build_directory, 'z_files', 'a')
             assert os.path.isfile(extracted_file_dest)
             with open(extracted_file_dest, 'r') as f:
                 output = f.read()
                 self.assertEqual(output, 'b')
 
 
     def test_links(self):
         """Can the FilesWriter handle linked files correctly?"""
 
         # Work in a temporary directory.
         with self.create_temp_cwd():
 
             # Create test file
             os.mkdir('sub')
             with open(os.path.join('sub', 'c'), 'w') as f:
                 f.write('d')
 
             # Create the resoruces dictionary
             res = {}
 
             # Create files writer, test output
             writer = FilesWriter()
             writer.files = [os.path.join('sub', 'c')]
             writer.build_directory = u'build'
             writer.write(u'y', res, notebook_name="z")
 
             # Check the output of the file
             assert os.path.isdir(writer.build_directory)
             dest = os.path.join(writer.build_directory, 'z')
             with open(dest, 'r') as f:
                 output = f.read()
                 self.assertEqual(output, u'y')
 
             # Check to make sure the linked file was copied
             path = os.path.join(writer.build_directory, 'sub')
             assert os.path.isdir(path)
             dest = os.path.join(path, 'c')
             assert os.path.isfile(dest)
             with open(dest, 'r') as f:
                 output = f.read()
                 self.assertEqual(output, 'd')

IPython/nbconvert/writers/tests/test_stdout.py

@@ -1,46 +1,51
 """
 Module with tests for stdout
 """
 
 #-----------------------------------------------------------------------------
 # 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 sys
-from io import StringIO
 
 from ...tests.base import TestsBase
 from ..stdout import StdoutWriter
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 
 #-----------------------------------------------------------------------------
 # Class
 #-----------------------------------------------------------------------------
 
 class TestStdout(TestsBase):
     """Contains test functions for stdout.py"""
 
     def test_output(self):
         """Test stdout writer output."""
         
         # Capture the stdout.  Remember original.
         stdout = sys.stdout
         stream = StringIO()
         sys.stdout = stream
 
         # Create stdout writer, test output
         writer = StdoutWriter()
         writer.write('a', {'b': 'c'})
         output = stream.getvalue()
         self.fuzzy_compare(output, 'a')
 
         # Revert stdout
         sys.stdout = stdout
  No newline at end of file

IPython/parallel/tests/clienttest.py

@@ -1,193 +1,192
 """base class for parallel client tests
 
 Authors:
 
 * Min RK
 """
 
 #-------------------------------------------------------------------------------
 #  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.
 #-------------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 import tempfile
 import time
-from io import StringIO
 
 from nose import SkipTest
 
 import zmq
 from zmq.tests import BaseZMQTestCase
 
 from IPython.external.decorator import decorator
 
 from IPython.parallel import error
 from IPython.parallel import Client
 
 from IPython.parallel.tests import launchers, add_engines
 
 # simple tasks for use in apply tests
 
 def segfault():
     """this will segfault"""
     import ctypes
     ctypes.memset(-1,0,1)
 
 def crash():
     """from stdlib crashers in the test suite"""
     import types
     if sys.platform.startswith('win'):
         import ctypes
         ctypes.windll.kernel32.SetErrorMode(0x0002);
     args = [ 0, 0, 0, 0, b'\x04\x71\x00\x00', (), (), (), '', '', 1, b'']
     if sys.version_info[0] >= 3:
         # Python3 adds 'kwonlyargcount' as the second argument to Code
         args.insert(1, 0)
         
     co = types.CodeType(*args)
     exec(co)
 
 def wait(n):
     """sleep for a time"""
     import time
     time.sleep(n)
     return n
 
 def raiser(eclass):
     """raise an exception"""
     raise eclass()
 
 def generate_output():
     """function for testing output
     
     publishes two outputs of each type, and returns
     a rich displayable object.
     """
     
     import sys
     from IPython.core.display import display, HTML, Math
     
     print("stdout")
     print("stderr", file=sys.stderr)
     
     display(HTML("<b>HTML</b>"))
     
     print("stdout2")
     print("stderr2", file=sys.stderr)
     
     display(Math(r"\alpha=\beta"))
     
     return Math("42")
 
 # test decorator for skipping tests when libraries are unavailable
 def skip_without(*names):
     """skip a test if some names are not importable"""
     @decorator
     def skip_without_names(f, *args, **kwargs):
         """decorator to skip tests in the absence of numpy."""
         for name in names:
             try:
                 __import__(name)
             except ImportError:
                 raise SkipTest
         return f(*args, **kwargs)
     return skip_without_names
 
 #-------------------------------------------------------------------------------
 # Classes
 #-------------------------------------------------------------------------------
 
 
 class ClusterTestCase(BaseZMQTestCase):
     timeout = 10
     
     def add_engines(self, n=1, block=True):
         """add multiple engines to our cluster"""
         self.engines.extend(add_engines(n))
         if block:
             self.wait_on_engines()
 
     def minimum_engines(self, n=1, block=True):
         """add engines until there are at least n connected"""
         self.engines.extend(add_engines(n, total=True))
         if block:
             self.wait_on_engines()
             
     
     def wait_on_engines(self, timeout=5):
         """wait for our engines to connect."""
         n = len(self.engines)+self.base_engine_count
         tic = time.time()
         while time.time()-tic < timeout and len(self.client.ids) < n:
             time.sleep(0.1)
         
         assert not len(self.client.ids) < n, "waiting for engines timed out"
     
     def client_wait(self, client, jobs=None, timeout=-1):
         """my wait wrapper, sets a default finite timeout to avoid hangs"""
         if timeout < 0:
             timeout = self.timeout
         return Client.wait(client, jobs, timeout)
     
     def connect_client(self):
         """connect a client with my Context, and track its sockets for cleanup"""
         c = Client(profile='iptest', context=self.context)
         c.wait = lambda *a, **kw: self.client_wait(c, *a, **kw)
         
         for name in filter(lambda n:n.endswith('socket'), dir(c)):
             s = getattr(c, name)
             s.setsockopt(zmq.LINGER, 0)
             self.sockets.append(s)
         return c
     
     def assertRaisesRemote(self, etype, f, *args, **kwargs):
         try:
             try:
                 f(*args, **kwargs)
             except error.CompositeError as e:
                 e.raise_exception()
         except error.RemoteError as e:
             self.assertEqual(etype.__name__, e.ename, "Should have raised %r, but raised %r"%(etype.__name__, e.ename))
         else:
             self.fail("should have raised a RemoteError")
             
     def _wait_for(self, f, timeout=10):
         """wait for a condition"""
         tic = time.time()
         while time.time() <= tic + timeout:
             if f():
                 return
             time.sleep(0.1)
             self.client.spin()
         if not f():
             print("Warning: Awaited condition never arrived")
     
     def setUp(self):
         BaseZMQTestCase.setUp(self)
         self.client = self.connect_client()
         # start every test with clean engine namespaces:
         self.client.clear(block=True)
         self.base_engine_count=len(self.client.ids)
         self.engines=[]
     
     def tearDown(self):
         # self.client.clear(block=True)
         # close fds:
         for e in filter(lambda e: e.poll() is not None, launchers):
             launchers.remove(e)
         
         # allow flushing of incoming messages to prevent crash on socket close
         self.client.wait(timeout=2)
         # time.sleep(2)
         self.client.spin()
         self.client.close()
         BaseZMQTestCase.tearDown(self)
         # this will be redundant when pyzmq merges PR #88
         # self.context.term()
         # print tempfile.TemporaryFile().fileno(),
         # sys.stdout.flush()
         

IPython/sphinxext/ipython_directive.py

@@ -1,830 +1,835
 # -*- coding: utf-8 -*-
 """Sphinx directive to support embedded IPython code.
 
 This directive allows pasting of entire interactive IPython sessions, prompts
 and all, and their code will actually get re-executed at doc build time, with
 all prompts renumbered sequentially. It also allows you to input code as a pure
 python input by giving the argument python to the directive. The output looks
 like an interactive ipython section.
 
 To enable this directive, simply list it in your Sphinx ``conf.py`` file
 (making sure the directory where you placed it is visible to sphinx, as is
 needed for all Sphinx directives).
 
 By default this directive assumes that your prompts are unchanged IPython ones,
 but this can be customized. The configurable options that can be placed in
 conf.py are
 
 ipython_savefig_dir:
     The directory in which to save the figures. This is relative to the
     Sphinx source directory. The default is `html_static_path`.
 ipython_rgxin:
     The compiled regular expression to denote the start of IPython input
     lines. The default is re.compile('In \[(\d+)\]:\s?(.*)\s*'). You
     shouldn't need to change this.
 ipython_rgxout:
     The compiled regular expression to denote the start of IPython output
     lines. The default is re.compile('Out\[(\d+)\]:\s?(.*)\s*'). You
     shouldn't need to change this.
 ipython_promptin:
     The string to represent the IPython input prompt in the generated ReST.
     The default is 'In [%d]:'. This expects that the line numbers are used
     in the prompt.
 ipython_promptout:
 
     The string to represent the IPython prompt in the generated ReST. The
     default is 'Out [%d]:'. This expects that the line numbers are used
     in the prompt.
 
 ToDo
 ----
 
 - Turn the ad-hoc test() function into a real test suite.
 - Break up ipython-specific functionality from matplotlib stuff into better
   separated code.
 
 Authors
 -------
 
 - John D Hunter: orignal author.
 - Fernando Perez: refactoring, documentation, cleanups, port to 0.11.
 - VáclavŠmilauer <eudoxos-AT-arcig.cz>: Prompt generalizations.
 - Skipper Seabold, refactoring, cleanups, pure python addition
 """
 from __future__ import print_function
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Stdlib
-import io
 import os
 import re
 import sys
 import tempfile
 import ast
 
 # To keep compatibility with various python versions
 try:
     from hashlib import md5
 except ImportError:
     from md5 import md5
 
 # Third-party
 import matplotlib
 import sphinx
 from docutils.parsers.rst import directives
 from docutils import nodes
 from sphinx.util.compat import Directive
 
 matplotlib.use('Agg')
 
 # Our own
 from IPython import Config, InteractiveShell
 from IPython.core.profiledir import ProfileDir
 from IPython.utils import io
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 # for tokenizing blocks
 COMMENT, INPUT, OUTPUT =  range(3)
 
 #-----------------------------------------------------------------------------
 # Functions and class declarations
 #-----------------------------------------------------------------------------
 def block_parser(part, rgxin, rgxout, fmtin, fmtout):
     """
     part is a string of ipython text, comprised of at most one
     input, one ouput, comments, and blank lines.  The block parser
     parses the text into a list of::
 
       blocks = [ (TOKEN0, data0), (TOKEN1, data1), ...]
 
     where TOKEN is one of [COMMENT | INPUT | OUTPUT ] and
     data is, depending on the type of token::
 
       COMMENT : the comment string
 
       INPUT: the (DECORATOR, INPUT_LINE, REST) where
          DECORATOR: the input decorator (or None)
          INPUT_LINE: the input as string (possibly multi-line)
          REST : any stdout generated by the input line (not OUTPUT)
 
 
       OUTPUT: the output string, possibly multi-line
     """
 
     block = []
     lines = part.split('\n')
     N = len(lines)
     i = 0
     decorator = None
     while 1:
 
         if i==N:
             # nothing left to parse -- the last line
             break
 
         line = lines[i]
         i += 1
         line_stripped = line.strip()
         if line_stripped.startswith('#'):
             block.append((COMMENT, line))
             continue
 
         if line_stripped.startswith('@'):
             # we're assuming at most one decorator -- may need to
             # rethink
             decorator = line_stripped
             continue
 
         # does this look like an input line?
         matchin = rgxin.match(line)
         if matchin:
             lineno, inputline = int(matchin.group(1)), matchin.group(2)
 
             # the ....: continuation string
             continuation = '   %s:'%''.join(['.']*(len(str(lineno))+2))
             Nc = len(continuation)
             # input lines can continue on for more than one line, if
             # we have a '\' line continuation char or a function call
             # echo line 'print'.  The input line can only be
             # terminated by the end of the block or an output line, so
             # we parse out the rest of the input line if it is
             # multiline as well as any echo text
 
             rest = []
             while i<N:
 
                 # look ahead; if the next line is blank, or a comment, or
                 # an output line, we're done
 
                 nextline = lines[i]
                 matchout = rgxout.match(nextline)
                 #print "nextline=%s, continuation=%s, starts=%s"%(nextline, continuation, nextline.startswith(continuation))
                 if matchout or nextline.startswith('#'):
                     break
                 elif nextline.startswith(continuation):
                     inputline += '\n' + nextline[Nc:]
                 else:
                     rest.append(nextline)
                 i+= 1
 
             block.append((INPUT, (decorator, inputline, '\n'.join(rest))))
             continue
 
         # if it looks like an output line grab all the text to the end
         # of the block
         matchout = rgxout.match(line)
         if matchout:
             lineno, output = int(matchout.group(1)), matchout.group(2)
             if i<N-1:
                 output = '\n'.join([output] + lines[i:])
 
             block.append((OUTPUT, output))
             break
 
     return block
 
 class EmbeddedSphinxShell(object):
     """An embedded IPython instance to run inside Sphinx"""
 
     def __init__(self):
 
-        self.cout = io.StringIO()
+        self.cout = StringIO()
 
 
         # Create config object for IPython
         config = Config()
         config.Global.display_banner = False
         config.Global.exec_lines = ['import numpy as np',
                                     'from pylab import *'
                                     ]
         config.InteractiveShell.autocall = False
         config.InteractiveShell.autoindent = False
         config.InteractiveShell.colors = 'NoColor'
 
         # create a profile so instance history isn't saved
         tmp_profile_dir = tempfile.mkdtemp(prefix='profile_')
         profname = 'auto_profile_sphinx_build'
         pdir = os.path.join(tmp_profile_dir,profname)
         profile = ProfileDir.create_profile_dir(pdir)
 
         # Create and initialize ipython, but don't start its mainloop
         IP = InteractiveShell.instance(config=config, profile_dir=profile)
         # io.stdout redirect must be done *after* instantiating InteractiveShell
         io.stdout = self.cout
         io.stderr = self.cout
 
         # For debugging, so we can see normal output, use this:
         #from IPython.utils.io import Tee
         #io.stdout = Tee(self.cout, channel='stdout') # dbg
         #io.stderr = Tee(self.cout, channel='stderr') # dbg
 
         # Store a few parts of IPython we'll need.
         self.IP = IP
         self.user_ns = self.IP.user_ns
         self.user_global_ns = self.IP.user_global_ns
 
         self.input = ''
         self.output = ''
 
         self.is_verbatim = False
         self.is_doctest = False
         self.is_suppress = False
 
         # on the first call to the savefig decorator, we'll import
         # pyplot as plt so we can make a call to the plt.gcf().savefig
         self._pyplot_imported = False
 
     def clear_cout(self):
         self.cout.seek(0)
         self.cout.truncate(0)
 
     def process_input_line(self, line, store_history=True):
         """process the input, capturing stdout"""
         #print "input='%s'"%self.input
         stdout = sys.stdout
         splitter = self.IP.input_splitter
         try:
             sys.stdout = self.cout
             splitter.push(line)
             more = splitter.push_accepts_more()
             if not more:
                 source_raw = splitter.source_raw_reset()[1]
                 self.IP.run_cell(source_raw, store_history=store_history)
         finally:
             sys.stdout = stdout
 
     def process_image(self, decorator):
         """
         # build out an image directive like
         # .. image:: somefile.png
         #    :width 4in
         #
         # from an input like
         # savefig somefile.png width=4in
         """
         savefig_dir = self.savefig_dir
         source_dir = self.source_dir
         saveargs = decorator.split(' ')
         filename = saveargs[1]
         # insert relative path to image file in source
         outfile = os.path.relpath(os.path.join(savefig_dir,filename),
                     source_dir)
 
         imagerows = ['.. image:: %s'%outfile]
 
         for kwarg in saveargs[2:]:
             arg, val = kwarg.split('=')
             arg = arg.strip()
             val = val.strip()
             imagerows.append('   :%s: %s'%(arg, val))
 
         image_file = os.path.basename(outfile) # only return file name
         image_directive = '\n'.join(imagerows)
         return image_file, image_directive
 
 
     # Callbacks for each type of token
     def process_input(self, data, input_prompt, lineno):
         """Process data block for INPUT token."""
         decorator, input, rest = data
         image_file = None
         image_directive = None
         #print 'INPUT:', data  # dbg
         is_verbatim = decorator=='@verbatim' or self.is_verbatim
         is_doctest = decorator=='@doctest' or self.is_doctest
         is_suppress = decorator=='@suppress' or self.is_suppress
         is_savefig = decorator is not None and \
                      decorator.startswith('@savefig')
 
         input_lines = input.split('\n')
         if len(input_lines) > 1:
             if input_lines[-1] != "":
                 input_lines.append('') # make sure there's a blank line
                                        # so splitter buffer gets reset
 
         continuation = '   %s:'%''.join(['.']*(len(str(lineno))+2))
         Nc = len(continuation)
 
         if is_savefig:
             image_file, image_directive = self.process_image(decorator)
 
         ret = []
         is_semicolon = False
 
         for i, line in enumerate(input_lines):
             if line.endswith(';'):
                 is_semicolon = True
 
             if i==0:
                 # process the first input line
                 if is_verbatim:
                     self.process_input_line('')
                     self.IP.execution_count += 1 # increment it anyway
                 else:
                     # only submit the line in non-verbatim mode
                     self.process_input_line(line, store_history=True)
                 formatted_line = '%s %s'%(input_prompt, line)
             else:
                 # process a continuation line
                 if not is_verbatim:
                     self.process_input_line(line, store_history=True)
 
                 formatted_line = '%s %s'%(continuation, line)
 
             if not is_suppress:
                 ret.append(formatted_line)
 
         if not is_suppress and len(rest.strip()) and is_verbatim:
             # the "rest" is the standard output of the
             # input, which needs to be added in
             # verbatim mode
             ret.append(rest)
 
         self.cout.seek(0)
         output = self.cout.read()
         if not is_suppress and not is_semicolon:
             ret.append(output)
         elif is_semicolon: # get spacing right
             ret.append('')
 
         self.cout.truncate(0)
         return (ret, input_lines, output, is_doctest, image_file,
                     image_directive)
         #print 'OUTPUT', output  # dbg
 
     def process_output(self, data, output_prompt,
                        input_lines, output, is_doctest, image_file):
         """Process data block for OUTPUT token."""
         if is_doctest:
             submitted = data.strip()
             found = output
             if found is not None:
                 found = found.strip()
 
                 # XXX - fperez: in 0.11, 'output' never comes with the prompt
                 # in it, just the actual output text.  So I think all this code
                 # can be nuked...
 
                 # the above comment does not appear to be accurate... (minrk)
 
                 ind = found.find(output_prompt)
                 if ind<0:
                     e='output prompt="%s" does not match out line=%s' % \
                        (output_prompt, found)
                     raise RuntimeError(e)
                 found = found[len(output_prompt):].strip()
 
                 if found!=submitted:
                     e = ('doctest failure for input_lines="%s" with '
                          'found_output="%s" and submitted output="%s"' %
                          (input_lines, found, submitted) )
                     raise RuntimeError(e)
                 #print 'doctest PASSED for input_lines="%s" with found_output="%s" and submitted output="%s"'%(input_lines, found, submitted)
 
     def process_comment(self, data):
         """Process data fPblock for COMMENT token."""
         if not self.is_suppress:
             return [data]
 
     def save_image(self, image_file):
         """
         Saves the image file to disk.
         """
         self.ensure_pyplot()
         command = 'plt.gcf().savefig("%s")'%image_file
         #print 'SAVEFIG', command  # dbg
         self.process_input_line('bookmark ipy_thisdir', store_history=False)
         self.process_input_line('cd -b ipy_savedir', store_history=False)
         self.process_input_line(command, store_history=False)
         self.process_input_line('cd -b ipy_thisdir', store_history=False)
         self.process_input_line('bookmark -d ipy_thisdir', store_history=False)
         self.clear_cout()
 
 
     def process_block(self, block):
         """
         process block from the block_parser and return a list of processed lines
         """
         ret = []
         output = None
         input_lines = None
         lineno = self.IP.execution_count
 
         input_prompt = self.promptin%lineno
         output_prompt = self.promptout%lineno
         image_file = None
         image_directive = None
 
         for token, data in block:
             if token==COMMENT:
                 out_data = self.process_comment(data)
             elif token==INPUT:
                 (out_data, input_lines, output, is_doctest, image_file,
                     image_directive) = \
                           self.process_input(data, input_prompt, lineno)
             elif token==OUTPUT:
                 out_data = \
                     self.process_output(data, output_prompt,
                                         input_lines, output, is_doctest,
                                         image_file)
             if out_data:
                 ret.extend(out_data)
 
         # save the image files
         if image_file is not None:
             self.save_image(image_file)
 
         return ret, image_directive
 
     def ensure_pyplot(self):
         if self._pyplot_imported:
             return
         self.process_input_line('import matplotlib.pyplot as plt',
                                 store_history=False)
 
     def process_pure_python(self, content):
         """
         content is a list of strings. it is unedited directive conent
 
         This runs it line by line in the InteractiveShell, prepends
         prompts as needed capturing stderr and stdout, then returns
         the content as a list as if it were ipython code
         """
         output = []
         savefig = False # keep up with this to clear figure
         multiline = False # to handle line continuation
         multiline_start = None
         fmtin = self.promptin
 
         ct = 0
 
         for lineno, line in enumerate(content):
 
             line_stripped = line.strip()
             if not len(line):
                 output.append(line)
                 continue
 
             # handle decorators
             if line_stripped.startswith('@'):
                 output.extend([line])
                 if 'savefig' in line:
                     savefig = True # and need to clear figure
                 continue
 
             # handle comments
             if line_stripped.startswith('#'):
                 output.extend([line])
                 continue
 
             # deal with lines checking for multiline
             continuation  = u'   %s:'% ''.join(['.']*(len(str(ct))+2))
             if not multiline:
                 modified = u"%s %s" % (fmtin % ct, line_stripped)
                 output.append(modified)
                 ct += 1
                 try:
                     ast.parse(line_stripped)
                     output.append(u'')
                 except Exception: # on a multiline
                     multiline = True
                     multiline_start = lineno
             else: # still on a multiline
                 modified = u'%s %s' % (continuation, line)
                 output.append(modified)
 
                 # if the next line is indented, it should be part of multiline
                 if len(content) > lineno + 1:
                     nextline = content[lineno + 1]
                     if len(nextline) - len(nextline.lstrip()) > 3:
                         continue
                 try:
                     mod = ast.parse(
                             '\n'.join(content[multiline_start:lineno+1]))
                     if isinstance(mod.body[0], ast.FunctionDef):
                         # check to see if we have the whole function
                         for element in mod.body[0].body:
                             if isinstance(element, ast.Return):
                                 multiline = False
                     else:
                         output.append(u'')
                         multiline = False
                 except Exception:
                     pass
 
             if savefig: # clear figure if plotted
                 self.ensure_pyplot()
                 self.process_input_line('plt.clf()', store_history=False)
                 self.clear_cout()
                 savefig = False
 
         return output
 
 class IPythonDirective(Directive):
 
     has_content = True
     required_arguments = 0
     optional_arguments = 4 # python, suppress, verbatim, doctest
     final_argumuent_whitespace = True
     option_spec = { 'python': directives.unchanged,
                     'suppress' : directives.flag,
                     'verbatim' : directives.flag,
                     'doctest' : directives.flag,
                   }
 
     shell = None
 
     seen_docs = set()
 
     def get_config_options(self):
         # contains sphinx configuration variables
         config = self.state.document.settings.env.config
 
         # get config variables to set figure output directory
         confdir = self.state.document.settings.env.app.confdir
         savefig_dir = config.ipython_savefig_dir
         source_dir = os.path.dirname(self.state.document.current_source)
         if savefig_dir is None:
             savefig_dir = config.html_static_path
         if isinstance(savefig_dir, list):
             savefig_dir = savefig_dir[0] # safe to assume only one path?
         savefig_dir = os.path.join(confdir, savefig_dir)
 
         # get regex and prompt stuff
         rgxin     = config.ipython_rgxin
         rgxout    = config.ipython_rgxout
         promptin  = config.ipython_promptin
         promptout = config.ipython_promptout
 
         return savefig_dir, source_dir, rgxin, rgxout, promptin, promptout
 
     def setup(self):
         if self.shell is None:
             self.shell = EmbeddedSphinxShell()
         # reset the execution count if we haven't processed this doc
         #NOTE: this may be borked if there are multiple seen_doc tmp files
         #check time stamp?
 
         if not self.state.document.current_source in self.seen_docs:
                 self.shell.IP.history_manager.reset()
                 self.shell.IP.execution_count = 1
                 self.seen_docs.add(self.state.document.current_source)
 
 
 
         # get config values
         (savefig_dir, source_dir, rgxin,
                 rgxout, promptin, promptout) = self.get_config_options()
 
         # and attach to shell so we don't have to pass them around
         self.shell.rgxin = rgxin
         self.shell.rgxout = rgxout
         self.shell.promptin = promptin
         self.shell.promptout = promptout
         self.shell.savefig_dir = savefig_dir
         self.shell.source_dir = source_dir
 
         # setup bookmark for saving figures directory
 
         self.shell.process_input_line('bookmark ipy_savedir %s'%savefig_dir,
                                       store_history=False)
         self.shell.clear_cout()
 
         return rgxin, rgxout, promptin, promptout
 
 
     def teardown(self):
         # delete last bookmark
         self.shell.process_input_line('bookmark -d ipy_savedir',
                                       store_history=False)
         self.shell.clear_cout()
 
     def run(self):
         debug = False
 
         #TODO, any reason block_parser can't be a method of embeddable shell
         # then we wouldn't have to carry these around
         rgxin, rgxout, promptin, promptout = self.setup()
 
         options = self.options
         self.shell.is_suppress = 'suppress' in options
         self.shell.is_doctest = 'doctest' in options
         self.shell.is_verbatim = 'verbatim' in options
 
 
         # handle pure python code
         if 'python' in self.arguments:
             content = self.content
             self.content = self.shell.process_pure_python(content)
 
         parts = '\n'.join(self.content).split('\n\n')
 
         lines = ['.. code-block:: ipython','']
         figures = []
 
         for part in parts:
 
             block = block_parser(part, rgxin, rgxout, promptin, promptout)
 
             if len(block):
                 rows, figure = self.shell.process_block(block)
                 for row in rows:
                     lines.extend(['   %s'%line for line in row.split('\n')])
 
                 if figure is not None:
                     figures.append(figure)
 
         #text = '\n'.join(lines)
         #figs = '\n'.join(figures)
 
         for figure in figures:
             lines.append('')
             lines.extend(figure.split('\n'))
             lines.append('')
 
         #print lines
         if len(lines)>2:
             if debug:
                 print('\n'.join(lines))
             else: #NOTE: this raises some errors, what's it for?
                 #print 'INSERTING %d lines'%len(lines)
                 self.state_machine.insert_input(
                     lines, self.state_machine.input_lines.source(0))
 
         text = '\n'.join(lines)
         txtnode = nodes.literal_block(text, text)
         txtnode['language'] = 'ipython'
         #imgnode = nodes.image(figs)
 
         # cleanup
         self.teardown()
 
         return []#, imgnode]
 
 # Enable as a proper Sphinx directive
 def setup(app):
     setup.app = app
 
     app.add_directive('ipython', IPythonDirective)
     app.add_config_value('ipython_savefig_dir', None, True)
     app.add_config_value('ipython_rgxin',
                          re.compile('In \[(\d+)\]:\s?(.*)\s*'), True)
     app.add_config_value('ipython_rgxout',
                          re.compile('Out\[(\d+)\]:\s?(.*)\s*'), True)
     app.add_config_value('ipython_promptin', 'In [%d]:', True)
     app.add_config_value('ipython_promptout', 'Out[%d]:', True)
 
 
 # Simple smoke test, needs to be converted to a proper automatic test.
 def test():
 
     examples = [
         r"""
 In [9]: pwd
 Out[9]: '/home/jdhunter/py4science/book'
 
 In [10]: cd bookdata/
 /home/jdhunter/py4science/book/bookdata
 
 In [2]: from pylab import *
 
 In [2]: ion()
 
 In [3]: im = imread('stinkbug.png')
 
 @savefig mystinkbug.png width=4in
 In [4]: imshow(im)
 Out[4]: <matplotlib.image.AxesImage object at 0x39ea850>
 
 """,
         r"""
 
 In [1]: x = 'hello world'
 
 # string methods can be
 # used to alter the string
 @doctest
 In [2]: x.upper()
 Out[2]: 'HELLO WORLD'
 
 @verbatim
 In [3]: x.st<TAB>
 x.startswith  x.strip
 """,
     r"""
 
 In [130]: url = 'http://ichart.finance.yahoo.com/table.csv?s=CROX\
    .....: &d=9&e=22&f=2009&g=d&a=1&br=8&c=2006&ignore=.csv'
 
 In [131]: print url.split('&')
 ['http://ichart.finance.yahoo.com/table.csv?s=CROX', 'd=9', 'e=22', 'f=2009', 'g=d', 'a=1', 'b=8', 'c=2006', 'ignore=.csv']
 
 In [60]: import urllib
 
 """,
     r"""\
 
 In [133]: import numpy.random
 
 @suppress
 In [134]: numpy.random.seed(2358)
 
 @doctest
 In [135]: numpy.random.rand(10,2)
 Out[135]:
 array([[ 0.64524308,  0.59943846],
        [ 0.47102322,  0.8715456 ],
        [ 0.29370834,  0.74776844],
        [ 0.99539577,  0.1313423 ],
        [ 0.16250302,  0.21103583],
        [ 0.81626524,  0.1312433 ],
        [ 0.67338089,  0.72302393],
        [ 0.7566368 ,  0.07033696],
        [ 0.22591016,  0.77731835],
        [ 0.0072729 ,  0.34273127]])
 
 """,
 
     r"""
 In [106]: print x
 jdh
 
 In [109]: for i in range(10):
    .....:     print i
    .....:
    .....:
 0
 1
 2
 3
 4
 5
 6
 7
 8
 9
 """,
 
         r"""
 
 In [144]: from pylab import *
 
 In [145]: ion()
 
 # use a semicolon to suppress the output
 @savefig test_hist.png width=4in
 In [151]: hist(np.random.randn(10000), 100);
 
 
 @savefig test_plot.png width=4in
 In [151]: plot(np.random.randn(10000), 'o');
    """,
 
         r"""
 # use a semicolon to suppress the output
 In [151]: plt.clf()
 
 @savefig plot_simple.png width=4in
 In [151]: plot([1,2,3])
 
 @savefig hist_simple.png width=4in
 In [151]: hist(np.random.randn(10000), 100);
 
 """,
      r"""
 # update the current fig
 In [151]: ylabel('number')
 
 In [152]: title('normal distribution')
 
 
 @savefig hist_with_text.png
 In [153]: grid(True)
 
         """,
         ]
     # skip local-file depending first example:
     examples = examples[1:]
 
     #ipython_directive.DEBUG = True  # dbg
     #options = dict(suppress=True)  # dbg
     options = dict()
     for example in examples:
         content = example.split('\n')
         IPythonDirective('debug', arguments=None, options=options,
                           content=content, lineno=0,
                           content_offset=None, block_text=None,
                           state=None, state_machine=None,
                           )
 
 # Run test suite as a script
 if __name__=='__main__':
     if not os.path.isdir('_static'):
         os.mkdir('_static')
     test()
     print('All OK? Check figures in _static/')

IPython/testing/plugin/ipdoctest.py

@@ -1,760 +1,764
 """Nose Plugin that supports IPython doctests.
 
 Limitations:
 
 - When generating examples for use as doctests, make sure that you have
   pretty-printing OFF.  This can be done either by setting the
   ``PlainTextFormatter.pprint`` option in your configuration file to  False, or
   by interactively disabling it with  %Pprint.  This is required so that IPython
   output matches that of normal Python, which is used by doctest for internal
   execution.
 
 - Do not rely on specific prompt numbers for results (such as using
   '_34==True', for example).  For IPython tests run via an external process the
   prompt numbers may be different, and IPython tests run as normal python code
   won't even have these special _NN variables set at all.
 """
 
 #-----------------------------------------------------------------------------
 # Module imports
 
 # From the standard library
 import doctest
 import inspect
 import logging
 import os
 import re
 import sys
 import traceback
 import unittest
 
 from inspect import getmodule
-from io import StringIO
 
 # We are overriding the default doctest runner, so we need to import a few
 # things from doctest directly
 from doctest import (REPORTING_FLAGS, REPORT_ONLY_FIRST_FAILURE,
                      _unittest_reportflags, DocTestRunner,
                      _extract_future_flags, pdb, _OutputRedirectingPdb,
                      _exception_traceback,
                      linecache)
 
 # Third-party modules
 import nose.core
 
 from nose.plugins import doctests, Plugin
 from nose.util import anyp, getpackage, test_address, resolve_name, tolist
 
 # Our own imports
-from IPython.utils.py3compat import builtin_mod
+from IPython.utils.py3compat import builtin_mod, PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Module globals and other constants
 #-----------------------------------------------------------------------------
 
 log = logging.getLogger(__name__)
 
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 def is_extension_module(filename):
     """Return whether the given filename is an extension module.
 
     This simply checks that the extension is either .so or .pyd.
     """
     return os.path.splitext(filename)[1].lower() in ('.so','.pyd')
 
 
 class DocTestSkip(object):
     """Object wrapper for doctests to be skipped."""
 
     ds_skip = """Doctest to skip.
     >>> 1 #doctest: +SKIP
     """
 
     def __init__(self,obj):
         self.obj = obj
 
     def __getattribute__(self,key):
         if key == '__doc__':
             return DocTestSkip.ds_skip
         else:
             return getattr(object.__getattribute__(self,'obj'),key)
 
 # Modified version of the one in the stdlib, that fixes a python bug (doctests
 # not found in extension modules, http://bugs.python.org/issue3158)
 class DocTestFinder(doctest.DocTestFinder):
 
     def _from_module(self, module, object):
         """
         Return true if the given object is defined in the given
         module.
         """
         if module is None:
             return True
         elif inspect.isfunction(object):
             return module.__dict__ is object.__globals__
         elif inspect.isbuiltin(object):
             return module.__name__ == object.__module__
         elif inspect.isclass(object):
             return module.__name__ == object.__module__
         elif inspect.ismethod(object):
             # This one may be a bug in cython that fails to correctly set the
             # __module__ attribute of methods, but since the same error is easy
             # to make by extension code writers, having this safety in place
             # isn't such a bad idea
             return module.__name__ == object.im_class.__module__
         elif inspect.getmodule(object) is not None:
             return module is inspect.getmodule(object)
         elif hasattr(object, '__module__'):
             return module.__name__ == object.__module__
         elif isinstance(object, property):
             return True # [XX] no way not be sure.
         else:
             raise ValueError("object must be a class or function")
 
     def _find(self, tests, obj, name, module, source_lines, globs, seen):
         """
         Find tests for the given object and any contained objects, and
         add them to `tests`.
         """
         #print '_find for:', obj, name, module  # dbg
         if hasattr(obj,"skip_doctest"):
             #print 'SKIPPING DOCTEST FOR:',obj  # dbg
             obj = DocTestSkip(obj)
 
         doctest.DocTestFinder._find(self,tests, obj, name, module,
                                     source_lines, globs, seen)
 
         # Below we re-run pieces of the above method with manual modifications,
         # because the original code is buggy and fails to correctly identify
         # doctests in extension modules.
 
         # Local shorthands
         from inspect import isroutine, isclass, ismodule
 
         # Look for tests in a module's contained objects.
         if inspect.ismodule(obj) and self._recurse:
             for valname, val in obj.__dict__.items():
                 valname1 = '%s.%s' % (name, valname)
                 if ( (isroutine(val) or isclass(val))
                      and self._from_module(module, val) ):
 
                     self._find(tests, val, valname1, module, source_lines,
                                globs, seen)
 
         # Look for tests in a class's contained objects.
         if inspect.isclass(obj) and self._recurse:
             #print 'RECURSE into class:',obj  # dbg
             for valname, val in obj.__dict__.items():
                 # Special handling for staticmethod/classmethod.
                 if isinstance(val, staticmethod):
                     val = getattr(obj, valname)
                 if isinstance(val, classmethod):
                     val = getattr(obj, valname).im_func
 
                 # Recurse to methods, properties, and nested classes.
                 if ((inspect.isfunction(val) or inspect.isclass(val) or
                      inspect.ismethod(val) or
                       isinstance(val, property)) and
                       self._from_module(module, val)):
                     valname = '%s.%s' % (name, valname)
                     self._find(tests, val, valname, module, source_lines,
                                globs, seen)
 
 
 class IPDoctestOutputChecker(doctest.OutputChecker):
     """Second-chance checker with support for random tests.
 
     If the default comparison doesn't pass, this checker looks in the expected
     output string for flags that tell us to ignore the output.
     """
 
     random_re = re.compile(r'#\s*random\s+')
 
     def check_output(self, want, got, optionflags):
         """Check output, accepting special markers embedded in the output.
 
         If the output didn't pass the default validation but the special string
         '#random' is included, we accept it."""
 
         # Let the original tester verify first, in case people have valid tests
         # that happen to have a comment saying '#random' embedded in.
         ret = doctest.OutputChecker.check_output(self, want, got,
                                                  optionflags)
         if not ret and self.random_re.search(want):
             #print >> sys.stderr, 'RANDOM OK:',want  # dbg
             return True
 
         return ret
 
 
 class DocTestCase(doctests.DocTestCase):
     """Proxy for DocTestCase: provides an address() method that
     returns the correct address for the doctest case. Otherwise
     acts as a proxy to the test case. To provide hints for address(),
     an obj may also be passed -- this will be used as the test object
     for purposes of determining the test address, if it is provided.
     """
 
     # Note: this method was taken from numpy's nosetester module.
 
     # Subclass nose.plugins.doctests.DocTestCase to work around a bug in
     # its constructor that blocks non-default arguments from being passed
     # down into doctest.DocTestCase
 
     def __init__(self, test, optionflags=0, setUp=None, tearDown=None,
                  checker=None, obj=None, result_var='_'):
         self._result_var = result_var
         doctests.DocTestCase.__init__(self, test,
                                       optionflags=optionflags,
                                       setUp=setUp, tearDown=tearDown,
                                       checker=checker)
         # Now we must actually copy the original constructor from the stdlib
         # doctest class, because we can't call it directly and a bug in nose
         # means it never gets passed the right arguments.
 
         self._dt_optionflags = optionflags
         self._dt_checker = checker
         self._dt_test = test
         self._dt_test_globs_ori = test.globs
         self._dt_setUp = setUp
         self._dt_tearDown = tearDown
 
         # XXX - store this runner once in the object!
         runner = IPDocTestRunner(optionflags=optionflags,
                                  checker=checker, verbose=False)
         self._dt_runner = runner
 
 
         # Each doctest should remember the directory it was loaded from, so
         # things like %run work without too many contortions
         self._ori_dir = os.path.dirname(test.filename)
 
     # Modified runTest from the default stdlib
     def runTest(self):
         test = self._dt_test
         runner = self._dt_runner
 
         old = sys.stdout
         new = StringIO()
         optionflags = self._dt_optionflags
 
         if not (optionflags & REPORTING_FLAGS):
             # The option flags don't include any reporting flags,
             # so add the default reporting flags
             optionflags |= _unittest_reportflags
 
         try:
             # Save our current directory and switch out to the one where the
             # test was originally created, in case another doctest did a
             # directory change.  We'll restore this in the finally clause.
             curdir = os.getcwdu()
             #print 'runTest in dir:', self._ori_dir  # dbg
             os.chdir(self._ori_dir)
 
             runner.DIVIDER = "-"*70
             failures, tries = runner.run(test,out=new.write,
                                          clear_globs=False)
         finally:
             sys.stdout = old
             os.chdir(curdir)
 
         if failures:
             raise self.failureException(self.format_failure(new.getvalue()))
 
     def setUp(self):
         """Modified test setup that syncs with ipython namespace"""
         #print "setUp test", self._dt_test.examples # dbg
         if isinstance(self._dt_test.examples[0], IPExample):
             # for IPython examples *only*, we swap the globals with the ipython
             # namespace, after updating it with the globals (which doctest
             # fills with the necessary info from the module being tested).
             self.user_ns_orig = {}
             self.user_ns_orig.update(_ip.user_ns)
             _ip.user_ns.update(self._dt_test.globs)
             # We must remove the _ key in the namespace, so that Python's
             # doctest code sets it naturally
             _ip.user_ns.pop('_', None)
             _ip.user_ns['__builtins__'] = builtin_mod
             self._dt_test.globs = _ip.user_ns
 
         super(DocTestCase, self).setUp()
 
     def tearDown(self):
 
         # Undo the test.globs reassignment we made, so that the parent class
         # teardown doesn't destroy the ipython namespace
         if isinstance(self._dt_test.examples[0], IPExample):
             self._dt_test.globs = self._dt_test_globs_ori
             _ip.user_ns.clear()
             _ip.user_ns.update(self.user_ns_orig)
 
         # XXX - fperez: I am not sure if this is truly a bug in nose 0.11, but
         # it does look like one to me: its tearDown method tries to run
         #
         # delattr(builtin_mod, self._result_var)
         #
         # without checking that the attribute really is there; it implicitly
         # assumes it should have been set via displayhook.  But if the
         # displayhook was never called, this doesn't necessarily happen.  I
         # haven't been able to find a little self-contained example outside of
         # ipython that would show the problem so I can report it to the nose
         # team, but it does happen a lot in our code.
         #
         # So here, we just protect as narrowly as possible by trapping an
         # attribute error whose message would be the name of self._result_var,
         # and letting any other error propagate.
         try:
             super(DocTestCase, self).tearDown()
         except AttributeError as exc:
             if exc.args[0] != self._result_var:
                 raise
 
 
 # A simple subclassing of the original with a different class name, so we can
 # distinguish and treat differently IPython examples from pure python ones.
 class IPExample(doctest.Example): pass
 
 
 class IPExternalExample(doctest.Example):
     """Doctest examples to be run in an external process."""
 
     def __init__(self, source, want, exc_msg=None, lineno=0, indent=0,
                  options=None):
         # Parent constructor
         doctest.Example.__init__(self,source,want,exc_msg,lineno,indent,options)
 
         # An EXTRA newline is needed to prevent pexpect hangs
         self.source += '\n'
 
 
 class IPDocTestParser(doctest.DocTestParser):
     """
     A class used to parse strings containing doctest examples.
 
     Note: This is a version modified to properly recognize IPython input and
     convert any IPython examples into valid Python ones.
     """
     # This regular expression is used to find doctest examples in a
     # string.  It defines three groups: `source` is the source code
     # (including leading indentation and prompts); `indent` is the
     # indentation of the first (PS1) line of the source code; and
     # `want` is the expected output (including leading indentation).
 
     # Classic Python prompts or default IPython ones
     _PS1_PY = r'>>>'
     _PS2_PY = r'\.\.\.'
 
     _PS1_IP = r'In\ \[\d+\]:'
     _PS2_IP = r'\ \ \ \.\.\.+:'
 
     _RE_TPL = r'''
         # Source consists of a PS1 line followed by zero or more PS2 lines.
         (?P<source>
             (?:^(?P<indent> [ ]*) (?P<ps1> %s) .*)    # PS1 line
             (?:\n           [ ]*  (?P<ps2> %s) .*)*)  # PS2 lines
         \n? # a newline
         # Want consists of any non-blank lines that do not start with PS1.
         (?P<want> (?:(?![ ]*$)    # Not a blank line
                      (?![ ]*%s)   # Not a line starting with PS1
                      (?![ ]*%s)   # Not a line starting with PS2
                      .*$\n?       # But any other line
                   )*)
                   '''
 
     _EXAMPLE_RE_PY = re.compile( _RE_TPL % (_PS1_PY,_PS2_PY,_PS1_PY,_PS2_PY),
                                  re.MULTILINE | re.VERBOSE)
 
     _EXAMPLE_RE_IP = re.compile( _RE_TPL % (_PS1_IP,_PS2_IP,_PS1_IP,_PS2_IP),
                                  re.MULTILINE | re.VERBOSE)
 
     # Mark a test as being fully random.  In this case, we simply append the
     # random marker ('#random') to each individual example's output.  This way
     # we don't need to modify any other code.
     _RANDOM_TEST = re.compile(r'#\s*all-random\s+')
 
     # Mark tests to be executed in an external process - currently unsupported.
     _EXTERNAL_IP = re.compile(r'#\s*ipdoctest:\s*EXTERNAL')
 
     def ip2py(self,source):
         """Convert input IPython source into valid Python."""
         block = _ip.input_transformer_manager.transform_cell(source)
         if len(block.splitlines()) == 1:
             return _ip.prefilter(block)
         else:
             return block
 
     def parse(self, string, name='<string>'):
         """
         Divide the given string into examples and intervening text,
         and return them as a list of alternating Examples and strings.
         Line numbers for the Examples are 0-based.  The optional
         argument `name` is a name identifying this string, and is only
         used for error messages.
         """
 
         #print 'Parse string:\n',string # dbg
 
         string = string.expandtabs()
         # If all lines begin with the same indentation, then strip it.
         min_indent = self._min_indent(string)
         if min_indent > 0:
             string = '\n'.join([l[min_indent:] for l in string.split('\n')])
 
         output = []
         charno, lineno = 0, 0
 
         # We make 'all random' tests by adding the '# random' mark to every
         # block of output in the test.
         if self._RANDOM_TEST.search(string):
             random_marker = '\n# random'
         else:
             random_marker = ''
 
         # Whether to convert the input from ipython to python syntax
         ip2py = False
         # Find all doctest examples in the string.  First, try them as Python
         # examples, then as IPython ones
         terms = list(self._EXAMPLE_RE_PY.finditer(string))
         if terms:
             # Normal Python example
             #print '-'*70  # dbg
             #print 'PyExample, Source:\n',string  # dbg
             #print '-'*70  # dbg
             Example = doctest.Example
         else:
             # It's an ipython example.  Note that IPExamples are run
             # in-process, so their syntax must be turned into valid python.
             # IPExternalExamples are run out-of-process (via pexpect) so they
             # don't need any filtering (a real ipython will be executing them).
             terms = list(self._EXAMPLE_RE_IP.finditer(string))
             if self._EXTERNAL_IP.search(string):
                 #print '-'*70  # dbg
                 #print 'IPExternalExample, Source:\n',string  # dbg
                 #print '-'*70  # dbg
                 Example = IPExternalExample
             else:
                 #print '-'*70  # dbg
                 #print 'IPExample, Source:\n',string  # dbg
                 #print '-'*70  # dbg
                 Example = IPExample
                 ip2py = True
 
         for m in terms:
             # Add the pre-example text to `output`.
             output.append(string[charno:m.start()])
             # Update lineno (lines before this example)
             lineno += string.count('\n', charno, m.start())
             # Extract info from the regexp match.
             (source, options, want, exc_msg) = \
                      self._parse_example(m, name, lineno,ip2py)
 
             # Append the random-output marker (it defaults to empty in most
             # cases, it's only non-empty for 'all-random' tests):
             want += random_marker
 
             if Example is IPExternalExample:
                 options[doctest.NORMALIZE_WHITESPACE] = True
                 want += '\n'
 
             # Create an Example, and add it to the list.
             if not self._IS_BLANK_OR_COMMENT(source):
                 output.append(Example(source, want, exc_msg,
                                       lineno=lineno,
                                       indent=min_indent+len(m.group('indent')),
                                       options=options))
             # Update lineno (lines inside this example)
             lineno += string.count('\n', m.start(), m.end())
             # Update charno.
             charno = m.end()
         # Add any remaining post-example text to `output`.
         output.append(string[charno:])
         return output
 
     def _parse_example(self, m, name, lineno,ip2py=False):
         """
         Given a regular expression match from `_EXAMPLE_RE` (`m`),
         return a pair `(source, want)`, where `source` is the matched
         example's source code (with prompts and indentation stripped);
         and `want` is the example's expected output (with indentation
         stripped).
 
         `name` is the string's name, and `lineno` is the line number
         where the example starts; both are used for error messages.
 
         Optional:
         `ip2py`: if true, filter the input via IPython to convert the syntax
         into valid python.
         """
 
         # Get the example's indentation level.
         indent = len(m.group('indent'))
 
         # Divide source into lines; check that they're properly
         # indented; and then strip their indentation & prompts.
         source_lines = m.group('source').split('\n')
 
         # We're using variable-length input prompts
         ps1 = m.group('ps1')
         ps2 = m.group('ps2')
         ps1_len = len(ps1)
 
         self._check_prompt_blank(source_lines, indent, name, lineno,ps1_len)
         if ps2:
             self._check_prefix(source_lines[1:], ' '*indent + ps2, name, lineno)
 
         source = '\n'.join([sl[indent+ps1_len+1:] for sl in source_lines])
 
         if ip2py:
             # Convert source input from IPython into valid Python syntax
             source = self.ip2py(source)
 
         # Divide want into lines; check that it's properly indented; and
         # then strip the indentation.  Spaces before the last newline should
         # be preserved, so plain rstrip() isn't good enough.
         want = m.group('want')
         want_lines = want.split('\n')
         if len(want_lines) > 1 and re.match(r' *$', want_lines[-1]):
             del want_lines[-1]  # forget final newline & spaces after it
         self._check_prefix(want_lines, ' '*indent, name,
                            lineno + len(source_lines))
 
         # Remove ipython output prompt that might be present in the first line
         want_lines[0] = re.sub(r'Out\[\d+\]: \s*?\n?','',want_lines[0])
 
         want = '\n'.join([wl[indent:] for wl in want_lines])
 
         # If `want` contains a traceback message, then extract it.
         m = self._EXCEPTION_RE.match(want)
         if m:
             exc_msg = m.group('msg')
         else:
             exc_msg = None
 
         # Extract options from the source.
         options = self._find_options(source, name, lineno)
 
         return source, options, want, exc_msg
 
     def _check_prompt_blank(self, lines, indent, name, lineno, ps1_len):
         """
         Given the lines of a source string (including prompts and
         leading indentation), check to make sure that every prompt is
         followed by a space character.  If any line is not followed by
         a space character, then raise ValueError.
 
         Note: IPython-modified version which takes the input prompt length as a
         parameter, so that prompts of variable length can be dealt with.
         """
         space_idx = indent+ps1_len
         min_len = space_idx+1
         for i, line in enumerate(lines):
             if len(line) >=  min_len and line[space_idx] != ' ':
                 raise ValueError('line %r of the docstring for %s '
                                  'lacks blank after %s: %r' %
                                  (lineno+i+1, name,
                                   line[indent:space_idx], line))
 
 
 SKIP = doctest.register_optionflag('SKIP')
 
 
 class IPDocTestRunner(doctest.DocTestRunner,object):
     """Test runner that synchronizes the IPython namespace with test globals.
     """
 
     def run(self, test, compileflags=None, out=None, clear_globs=True):
 
         # Hack: ipython needs access to the execution context of the example,
         # so that it can propagate user variables loaded by %run into
         # test.globs.  We put them here into our modified %run as a function
         # attribute.  Our new %run will then only make the namespace update
         # when called (rather than unconconditionally updating test.globs here
         # for all examples, most of which won't be calling %run anyway).
         #_ip._ipdoctest_test_globs = test.globs
         #_ip._ipdoctest_test_filename = test.filename
 
         test.globs.update(_ip.user_ns)
 
         return super(IPDocTestRunner,self).run(test,
                                                compileflags,out,clear_globs)
 
 
 class DocFileCase(doctest.DocFileCase):
     """Overrides to provide filename
     """
     def address(self):
         return (self._dt_test.filename, None, None)
 
 
 class ExtensionDoctest(doctests.Doctest):
     """Nose Plugin that supports doctests in extension modules.
     """
     name = 'extdoctest'   # call nosetests with --with-extdoctest
     enabled = True
 
     def options(self, parser, env=os.environ):
         Plugin.options(self, parser, env)
         parser.add_option('--doctest-tests', action='store_true',
                           dest='doctest_tests',
                           default=env.get('NOSE_DOCTEST_TESTS',True),
                           help="Also look for doctests in test modules. "
                           "Note that classes, methods and functions should "
                           "have either doctests or non-doctest tests, "
                           "not both. [NOSE_DOCTEST_TESTS]")
         parser.add_option('--doctest-extension', action="append",
                           dest="doctestExtension",
                           help="Also look for doctests in files with "
                           "this extension [NOSE_DOCTEST_EXTENSION]")
         # Set the default as a list, if given in env; otherwise
         # an additional value set on the command line will cause
         # an error.
         env_setting = env.get('NOSE_DOCTEST_EXTENSION')
         if env_setting is not None:
             parser.set_defaults(doctestExtension=tolist(env_setting))
 
 
     def configure(self, options, config):
         Plugin.configure(self, options, config)
         # Pull standard doctest plugin out of config; we will do doctesting
         config.plugins.plugins = [p for p in config.plugins.plugins
                                   if p.name != 'doctest']
         self.doctest_tests = options.doctest_tests
         self.extension = tolist(options.doctestExtension)
 
         self.parser = doctest.DocTestParser()
         self.finder = DocTestFinder()
         self.checker = IPDoctestOutputChecker()
         self.globs = None
         self.extraglobs = None
 
 
     def loadTestsFromExtensionModule(self,filename):
         bpath,mod = os.path.split(filename)
         modname = os.path.splitext(mod)[0]
         try:
             sys.path.append(bpath)
             module = __import__(modname)
             tests = list(self.loadTestsFromModule(module))
         finally:
             sys.path.pop()
         return tests
 
     # NOTE: the method below is almost a copy of the original one in nose, with
     # a  few modifications to control output checking.
 
     def loadTestsFromModule(self, module):
         #print '*** ipdoctest - lTM',module  # dbg
 
         if not self.matches(module.__name__):
             log.debug("Doctest doesn't want module %s", module)
             return
 
         tests = self.finder.find(module,globs=self.globs,
                                  extraglobs=self.extraglobs)
         if not tests:
             return
 
         # always use whitespace and ellipsis options
         optionflags = doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS
 
         tests.sort()
         module_file = module.__file__
         if module_file[-4:] in ('.pyc', '.pyo'):
             module_file = module_file[:-1]
         for test in tests:
             if not test.examples:
                 continue
             if not test.filename:
                 test.filename = module_file
 
             yield DocTestCase(test,
                               optionflags=optionflags,
                               checker=self.checker)
 
 
     def loadTestsFromFile(self, filename):
         #print "ipdoctest - from file", filename # dbg
         if is_extension_module(filename):
             for t in self.loadTestsFromExtensionModule(filename):
                 yield t
         else:
             if self.extension and anyp(filename.endswith, self.extension):
                 name = os.path.basename(filename)
                 dh = open(filename)
                 try:
                     doc = dh.read()
                 finally:
                     dh.close()
                 test = self.parser.get_doctest(
                     doc, globs={'__file__': filename}, name=name,
                     filename=filename, lineno=0)
                 if test.examples:
                     #print 'FileCase:',test.examples  # dbg
                     yield DocFileCase(test)
                 else:
                     yield False # no tests to load
 
 
 class IPythonDoctest(ExtensionDoctest):
     """Nose Plugin that supports doctests in extension modules.
     """
     name = 'ipdoctest'   # call nosetests with --with-ipdoctest
     enabled = True
 
     def makeTest(self, obj, parent):
         """Look for doctests in the given object, which will be a
         function, method or class.
         """
         #print 'Plugin analyzing:', obj, parent  # dbg
         # always use whitespace and ellipsis options
         optionflags = doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS
 
         doctests = self.finder.find(obj, module=getmodule(parent))
         if doctests:
             for test in doctests:
                 if len(test.examples) == 0:
                     continue
 
                 yield DocTestCase(test, obj=obj,
                                   optionflags=optionflags,
                                   checker=self.checker)
 
     def options(self, parser, env=os.environ):
         #print "Options for nose plugin:", self.name # dbg
         Plugin.options(self, parser, env)
         parser.add_option('--ipdoctest-tests', action='store_true',
                           dest='ipdoctest_tests',
                           default=env.get('NOSE_IPDOCTEST_TESTS',True),
                           help="Also look for doctests in test modules. "
                           "Note that classes, methods and functions should "
                           "have either doctests or non-doctest tests, "
                           "not both. [NOSE_IPDOCTEST_TESTS]")
         parser.add_option('--ipdoctest-extension', action="append",
                           dest="ipdoctest_extension",
                           help="Also look for doctests in files with "
                           "this extension [NOSE_IPDOCTEST_EXTENSION]")
         # Set the default as a list, if given in env; otherwise
         # an additional value set on the command line will cause
         # an error.
         env_setting = env.get('NOSE_IPDOCTEST_EXTENSION')
         if env_setting is not None:
             parser.set_defaults(ipdoctest_extension=tolist(env_setting))
 
     def configure(self, options, config):
         #print "Configuring nose plugin:", self.name # dbg
         Plugin.configure(self, options, config)
         # Pull standard doctest plugin out of config; we will do doctesting
         config.plugins.plugins = [p for p in config.plugins.plugins
                                   if p.name != 'doctest']
         self.doctest_tests = options.ipdoctest_tests
         self.extension = tolist(options.ipdoctest_extension)
 
         self.parser = IPDocTestParser()
         self.finder = DocTestFinder(parser=self.parser)
         self.checker = IPDoctestOutputChecker()
         self.globs = None
         self.extraglobs = None

IPython/utils/PyColorize.py

@@ -1,310 +1,315
 # -*- coding: utf-8 -*-
 """
 Class and program to colorize python source code for ANSI terminals.
 
 Based on an HTML code highlighter by Jurgen Hermann found at:
 http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/52298
 
 Modifications by Fernando Perez ([e-mail]).
 
 Information on the original HTML highlighter follows:
 
 MoinMoin - Python Source Parser
 
 Title: Colorize Python source using the built-in tokenizer
 
 Submitter: Jurgen Hermann
 Last Updated:2001/04/06
 
 Version no:1.2
 
 Description:
 
 This code is part of MoinMoin (http://moin.sourceforge.net/) and converts
 Python source code to HTML markup, rendering comments, keywords,
 operators, numeric and string literals in different colors.
 
 It shows how to use the built-in keyword, token and tokenize modules to
 scan Python source code and re-emit it with no changes to its original
 formatting (which is the hard part).
 """
 from __future__ import print_function
 from __future__ import absolute_import
 from __future__ import unicode_literals
 
 __all__ = ['ANSICodeColors','Parser']
 
 _scheme_default = 'Linux'
 
 
 # Imports
-import io
 import keyword
 import os
 import sys
 import token
 import tokenize
 
 try:
     generate_tokens = tokenize.generate_tokens
 except AttributeError:
     # Python 3. Note that we use the undocumented _tokenize because it expects
     # strings, not bytes. See also Python issue #9969.
     generate_tokens = tokenize._tokenize
 
 from IPython.utils.coloransi import *
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #############################################################################
 ### Python Source Parser (does Hilighting)
 #############################################################################
 
 _KEYWORD = token.NT_OFFSET + 1
 _TEXT    = token.NT_OFFSET + 2
 
 #****************************************************************************
 # Builtin color schemes
 
 Colors = TermColors  # just a shorthand
 
 # Build a few color schemes
 NoColor = ColorScheme(
     'NoColor',{
     token.NUMBER     : Colors.NoColor,
     token.OP         : Colors.NoColor,
     token.STRING     : Colors.NoColor,
     tokenize.COMMENT : Colors.NoColor,
     token.NAME       : Colors.NoColor,
     token.ERRORTOKEN : Colors.NoColor,
 
     _KEYWORD         : Colors.NoColor,
     _TEXT            : Colors.NoColor,
 
     'normal'         : Colors.NoColor  # color off (usu. Colors.Normal)
     }  )
 
 LinuxColors = ColorScheme(
     'Linux',{
     token.NUMBER     : Colors.LightCyan,
     token.OP         : Colors.Yellow,
     token.STRING     : Colors.LightBlue,
     tokenize.COMMENT : Colors.LightRed,
     token.NAME       : Colors.Normal,
     token.ERRORTOKEN : Colors.Red,
 
     _KEYWORD         : Colors.LightGreen,
     _TEXT            : Colors.Yellow,
 
     'normal'         : Colors.Normal  # color off (usu. Colors.Normal)
     } )
 
 LightBGColors = ColorScheme(
     'LightBG',{
     token.NUMBER     : Colors.Cyan,
     token.OP         : Colors.Blue,
     token.STRING     : Colors.Blue,
     tokenize.COMMENT : Colors.Red,
     token.NAME       : Colors.Normal,
     token.ERRORTOKEN : Colors.Red,
 
     _KEYWORD         : Colors.Green,
     _TEXT            : Colors.Blue,
 
     'normal'         : Colors.Normal  # color off (usu. Colors.Normal)
     }  )
 
 # Build table of color schemes (needed by the parser)
 ANSICodeColors = ColorSchemeTable([NoColor,LinuxColors,LightBGColors],
                                   _scheme_default)
 
 class Parser:
     """ Format colored Python source.
     """
 
     def __init__(self, color_table=None,out = sys.stdout):
         """ Create a parser with a specified color table and output channel.
 
         Call format() to process code.
         """
         self.color_table = color_table and color_table or ANSICodeColors
         self.out = out
 
     def format(self, raw, out = None, scheme = ''):
         return self.format2(raw, out, scheme)[0]
 
     def format2(self, raw, out = None, scheme = ''):
         """ Parse and send the colored source.
 
         If out and scheme are not specified, the defaults (given to
         constructor) are used.
 
         out should be a file-type object. Optionally, out can be given as the
         string 'str' and the parser will automatically return the output in a
         string."""
 
         string_output = 0
         if out == 'str' or self.out == 'str' or \
-           isinstance(self.out,io.StringIO):
+           isinstance(self.out,StringIO):
             # XXX - I don't really like this state handling logic, but at this
             # point I don't want to make major changes, so adding the
             # isinstance() check is the simplest I can do to ensure correct
             # behavior.
             out_old = self.out
-            self.out = io.StringIO()
+            self.out = StringIO()
             string_output = 1
         elif out is not None:
             self.out = out
 
         # Fast return of the unmodified input for NoColor scheme
         if scheme == 'NoColor':
             error = False
             self.out.write(raw)
             if string_output:
                 return raw,error
             else:
                 return None,error
 
         # local shorthands
         colors = self.color_table[scheme].colors
         self.colors = colors # put in object so __call__ sees it
 
         # Remove trailing whitespace and normalize tabs
         self.raw = raw.expandtabs().rstrip()
 
         # store line offsets in self.lines
         self.lines = [0, 0]
         pos = 0
         raw_find = self.raw.find
         lines_append = self.lines.append
         while 1:
             pos = raw_find('\n', pos) + 1
             if not pos: break
             lines_append(pos)
         lines_append(len(self.raw))
 
         # parse the source and write it
         self.pos = 0
-        text = io.StringIO(self.raw)
+        text = StringIO(self.raw)
 
         error = False
         try:
             for atoken in generate_tokens(text.readline):
                 self(*atoken)
         except tokenize.TokenError as ex:
             msg = ex.args[0]
             line = ex.args[1][0]
             self.out.write("%s\n\n*** ERROR: %s%s%s\n" %
                            (colors[token.ERRORTOKEN],
                             msg, self.raw[self.lines[line]:],
                             colors.normal)
                            )
             error = True
         self.out.write(colors.normal+'\n')
         if string_output:
             output = self.out.getvalue()
             self.out = out_old
             return (output, error)
         return (None, error)
 
     def __call__(self, toktype, toktext, start_pos, end_pos, line):
         """ Token handler, with syntax highlighting."""
         (srow,scol) = start_pos
         (erow,ecol) = end_pos
         colors = self.colors
         owrite = self.out.write
 
         # line separator, so this works across platforms
         linesep = os.linesep
 
         # calculate new positions
         oldpos = self.pos
         newpos = self.lines[srow] + scol
         self.pos = newpos + len(toktext)
 
         # send the original whitespace, if needed
         if newpos > oldpos:
             owrite(self.raw[oldpos:newpos])
 
         # skip indenting tokens
         if toktype in [token.INDENT, token.DEDENT]:
             self.pos = newpos
             return
 
         # map token type to a color group
         if token.LPAR <= toktype and toktype <= token.OP:
             toktype = token.OP
         elif toktype == token.NAME and keyword.iskeyword(toktext):
             toktype = _KEYWORD
         color = colors.get(toktype, colors[_TEXT])
 
         #print '<%s>' % toktext,    # dbg
 
         # Triple quoted strings must be handled carefully so that backtracking
         # in pagers works correctly. We need color terminators on _each_ line.
         if linesep in toktext:
             toktext = toktext.replace(linesep, '%s%s%s' %
                                       (colors.normal,linesep,color))
 
         # send text
         owrite('%s%s%s' % (color,toktext,colors.normal))
 
 def main(argv=None):
     """Run as a command-line script: colorize a python file or stdin using ANSI
     color escapes and print to stdout.
 
     Inputs:
 
       - argv(None): a list of strings like sys.argv[1:] giving the command-line
         arguments. If None, use sys.argv[1:].
     """
 
     usage_msg = """%prog [options] [filename]
 
 Colorize a python file or stdin using ANSI color escapes and print to stdout.
 If no filename is given, or if filename is -, read standard input."""
 
     import optparse
     parser = optparse.OptionParser(usage=usage_msg)
     newopt = parser.add_option
     newopt('-s','--scheme',metavar='NAME',dest='scheme_name',action='store',
            choices=['Linux','LightBG','NoColor'],default=_scheme_default,
            help="give the color scheme to use. Currently only 'Linux'\
  (default) and 'LightBG' and 'NoColor' are implemented (give without\
  quotes)")
 
     opts,args = parser.parse_args(argv)
 
     if len(args) > 1:
         parser.error("you must give at most one filename.")
 
     if len(args) == 0:
         fname = '-' # no filename given; setup to read from stdin
     else:
         fname = args[0]
 
     if fname == '-':
         stream = sys.stdin
     else:
         try:
             stream = open(fname)
         except IOError as msg:
             print(msg, file=sys.stderr)
             sys.exit(1)
 
     parser = Parser()
 
     # we need nested try blocks because pre-2.5 python doesn't support unified
     # try-except-finally
     try:
         try:
             # write colorized version to stdout
             parser.format(stream.read(),scheme=opts.scheme_name)
         except IOError as msg:
             # if user reads through a pager and quits, don't print traceback
             if msg.args != (32,'Broken pipe'):
                 raise
     finally:
         if stream is not sys.stdin:
             stream.close() # in case a non-handled exception happened above
 
 if __name__ == "__main__":
     main()

IPython/utils/capture.py

@@ -1,173 +1,179
 # encoding: utf-8
 """
 IO capturing utilities.
 """
 
 #-----------------------------------------------------------------------------
 #  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.
 #-----------------------------------------------------------------------------
 from __future__ import print_function, absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import sys
-from io import StringIO
+
+from IPython.utils.py3compat import PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 
 class RichOutput(object):
     def __init__(self, source="", data=None, metadata=None):
         self.source = source
         self.data = data or {}
         self.metadata = metadata or {}
     
     def display(self):
         from IPython.display import publish_display_data
         publish_display_data(self.source, self.data, self.metadata)
     
     def _repr_mime_(self, mime):
         if mime not in self.data:
             return
         data = self.data[mime]
         if mime in self.metadata:
             return data, self.metadata[mime]
         else:
             return data
             
     def _repr_html_(self):
         return self._repr_mime_("text/html")
     
     def _repr_latex_(self):
         return self._repr_mime_("text/latex")
     
     def _repr_json_(self):
         return self._repr_mime_("application/json")
     
     def _repr_javascript_(self):
         return self._repr_mime_("application/javascript")
     
     def _repr_png_(self):
         return self._repr_mime_("image/png")
     
     def _repr_jpeg_(self):
         return self._repr_mime_("image/jpeg")
     
     def _repr_svg_(self):
         return self._repr_mime_("image/svg+xml")
 
 
 class CapturedIO(object):
     """Simple object for containing captured stdout/err and rich display StringIO objects
 
     Each instance `c` has three attributes:
 
     - ``c.stdout`` : standard output as a string
     - ``c.stderr`` : standard error as a string
     - ``c.outputs``: a list of rich display outputs
 
     Additionally, there's a ``c.show()`` method which will print all of the
     above in the same order, and can be invoked simply via ``c()``.
     """
     
     def __init__(self, stdout, stderr, outputs=None):
         self._stdout = stdout
         self._stderr = stderr
         if outputs is None:
             outputs = []
         self._outputs = outputs
     
     def __str__(self):
         return self.stdout
     
     @property
     def stdout(self):
         "Captured standard output"
         if not self._stdout:
             return ''
         return self._stdout.getvalue()
     
     @property
     def stderr(self):
         "Captured standard error"
         if not self._stderr:
             return ''
         return self._stderr.getvalue()
     
     @property
     def outputs(self):
         """A list of the captured rich display outputs, if any.
         
         If you have a CapturedIO object ``c``, these can be displayed in IPython
         using::
 
             from IPython.display import display
             for o in c.outputs:
                 display(o)
         """
         return [ RichOutput(s, d, md) for s, d, md in self._outputs ]
     
     def show(self):
         """write my output to sys.stdout/err as appropriate"""
         sys.stdout.write(self.stdout)
         sys.stderr.write(self.stderr)
         sys.stdout.flush()
         sys.stderr.flush()
         for source, data, metadata in self._outputs:
             RichOutput(source, data, metadata).display()
     
     __call__ = show
 
 
 class capture_output(object):
     """context manager for capturing stdout/err"""
     stdout = True
     stderr = True
     display = True
     
     def __init__(self, stdout=True, stderr=True, display=True):
         self.stdout = stdout
         self.stderr = stderr
         self.display = display
         self.shell = None
     
     def __enter__(self):
         from IPython.core.getipython import get_ipython
         from IPython.core.displaypub import CapturingDisplayPublisher
         
         self.sys_stdout = sys.stdout
         self.sys_stderr = sys.stderr
         
         if self.display:
             self.shell = get_ipython()
             if self.shell is None:
                 self.save_display_pub = None
                 self.display = False
         
         stdout = stderr = outputs = None
         if self.stdout:
             stdout = sys.stdout = StringIO()
         if self.stderr:
             stderr = sys.stderr = StringIO()
         if self.display:
             self.save_display_pub = self.shell.display_pub
             self.shell.display_pub = CapturingDisplayPublisher()
             outputs = self.shell.display_pub.outputs
             
         
         return CapturedIO(stdout, stderr, outputs)
     
     def __exit__(self, exc_type, exc_value, traceback):
         sys.stdout = self.sys_stdout
         sys.stderr = self.sys_stderr
         if self.display and self.shell:
             self.shell.display_pub = self.save_display_pub
 
 

IPython/utils/tests/test_io.py

@@ -1,86 +1,90
 # encoding: utf-8
 """Tests for io.py"""
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 
-from io import StringIO
 from subprocess import Popen, PIPE
 import unittest
 
 import nose.tools as nt
 
 from IPython.utils.io import Tee, capture_output
-from IPython.utils.py3compat import doctest_refactor_print
+from IPython.utils.py3compat import doctest_refactor_print, PY3
+
+if PY3:
+    from io import StringIO
+else:
+    from StringIO import StringIO
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 
 def test_tee_simple():
     "Very simple check with stdout only"
     chan = StringIO()
     text = 'Hello'
     tee = Tee(chan, channel='stdout')
     print(text, file=chan)
     nt.assert_equal(chan.getvalue(), text+"\n")
 
 
 class TeeTestCase(unittest.TestCase):
 
     def tchan(self, channel, check='close'):
         trap = StringIO()
         chan = StringIO()
         text = 'Hello'
         
         std_ori = getattr(sys, channel)
         setattr(sys, channel, trap)
 
         tee = Tee(chan, channel=channel)
         print(text, end='', file=chan)
         setattr(sys, channel, std_ori)
         trap_val = trap.getvalue()
         nt.assert_equal(chan.getvalue(), text)
         if check=='close':
             tee.close()
         else:
             del tee
 
     def test(self):
         for chan in ['stdout', 'stderr']:
             for check in ['close', 'del']:
                 self.tchan(chan, check)
 
 def test_io_init():
     """Test that io.stdin/out/err exist at startup"""
     for name in ('stdin', 'stdout', 'stderr'):
         cmd = doctest_refactor_print("from IPython.utils import io;print io.%s.__class__"%name)
         p = Popen([sys.executable, '-c', cmd],
                     stdout=PIPE)
         p.wait()
         classname = p.stdout.read().strip().decode('ascii')
         # __class__ is a reference to the class object in Python 3, so we can't
         # just test for string equality.
         assert 'IPython.utils.io.IOStream' in classname, classname
 
 def test_capture_output():
     """capture_output() context works"""
     
     with capture_output() as io:
         print('hi, stdout')
         print('hi, stderr', file=sys.stderr)
     
     nt.assert_equal(io.stdout, 'hi, stdout\n')
     nt.assert_equal(io.stderr, 'hi, stderr\n')