Fix almost all IPython.core tests for Python 3.

IPython/core/oinspect.py

@@ -1,772 +1,771 b''
 # -*- coding: utf-8 -*-
 """Tools for inspecting Python objects.
 
 Uses syntax highlighting for presenting the various information elements.
 
 Similar in spirit to the inspect module, but all calls take a name argument to
 reference the name under which an object is being read.
 """
 
 #*****************************************************************************
 #       Copyright (C) 2001-2004 Fernando Perez <[e-mail]>
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #*****************************************************************************
 
 __all__ = ['Inspector','InspectColors']
 
 # stdlib modules
 import __builtin__
 import inspect
 import linecache
 import os
 import sys
 import types
 from collections import namedtuple
 try:
     from itertools import izip_longest
 except ImportError:
     from itertools import zip_longest as izip_longest
 
 # IPython's own
 from IPython.core import page
 from IPython.utils import PyColorize
 from IPython.utils import io
 from IPython.utils import py3compat
 from IPython.utils.text import indent
 from IPython.utils.wildcard import list_namespace
 from IPython.utils.coloransi import *
 
 #****************************************************************************
 # Builtin color schemes
 
 Colors = TermColors  # just a shorthand
 
 # Build a few color schemes
 NoColor = ColorScheme(
     'NoColor',{
     'header' : Colors.NoColor,
     'normal' : Colors.NoColor  # color off (usu. Colors.Normal)
     }  )
 
 LinuxColors = ColorScheme(
     'Linux',{
     'header' : Colors.LightRed,
     'normal' : Colors.Normal  # color off (usu. Colors.Normal)
     } )
 
 LightBGColors = ColorScheme(
     'LightBG',{
     'header' : Colors.Red,
     'normal' : Colors.Normal  # color off (usu. Colors.Normal)
     }  )
 
 # Build table of color schemes (needed by the parser)
 InspectColors = ColorSchemeTable([NoColor,LinuxColors,LightBGColors],
                                  'Linux')
 
 #****************************************************************************
 # Auxiliary functions and objects
 
 # See the messaging spec for the definition of all these fields.  This list
 # effectively defines the order of display
 info_fields = ['type_name', 'base_class', 'string_form', 'namespace',
                'length', 'file', 'definition', 'docstring', 'source',
                'init_definition', 'class_docstring', 'init_docstring',
                'call_def', 'call_docstring',
                # These won't be printed but will be used to determine how to
                # format the object
                'ismagic', 'isalias', 'isclass', 'argspec', 'found', 'name'
                ]
 
 
 def object_info(**kw):
     """Make an object info dict with all fields present."""
     infodict = dict(izip_longest(info_fields, [None]))
     infodict.update(kw)
     return infodict
 
 
 def getdoc(obj):
     """Stable wrapper around inspect.getdoc.
 
     This can't crash because of attribute problems.
 
     It also attempts to call a getdoc() method on the given object.  This
     allows objects which provide their docstrings via non-standard mechanisms
     (like Pyro proxies) to still be inspected by ipython's ? system."""
 
     ds = None  # default return value
     try:
         ds = inspect.getdoc(obj)
     except:
         # Harden against an inspect failure, which can occur with
         # SWIG-wrapped extensions.
         pass
     # Allow objects to offer customized documentation via a getdoc method:
     try:
         ds2 = obj.getdoc()
     except:
         pass
     else:
         # if we get extra info, we add it to the normal docstring.
         if ds is None:
             ds = ds2
         else:
             ds = '%s\n%s' % (ds,ds2)
     return ds
 
 
 def getsource(obj,is_binary=False):
     """Wrapper around inspect.getsource.
 
     This can be modified by other projects to provide customized source
     extraction.
 
     Inputs:
 
     - obj: an object whose source code we will attempt to extract.
 
     Optional inputs:
 
     - is_binary: whether the object is known to come from a binary source.
     This implementation will skip returning any output for binary objects, but
     custom extractors may know how to meaningfully process them."""
 
     if is_binary:
         return None
     else:
         # get source if obj was decorated with @decorator
         if hasattr(obj,"__wrapped__"):
             obj = obj.__wrapped__
         try:
             src = inspect.getsource(obj)
         except TypeError:
             if hasattr(obj,'__class__'):
                 src = inspect.getsource(obj.__class__)
         return src
 
 def getargspec(obj):
     """Get the names and default values of a function's arguments.
 
     A tuple of four things is returned: (args, varargs, varkw, defaults).
     'args' is a list of the argument names (it may contain nested lists).
     'varargs' and 'varkw' are the names of the * and ** arguments or None.
     'defaults' is an n-tuple of the default values of the last n arguments.
 
     Modified version of inspect.getargspec from the Python Standard
     Library."""
 
     if inspect.isfunction(obj):
         func_obj = obj
     elif inspect.ismethod(obj):
         func_obj = obj.im_func
     elif hasattr(obj, '__call__'):
         func_obj = obj.__call__
     else:
         raise TypeError('arg is not a Python function')
     args, varargs, varkw = inspect.getargs(func_obj.func_code)
     return args, varargs, varkw, func_obj.func_defaults
 
 
 def format_argspec(argspec):
     """Format argspect, convenience wrapper around inspect's.
 
     This takes a dict instead of ordered arguments and calls
     inspect.format_argspec with the arguments in the necessary order.
     """
     return inspect.formatargspec(argspec['args'], argspec['varargs'],
                                  argspec['varkw'], argspec['defaults'])
 
 
 def call_tip(oinfo, format_call=True):
     """Extract call tip data from an oinfo dict.
 
     Parameters
     ----------
     oinfo : dict
 
     format_call : bool, optional
       If True, the call line is formatted and returned as a string.  If not, a
       tuple of (name, argspec) is returned.
 
     Returns
     -------
     call_info : None, str or (str, dict) tuple.
       When format_call is True, the whole call information is formattted as a
       single string.  Otherwise, the object's name and its argspec dict are
       returned.  If no call information is available, None is returned.
 
     docstring : str or None
       The most relevant docstring for calling purposes is returned, if
       available.  The priority is: call docstring for callable instances, then
       constructor docstring for classes, then main object's docstring otherwise
       (regular functions).
     """
     # Get call definition
     argspec = oinfo.get('argspec')
     if argspec is None:
         call_line = None
     else:
         # Callable objects will have 'self' as their first argument, prune
         # it out if it's there for clarity (since users do *not* pass an
         # extra first argument explicitly).
         try:
             has_self = argspec['args'][0] == 'self'
         except (KeyError, IndexError):
             pass
         else:
             if has_self:
                 argspec['args'] = argspec['args'][1:]
 
         call_line = oinfo['name']+format_argspec(argspec)
 
     # Now get docstring.
     # The priority is: call docstring, constructor docstring, main one.
     doc = oinfo.get('call_docstring')
     if doc is None:
         doc = oinfo.get('init_docstring')
     if doc is None:
         doc = oinfo.get('docstring','')
 
     return call_line, doc
 
 
 class Inspector:
     def __init__(self, color_table=InspectColors,
                  code_color_table=PyColorize.ANSICodeColors,
                  scheme='NoColor',
                  str_detail_level=0):
         self.color_table = color_table
         self.parser = PyColorize.Parser(code_color_table,out='str')
         self.format = self.parser.format
         self.str_detail_level = str_detail_level
         self.set_active_scheme(scheme)
 
     def _getdef(self,obj,oname=''):
         """Return the definition header for any callable object.
 
         If any exception is generated, None is returned instead and the
         exception is suppressed."""
 
         try:
             # We need a plain string here, NOT unicode!
             hdef = oname + inspect.formatargspec(*getargspec(obj))
             return py3compat.unicode_to_str(hdef, 'ascii')
         except:
             return None
 
     def __head(self,h):
         """Return a header string with proper colors."""
         return '%s%s%s' % (self.color_table.active_colors.header,h,
                            self.color_table.active_colors.normal)
 
     def set_active_scheme(self,scheme):
         self.color_table.set_active_scheme(scheme)
         self.parser.color_table.set_active_scheme(scheme)
 
     def noinfo(self,msg,oname):
         """Generic message when no information is found."""
         print 'No %s found' % msg,
         if oname:
             print 'for %s' % oname
         else:
             print
 
     def pdef(self,obj,oname=''):
         """Print the definition header for any callable object.
 
         If the object is a class, print the constructor information."""
 
         if not callable(obj):
             print 'Object is not callable.'
             return
 
         header = ''
 
         if inspect.isclass(obj):
             header = self.__head('Class constructor information:\n')
             obj = obj.__init__
         elif type(obj) is types.InstanceType:
             obj = obj.__call__
 
         output = self._getdef(obj,oname)
         if output is None:
             self.noinfo('definition header',oname)
         else:
             print >>io.stdout, header,self.format(output),
 
     def pdoc(self,obj,oname='',formatter = None):
         """Print the docstring for any object.
 
         Optional:
         -formatter: a function to run the docstring through for specially
         formatted docstrings.
 
         Examples
         --------
 
         In [1]: class NoInit:
            ...:     pass
 
         In [2]: class NoDoc:
            ...:     def __init__(self):
            ...:         pass
 
         In [3]: %pdoc NoDoc
         No documentation found for NoDoc
 
         In [4]: %pdoc NoInit
         No documentation found for NoInit
 
         In [5]: obj = NoInit()
 
         In [6]: %pdoc obj
         No documentation found for obj
 
         In [5]: obj2 = NoDoc()
 
         In [6]: %pdoc obj2
         No documentation found for obj2
         """
 
         head = self.__head  # For convenience
         lines = []
         ds = getdoc(obj)
         if formatter:
             ds = formatter(ds)
         if ds:
             lines.append(head("Class Docstring:"))
             lines.append(indent(ds))
         if inspect.isclass(obj) and hasattr(obj, '__init__'):
             init_ds = getdoc(obj.__init__)
             if init_ds is not None:
                 lines.append(head("Constructor Docstring:"))
                 lines.append(indent(init_ds))
-        elif (type(obj) is types.InstanceType or isinstance(obj,object)) \
-                 and hasattr(obj,'__call__'):
+        elif hasattr(obj,'__call__'):
             call_ds = getdoc(obj.__call__)
             if call_ds:
                 lines.append(head("Calling Docstring:"))
                 lines.append(indent(call_ds))
 
         if not lines:
             self.noinfo('documentation',oname)
         else:
             page.page('\n'.join(lines))
 
     def psource(self,obj,oname=''):
         """Print the source code for an object."""
 
         # Flush the source cache because inspect can return out-of-date source
         linecache.checkcache()
         try:
             src = getsource(obj)
         except:
             self.noinfo('source',oname)
         else:
             page.page(self.format(py3compat.unicode_to_str(src)))
 
     def pfile(self,obj,oname=''):
         """Show the whole file where an object was defined."""
 
         try:
             try:
                 lineno = inspect.getsourcelines(obj)[1]
             except TypeError:
                 # For instances, try the class object like getsource() does
                 if hasattr(obj,'__class__'):
                     lineno = inspect.getsourcelines(obj.__class__)[1]
                     # Adjust the inspected object so getabsfile() below works
                     obj = obj.__class__
         except:
             self.noinfo('file',oname)
             return
 
         # We only reach this point if object was successfully queried
 
         # run contents of file through pager starting at line
         # where the object is defined
         ofile = inspect.getabsfile(obj)
 
         if ofile.endswith(('.so', '.dll', '.pyd')):
             print 'File %r is binary, not printing.' % ofile
         elif not os.path.isfile(ofile):
             print 'File %r does not exist, not printing.' % ofile
         else:
             # Print only text files, not extension binaries.  Note that
             # getsourcelines returns lineno with 1-offset and page() uses
             # 0-offset, so we must adjust.
             page.page(self.format(open(ofile).read()),lineno-1)
 
     def _format_fields(self, fields, title_width=12):
         """Formats a list of fields for display.
 
         Parameters
         ----------
         fields : list
           A list of 2-tuples: (field_title, field_content)
         title_width : int
           How many characters to pad titles to. Default 12.
         """
         out = []
         header = self.__head
         for title, content in fields:
             if len(content.splitlines()) > 1:
                 title = header(title + ":") + "\n"
             else:
                 title = header((title+":").ljust(title_width))
             out.append(title + content)
         return "\n".join(out)
 
     # The fields to be displayed by pinfo: (fancy_name, key_in_info_dict)
     pinfo_fields1 = [("Type", "type_name"),
                     ("Base Class", "base_class"),
                     ("String Form", "string_form"),
                     ("Namespace", "namespace"),
                     ("Length", "length"),
                     ("File", "file"),
                     ("Definition", "definition")]
 
     pinfo_fields_obj = [("Class Docstring", "class_docstring"),
                         ("Constructor Docstring","init_docstring"),
                         ("Call def", "call_def"),
                         ("Call docstring", "call_docstring")]
 
     def pinfo(self,obj,oname='',formatter=None,info=None,detail_level=0):
         """Show detailed information about an object.
 
         Optional arguments:
 
         - oname: name of the variable pointing to the object.
 
         - formatter: special formatter for docstrings (see pdoc)
 
         - info: a structure with some information fields which may have been
         precomputed already.
 
         - detail_level: if set to 1, more information is given.
         """
         info = self.info(obj, oname=oname, formatter=formatter,
                             info=info, detail_level=detail_level)
         displayfields = []
         for title, key in self.pinfo_fields1:
             field = info[key]
             if field is not None:
                 displayfields.append((title, field.rstrip()))
 
         # Source or docstring, depending on detail level and whether
         # source found.
         if detail_level > 0 and info['source'] is not None:
             displayfields.append(("Source", self.format(py3compat.unicode_to_str(info['source']))))
         elif info['docstring'] is not None:
             displayfields.append(("Docstring", info["docstring"]))
 
         # Constructor info for classes
         if info['isclass']:
             if info['init_definition'] or info['init_docstring']:
                 displayfields.append(("Constructor information", ""))
                 if info['init_definition'] is not None:
                     displayfields.append((" Definition",
                                     info['init_definition'].rstrip()))
                 if info['init_docstring'] is not None:
                     displayfields.append((" Docstring",
                                         indent(info['init_docstring'])))
 
         # Info for objects:
         else:
             for title, key in self.pinfo_fields_obj:
                 field = info[key]
                 if field is not None:
                     displayfields.append((title, field.rstrip()))
 
         # Finally send to printer/pager:
         if displayfields:
             page.page(self._format_fields(displayfields))
 
     def info(self, obj, oname='', formatter=None, info=None, detail_level=0):
         """Compute a dict with detailed information about an object.
 
         Optional arguments:
 
         - oname: name of the variable pointing to the object.
 
         - formatter: special formatter for docstrings (see pdoc)
 
         - info: a structure with some information fields which may have been
         precomputed already.
 
         - detail_level: if set to 1, more information is given.
         """
 
         obj_type = type(obj)
 
         header = self.__head
         if info is None:
             ismagic = 0
             isalias = 0
             ospace = ''
         else:
             ismagic = info.ismagic
             isalias = info.isalias
             ospace = info.namespace
 
         # Get docstring, special-casing aliases:
         if isalias:
             if not callable(obj):
                 try:
                     ds = "Alias to the system command:\n  %s" % obj[1]
                 except:
                     ds = "Alias: " + str(obj)
             else:
                 ds = "Alias to " + str(obj)
                 if obj.__doc__:
                     ds += "\nDocstring:\n" + obj.__doc__
         else:
             ds = getdoc(obj)
             if ds is None:
                 ds = '<no docstring>'
         if formatter is not None:
             ds = formatter(ds)
 
         # store output in a dict, we initialize it here and fill it as we go
         out = dict(name=oname, found=True, isalias=isalias, ismagic=ismagic)
 
         string_max = 200 # max size of strings to show (snipped if longer)
         shalf = int((string_max -5)/2)
 
         if ismagic:
             obj_type_name = 'Magic function'
         elif isalias:
             obj_type_name = 'System alias'
         else:
             obj_type_name = obj_type.__name__
         out['type_name'] = obj_type_name
 
         try:
             bclass = obj.__class__
             out['base_class'] = str(bclass)
         except: pass
 
         # String form, but snip if too long in ? form (full in ??)
         if detail_level >= self.str_detail_level:
             try:
                 ostr = str(obj)
                 str_head = 'string_form'
                 if not detail_level and len(ostr)>string_max:
                     ostr = ostr[:shalf] + ' <...> ' + ostr[-shalf:]
                     ostr = ("\n" + " " * len(str_head.expandtabs())).\
                             join(q.strip() for q in ostr.split("\n"))
                 out[str_head] = ostr
             except:
                 pass
 
         if ospace:
             out['namespace'] = ospace
 
         # Length (for strings and lists)
         try:
             out['length'] = str(len(obj))
         except: pass
 
         # Filename where object was defined
         binary_file = False
         try:
             try:
                 fname = inspect.getabsfile(obj)
             except TypeError:
                 # For an instance, the file that matters is where its class was
                 # declared.
                 if hasattr(obj,'__class__'):
                     fname = inspect.getabsfile(obj.__class__)
             if fname.endswith('<string>'):
                 fname = 'Dynamically generated function. No source code available.'
             if fname.endswith(('.so', '.dll', '.pyd')):
                 binary_file = True
             out['file'] = fname
         except:
             # if anything goes wrong, we don't want to show source, so it's as
             # if the file was binary
             binary_file = True
 
         # reconstruct the function definition and print it:
         defln = self._getdef(obj, oname)
         if defln:
             out['definition'] = self.format(defln)
 
         # Docstrings only in detail 0 mode, since source contains them (we
         # avoid repetitions).  If source fails, we add them back, see below.
         if ds and detail_level == 0:
                 out['docstring'] = ds
 
         # Original source code for any callable
         if detail_level:
             # Flush the source cache because inspect can return out-of-date
             # source
             linecache.checkcache()
             source = None
             try:
                 try:
                     source = getsource(obj,binary_file)
                 except TypeError:
                     if hasattr(obj,'__class__'):
                         source = getsource(obj.__class__,binary_file)
                 if source is not None:
                     out['source'] = source.rstrip()
             except Exception:
                 pass
 
             if ds and source is None:
                 out['docstring'] = ds
 
 
         # Constructor docstring for classes
         if inspect.isclass(obj):
             out['isclass'] = True
             # reconstruct the function definition and print it:
             try:
                 obj_init =  obj.__init__
             except AttributeError:
                 init_def = init_ds = None
             else:
                 init_def = self._getdef(obj_init,oname)
                 init_ds  = getdoc(obj_init)
                 # Skip Python's auto-generated docstrings
                 if init_ds and \
                        init_ds.startswith('x.__init__(...) initializes'):
                     init_ds = None
 
             if init_def or init_ds:
                 if init_def:
                     out['init_definition'] = self.format(init_def)
                 if init_ds:
                     out['init_docstring'] = init_ds
 
         # and class docstring for instances:
         else:
             # First, check whether the instance docstring is identical to the
             # class one, and print it separately if they don't coincide.  In
             # most cases they will, but it's nice to print all the info for
             # objects which use instance-customized docstrings.
             if ds:
                 try:
                     cls = getattr(obj,'__class__')
                 except:
                     class_ds = None
                 else:
                     class_ds = getdoc(cls)
                 # Skip Python's auto-generated docstrings
                 if class_ds and \
                        (class_ds.startswith('function(code, globals[,') or \
                    class_ds.startswith('instancemethod(function, instance,') or \
                    class_ds.startswith('module(name[,') ):
                     class_ds = None
                 if class_ds and ds != class_ds:
                     out['class_docstring'] = class_ds
 
             # Next, try to show constructor docstrings
             try:
                 init_ds = getdoc(obj.__init__)
                 # Skip Python's auto-generated docstrings
                 if init_ds and \
                        init_ds.startswith('x.__init__(...) initializes'):
                     init_ds = None
             except AttributeError:
                 init_ds = None
             if init_ds:
                 out['init_docstring'] = init_ds
 
             # Call form docstring for callable instances
             if hasattr(obj, '__call__'):
                 call_def = self._getdef(obj.__call__, oname)
                 if call_def is not None:
                     out['call_def'] = self.format(call_def)
                 call_ds = getdoc(obj.__call__)
                 # Skip Python's auto-generated docstrings
                 if call_ds and call_ds.startswith('x.__call__(...) <==> x(...)'):
                     call_ds = None
                 if call_ds:
                     out['call_docstring'] = call_ds
 
         # Compute the object's argspec as a callable.  The key is to decide
         # whether to pull it from the object itself, from its __init__ or
         # from its __call__ method.
 
         if inspect.isclass(obj):
             # Old-style classes need not have an __init__
             callable_obj = getattr(obj, "__init__", None)
         elif callable(obj):
             callable_obj = obj
         else:
             callable_obj = None
 
         if callable_obj:
             try:
                 args,  varargs, varkw, defaults = getargspec(callable_obj)
             except (TypeError, AttributeError):
                 # For extensions/builtins we can't retrieve the argspec
                 pass
             else:
                 out['argspec'] = dict(args=args, varargs=varargs,
                                       varkw=varkw, defaults=defaults)
 
         return object_info(**out)
 
 
     def psearch(self,pattern,ns_table,ns_search=[],
                 ignore_case=False,show_all=False):
         """Search namespaces with wildcards for objects.
 
         Arguments:
 
         - pattern: string containing shell-like wildcards to use in namespace
         searches and optionally a type specification to narrow the search to
         objects of that type.
 
         - ns_table: dict of name->namespaces for search.
 
         Optional arguments:
 
           - ns_search: list of namespace names to include in search.
 
           - ignore_case(False): make the search case-insensitive.
 
           - show_all(False): show all names, including those starting with
           underscores.
         """
         #print 'ps pattern:<%r>' % pattern # dbg
 
         # defaults
         type_pattern = 'all'
         filter = ''
 
         cmds = pattern.split()
         len_cmds  =  len(cmds)
         if len_cmds == 1:
             # Only filter pattern given
             filter = cmds[0]
         elif len_cmds == 2:
             # Both filter and type specified
             filter,type_pattern = cmds
         else:
             raise ValueError('invalid argument string for psearch: <%s>' %
                              pattern)
 
         # filter search namespaces
         for name in ns_search:
             if name not in ns_table:
                 raise ValueError('invalid namespace <%s>. Valid names: %s' %
                                  (name,ns_table.keys()))
 
         #print 'type_pattern:',type_pattern # dbg
         search_result = []
         for ns_name in ns_search:
             ns = ns_table[ns_name]
             tmp_res = list(list_namespace(ns,type_pattern,filter,
                                           ignore_case=ignore_case,
                                           show_all=show_all))
             search_result.extend(tmp_res)
         search_result.sort()
 
         page.page('\n'.join(search_result))

IPython/core/tests/test_handlers.py

@@ -1,167 +1,168 b''
 """Tests for input handlers.
 """
 #-----------------------------------------------------------------------------
 # Module imports
 #-----------------------------------------------------------------------------
 
 # third party
 import nose.tools as nt
 
 # our own packages
 from IPython.core import autocall
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.testing.globalipapp import get_ipython
+from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # Get the public instance of IPython
 ip = get_ipython()
 
 failures = []
 num_tests = 0
 
 #-----------------------------------------------------------------------------
 # Test functions
 #-----------------------------------------------------------------------------
 
 class CallableIndexable(object):
     def __getitem__(self, idx): return True
     def __call__(self, *args, **kws): return True
 
 
 class Autocallable(autocall.IPyAutocall):
     def __call__(self):
         return "called"
 
 
 def run(tests):
     """Loop through a list of (pre, post) inputs, where pre is the string
     handed to ipython, and post is how that string looks after it's been
     transformed (i.e. ipython's notion of _i)"""
     tt.check_pairs(ip.prefilter_manager.prefilter_lines, tests)
 
 
 def test_handlers():
     # alias expansion
 
     # We're using 'true' as our syscall of choice because it doesn't
     # write anything to stdout.
 
     # Turn off actual execution of aliases, because it's noisy
     old_system_cmd = ip.system
     ip.system = lambda cmd: None
 
 
     ip.alias_manager.alias_table['an_alias'] = (0, 'true')
     # These are useful for checking a particular recursive alias issue
     ip.alias_manager.alias_table['top'] = (0, 'd:/cygwin/top')
     ip.alias_manager.alias_table['d'] =   (0, 'true')
     run([(i,py3compat.u_format(o)) for i,o in \
         [("an_alias",    'get_ipython().system({u}"true ")'),     # alias
          # Below: recursive aliases should expand whitespace-surrounded
          # chars, *not* initial chars which happen to be aliases:
          ("top",         'get_ipython().system({u}"d:/cygwin/top ")'),
          ]])
     ip.system = old_system_cmd
 
     call_idx = CallableIndexable()
     ip.user_ns['call_idx'] = call_idx
 
     # For many of the below, we're also checking that leading whitespace
     # turns off the esc char, which it should unless there is a continuation
     # line.
     run([(i,py3compat.u_format(o)) for i,o in \
         [('"no change"', '"no change"'),             # normal
          ("!true",       'get_ipython().system({u}"true")'),      # shell_escapes
          ("!! true",     'get_ipython().magic({u}"sx  true")'),   # shell_escapes + magic
          ("!!true",      'get_ipython().magic({u}"sx true")'),    # shell_escapes + magic
          ("%lsmagic",    'get_ipython().magic({u}"lsmagic ")'),   # magic
          ("lsmagic",     'get_ipython().magic({u}"lsmagic ")'),   # magic
          #("a = b # PYTHON-MODE", '_i'),          # emacs -- avoids _in cache
 
          # post-esc-char whitespace goes inside
          ("! true",   'get_ipython().system({u}" true")'),
 
          # handle_help
 
          # These are weak tests -- just looking at what the help handlers
          # logs, which is not how it really does its work.  But it still
          # lets us check the key paths through the handler.
 
          ("x=1 # what?", "x=1 # what?"), # no help if valid python
          ]])
 
     # multi_line_specials
     ip.prefilter_manager.multi_line_specials = False
     # W/ multi_line_specials off, leading ws kills esc chars/autoexpansion
     run([
         ('if 1:\n    !true',    'if 1:\n    !true'),
         ('if 1:\n    lsmagic',  'if 1:\n    lsmagic'),
         ('if 1:\n    an_alias', 'if 1:\n    an_alias'),
         ])
 
     ip.prefilter_manager.multi_line_specials = True
     # initial indents must be preserved.
     run([(i,py3compat.u_format(o)) for i,o in \
         [('if 1:\n    !true',    'if 1:\n    get_ipython().system({u}"true")'),
          ('if 2:\n    lsmagic',  'if 2:\n    get_ipython().magic({u}"lsmagic ")'),
          ('if 1:\n    an_alias', 'if 1:\n    get_ipython().system({u}"true ")'),
          # Weird one
          ('if 1:\n    !!true',   'if 1:\n    get_ipython().magic({u}"sx true")'),
 
          # Even with m_l_s on, autocall is off even with special chars
          ('if 1:\n    /fun 1 2', 'if 1:\n    /fun 1 2'),
          ('if 1:\n    ;fun 1 2', 'if 1:\n    ;fun 1 2'),
          ('if 1:\n    ,fun 1 2', 'if 1:\n    ,fun 1 2'),
          ('if 1:\n    ?fun 1 2', 'if 1:\n    ?fun 1 2'),
          # What about !!
          ]])
 
     # Objects which are instances of IPyAutocall are *always* autocalled
     autocallable = Autocallable()
     ip.user_ns['autocallable'] = autocallable
 
     # auto
     ip.magic('autocall 0')
     # Only explicit escapes or instances of IPyAutocallable should get
     # expanded
     run([
         ('len "abc"',       'len "abc"'),
         ('autocallable',    'autocallable()'),
         (",list 1 2 3",     'list("1", "2", "3")'),
         (";list 1 2 3",     'list("1 2 3")'),
         ("/len range(1,4)", 'len(range(1,4))'),
         ])
     ip.magic('autocall 1')
     run([
         (",list 1 2 3", 'list("1", "2", "3")'),
         (";list 1 2 3", 'list("1 2 3")'),
         ("/len range(1,4)", 'len(range(1,4))'),
         ('len "abc"', 'len("abc")'),
         ('len "abc";', 'len("abc");'),  # ; is special -- moves out of parens
         # Autocall is turned off if first arg is [] and the object
         # is both callable and indexable.  Like so:
         ('len [1,2]', 'len([1,2])'),      # len doesn't support __getitem__...
         ('call_idx [1]', 'call_idx [1]'), # call_idx *does*..
         ('call_idx 1', 'call_idx(1)'),
         ('len', 'len '), # only at 2 does it auto-call on single args
         ])
     ip.magic('autocall 2')
     run([
         (",list 1 2 3", 'list("1", "2", "3")'),
         (";list 1 2 3", 'list("1 2 3")'),
         ("/len range(1,4)", 'len(range(1,4))'),
         ('len "abc"', 'len("abc")'),
         ('len "abc";', 'len("abc");'),
         ('len [1,2]', 'len([1,2])'),
         ('call_idx [1]', 'call_idx [1]'),
         ('call_idx 1', 'call_idx(1)'),
         # This is what's different:
         ('len', 'len()'), # only at 2 does it auto-call on single args
         ])
     ip.magic('autocall 1')
 
     nt.assert_equals(failures, [])

IPython/core/tests/test_history.py

@@ -1,118 +1,118 b''
 # coding: utf-8
 """Tests for the IPython tab-completion machinery.
 """
 #-----------------------------------------------------------------------------
 # Module imports
 #-----------------------------------------------------------------------------
 
 # stdlib
 import os
 import sys
 import unittest
 from datetime import datetime
 # third party
 import nose.tools as nt
 
 # our own packages
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.core.history import HistoryManager, extract_hist_ranges
 from IPython.utils import py3compat
 
 def setUp():
     nt.assert_equal(sys.getdefaultencoding(), "utf-8" if py3compat.PY3 else "ascii")
 
 def test_history():
     ip = get_ipython()
     with TemporaryDirectory() as tmpdir:
         hist_manager_ori = ip.history_manager
         hist_file = os.path.join(tmpdir, 'history.sqlite')
         try:
             ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file)
             hist = ['a=1', 'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
             for i, h in enumerate(hist, start=1):
                 ip.history_manager.store_inputs(i, h)
 
             ip.history_manager.db_log_output = True
             # Doesn't match the input, but we'll just check it's stored.
             ip.history_manager.output_hist_reprs[3] = "spam"
             ip.history_manager.store_output(3)
 
             nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
 
             # Check whether specifying a range beyond the end of the current
             # session results in an error (gh-804)
             ip.magic('%hist 2-500')
 
             # New session
             ip.history_manager.reset()
             newcmds = ["z=5","class X(object):\n    pass", "k='p'"]
             for i, cmd in enumerate(newcmds, start=1):
                 ip.history_manager.store_inputs(i, cmd)
             gothist = ip.history_manager.get_range(start=1, stop=4)
             nt.assert_equal(list(gothist), zip([0,0,0],[1,2,3], newcmds))
             # Previous session:
             gothist = ip.history_manager.get_range(-1, 1, 4)
             nt.assert_equal(list(gothist), zip([1,1,1],[1,2,3], hist))
 
             # Check get_hist_tail
             gothist = ip.history_manager.get_tail(4, output=True,
                                                     include_latest=True)
             expected = [(1, 3, (hist[-1], "spam")),
                         (2, 1, (newcmds[0], None)),
                         (2, 2, (newcmds[1], None)),
                         (2, 3, (newcmds[2], None)),]
             nt.assert_equal(list(gothist), expected)
 
             gothist = ip.history_manager.get_tail(2)
             expected = [(2, 1, newcmds[0]),
                         (2, 2, newcmds[1])]
             nt.assert_equal(list(gothist), expected)
 
             # Check get_hist_search
             gothist = ip.history_manager.search("*test*")
             nt.assert_equal(list(gothist), [(1,2,hist[1])] )
             gothist = ip.history_manager.search("b*", output=True)
             nt.assert_equal(list(gothist), [(1,3,(hist[2],"spam"))] )
 
             # Cross testing: check that magic %save can get previous session.
             testfilename = os.path.realpath(os.path.join(tmpdir, "test.py"))
             ip.magic_save(testfilename + " ~1/1-3")
-            testfile = open(testfilename, "r")
-            nt.assert_equal(testfile.read().decode("utf-8"),
-                    "# coding: utf-8\n" + "\n".join(hist))
+            with py3compat.open(testfilename) as testfile:
+                nt.assert_equal(testfile.read(),
+                                        u"# coding: utf-8\n" + u"\n".join(hist))
 
             # Duplicate line numbers - check that it doesn't crash, and
             # gets a new session
             ip.history_manager.store_inputs(1, "rogue")
             ip.history_manager.writeout_cache()
             nt.assert_equal(ip.history_manager.session_number, 3)
         finally:
             # Restore history manager
             ip.history_manager = hist_manager_ori
 
 
 def test_extract_hist_ranges():
     instr = "1 2/3 ~4/5-6 ~4/7-~4/9 ~9/2-~7/5"
     expected = [(0, 1, 2),  # 0 == current session
                 (2, 3, 4),
                 (-4, 5, 7),
                 (-4, 7, 10),
                 (-9, 2, None),  # None == to end
                 (-8, 1, None),
                 (-7, 1, 6)]
     actual = list(extract_hist_ranges(instr))
     nt.assert_equal(actual, expected)
 
 def test_magic_rerun():
     """Simple test for %rerun (no args -> rerun last line)"""
     ip = get_ipython()
     ip.run_cell("a = 10")
     ip.run_cell("a += 1")
     nt.assert_equal(ip.user_ns["a"], 11)
     ip.run_cell("%rerun")
     nt.assert_equal(ip.user_ns["a"], 12)
 
 def test_timestamp_type():
     ip = get_ipython()
     info = ip.history_manager.get_session_info()
     nt.assert_true(isinstance(info[1], datetime))

IPython/core/tests/test_iplib.py

@@ -1,271 +1,272 b''
 """Tests for the key interactiveshell module, where the main ipython class is defined.
 """
 #-----------------------------------------------------------------------------
 # Module imports
 #-----------------------------------------------------------------------------
 
 # stdlib
 import os
 import shutil
 import tempfile
 
 # third party
 import nose.tools as nt
 
 # our own packages
 from IPython.testing import decorators as dec
 from IPython.testing.globalipapp import get_ipython
+from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # Get the public instance of IPython
 ip = get_ipython()
 
 #-----------------------------------------------------------------------------
 # Test functions
 #-----------------------------------------------------------------------------
 
 @dec.parametric
 def test_reset():
     """reset must clear most namespaces."""
     # The number of variables in the private user_ns_hidden is not zero, but it
     # should be constant regardless of what we do
     nvars_config_ns = len(ip.user_ns_hidden)
 
     # Check that reset runs without error
     ip.reset()
 
     # Once we've reset it (to clear of any junk that might have been there from
     # other tests, we can count how many variables are in the user's namespace
     nvars_user_ns = len(ip.user_ns)
 
     # Now add a few variables to user_ns, and check that reset clears them
     ip.user_ns['x'] = 1
     ip.user_ns['y'] = 1
     ip.reset()
     
     # Finally, check that all namespaces have only as many variables as we
     # expect to find in them:
     for ns in ip.ns_refs_table:
         if ns is ip.user_ns:
             nvars_expected = nvars_user_ns
         elif ns is ip.user_ns_hidden:
             nvars_expected = nvars_config_ns
         else:
             nvars_expected = 0
             
         yield nt.assert_equals(len(ns), nvars_expected)
 
 
 # Tests for reporting of exceptions in various modes, handling of SystemExit,
 # and %tb functionality.  This is really a mix of testing ultraTB and interactiveshell.
 
 def doctest_tb_plain():
     """
 In [18]: xmode plain
 Exception reporting mode: Plain
 
 In [19]: run simpleerr.py
 Traceback (most recent call last):
   ...line 32, in <module>
     bar(mode)
   ...line 16, in bar
     div0()
   ...line 8, in div0
     x/y
 ZeroDivisionError: ...
     """
 
 
 def doctest_tb_context():
     """
 In [3]: xmode context
 Exception reporting mode: Context
 
 In [4]: run simpleerr.py
 ---------------------------------------------------------------------------
 ZeroDivisionError                         Traceback (most recent call last)
 <BLANKLINE>
 ... in <module>()
      30         mode = 'div'
      31 
 ---> 32     bar(mode)
 <BLANKLINE>
 ... in bar(mode)
      14     "bar"
      15     if mode=='div':
 ---> 16         div0()
      17     elif mode=='exit':
      18         try:
 <BLANKLINE>
 ... in div0()
       6     x = 1
       7     y = 0
 ----> 8     x/y
       9 
      10 def sysexit(stat, mode):
 <BLANKLINE>
 ZeroDivisionError: ...
 """
 
 
 def doctest_tb_verbose():
     """
 In [5]: xmode verbose
 Exception reporting mode: Verbose
 
 In [6]: run simpleerr.py
 ---------------------------------------------------------------------------
 ZeroDivisionError                         Traceback (most recent call last)
 <BLANKLINE>
 ... in <module>()
      30         mode = 'div'
      31 
 ---> 32     bar(mode)
         global bar = <function bar at ...>
         global mode = 'div'
 <BLANKLINE>
 ... in bar(mode='div')
      14     "bar"
      15     if mode=='div':
 ---> 16         div0()
         global div0 = <function div0 at ...>
      17     elif mode=='exit':
      18         try:
 <BLANKLINE>
 ... in div0()
       6     x = 1
       7     y = 0
 ----> 8     x/y
         x = 1
         y = 0
       9 
      10 def sysexit(stat, mode):
 <BLANKLINE>
 ZeroDivisionError: ...
       """
 
 @py3compat.u_format
 def doctest_tb_sysexit():
     """
 In [17]: %xmode plain
 Exception reporting mode: Plain
 
 In [18]: %run simpleerr.py exit
 An exception has occurred, use %tb to see the full traceback.
 SystemExit: (1, {u}'Mode = exit')
 
 In [19]: %run simpleerr.py exit 2
 An exception has occurred, use %tb to see the full traceback.
 SystemExit: (2, {u}'Mode = exit')
 
 In [20]: %tb
 Traceback (most recent call last):
   File ... in <module>
     bar(mode)
   File ... line 22, in bar
     sysexit(stat, mode)
   File ... line 11, in sysexit
     raise SystemExit(stat, 'Mode = %s' % mode)
 SystemExit: (2, {u}'Mode = exit')
 
 In [21]: %xmode context
 Exception reporting mode: Context
 
 In [22]: %tb
 ---------------------------------------------------------------------------
 SystemExit                                Traceback (most recent call last)
 <BLANKLINE>
 ...<module>()
      30         mode = 'div'
      31 
 ---> 32     bar(mode)
 <BLANKLINE>
 ...bar(mode)
      20         except:
      21             stat = 1
 ---> 22         sysexit(stat, mode)
      23     else:
      24         raise ValueError('Unknown mode')
 <BLANKLINE>
 ...sysexit(stat, mode)
       9 
      10 def sysexit(stat, mode):
 ---> 11     raise SystemExit(stat, 'Mode = %s' % mode)
      12 
      13 def bar(mode):
 <BLANKLINE>
 SystemExit: (2, {u}'Mode = exit')
 
 In [23]: %xmode verbose
 Exception reporting mode: Verbose
 
 In [24]: %tb
 ---------------------------------------------------------------------------
 SystemExit                                Traceback (most recent call last)
 <BLANKLINE>
 ... in <module>()
      30         mode = 'div'
      31 
 ---> 32     bar(mode)
         global bar = <function bar at ...>
         global mode = {u}'exit'
 <BLANKLINE>
 ... in bar(mode={u}'exit')
      20         except:
      21             stat = 1
 ---> 22         sysexit(stat, mode)
         global sysexit = <function sysexit at ...>
         stat = 2
         mode = {u}'exit'
      23     else:
      24         raise ValueError('Unknown mode')
 <BLANKLINE>
 ... in sysexit(stat=2, mode={u}'exit')
       9 
      10 def sysexit(stat, mode):
 ---> 11     raise SystemExit(stat, 'Mode = %s' % mode)
         global SystemExit = undefined
         stat = 2
         mode = {u}'exit'
      12 
      13 def bar(mode):
 <BLANKLINE>
 SystemExit: (2, {u}'Mode = exit')
     """
 
 
 def test_run_cell():
     import textwrap
     ip.run_cell('a = 10\na+=1')
     ip.run_cell('assert a == 11\nassert 1')
 
     nt.assert_equals(ip.user_ns['a'], 11)
     complex = textwrap.dedent("""
     if 1:
         print "hello"
         if 1:
             print "world"
         
     if 2:
         print "foo"
 
     if 3:
         print "bar"
 
     if 4:
         print "bar"
     
     """)
     # Simply verifies that this kind of input is run
     ip.run_cell(complex)
     
 
 def test_db():
     """Test the internal database used for variable persistence."""
     ip.db['__unittest_'] = 12
     nt.assert_equals(ip.db['__unittest_'], 12)
     del ip.db['__unittest_']
     assert '__unittest_' not in ip.db

IPython/core/tests/test_magic.py

@@ -1,467 +1,461 b''
 """Tests for various magic functions.
 
 Needs to be run by nose (to make ipython session available).
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 import tempfile
 import types
 from StringIO import StringIO
 
 import nose.tools as nt
 
 from IPython.utils.path import get_long_path_name
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 def test_rehashx():
     # clear up everything
     _ip = get_ipython()
     _ip.alias_manager.alias_table.clear()
     del _ip.db['syscmdlist']
     
     _ip.magic('rehashx')
     # Practically ALL ipython development systems will have more than 10 aliases
 
     yield (nt.assert_true, len(_ip.alias_manager.alias_table) > 10)
     for key, val in _ip.alias_manager.alias_table.iteritems():
         # we must strip dots from alias names
         nt.assert_true('.' not in key)
 
     # rehashx must fill up syscmdlist
     scoms = _ip.db['syscmdlist']
     yield (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'
     opts = ip.parse_options('-f %s' % path,'f:')[0]
     # argv splitting is os-dependent
     if os.name == 'posix':
         expected = 'c:x'
     else:
         expected = path
     nt.assert_equals(opts['f'], expected)
 
     
 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)
     """
 
 
 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
     """
 
 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'
     >>> 
     """
   
 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.
     assert "test" in ip.magic("macro")
 
 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)
     nt.assert_equal(ip.user_ns["test"].value,
                             py3compat.doctest_refactor_print("a+=1\nprint a\n"))
     original_stdout = sys.stdout
     new_stdout = StringIO()
     sys.stdout = new_stdout
     try:
         ip.run_cell("test")
         nt.assert_true("12" in new_stdout.getvalue())
         ip.run_cell("test")
         nt.assert_true("13" in new_stdout.getvalue())
     finally:
         sys.stdout = original_stdout
         new_stdout.close()
 
     
 # XXX failing for now, until we get clearcmd out of quarantine.  But we should
 # fix this and revert the skip to happen only if numpy is not around.
 #@dec.skipif_not_numpy
 @dec.skip_known_failure
 def test_numpy_clear_array_undec():
     from IPython.extensions import clearcmd
 
     _ip.ex('import numpy as np')
     _ip.ex('a = np.empty(2)')
     yield (nt.assert_true, 'a' in _ip.user_ns)
     _ip.magic('clear array')
     yield (nt.assert_false, 'a' in _ip.user_ns)
     
 
 # Multiple tests for clipboard pasting
 @dec.parametric
 def test_paste():
     _ip = get_ipython()
     def paste(txt, flags='-q'):
         """Paste input text, by default in quiet mode"""
         hooks.clipboard_get = lambda : txt
         _ip.magic('paste '+flags)
 
     # Inject fake clipboard hook but save original so we can restore it later
     hooks = _ip.hooks
     user_ns = _ip.user_ns
     original_clip = hooks.clipboard_get
 
     try:
-        # This try/except with an emtpy except clause is here only because
-        # try/yield/finally is invalid syntax in Python 2.4.  This will be
-        # removed when we drop 2.4-compatibility, and the emtpy except below
-        # will be changed to a finally.
-
         # Run tests with fake clipboard function
         user_ns.pop('x', None)
         paste('x=1')
         yield nt.assert_equal(user_ns['x'], 1)
 
         user_ns.pop('x', None)
         paste('>>> x=2')
         yield nt.assert_equal(user_ns['x'], 2)
 
         paste("""
         >>> x = [1,2,3]
         >>> y = []
         >>> for i in x:
         ...     y.append(i**2)
         ...
         """)
         yield nt.assert_equal(user_ns['x'], [1,2,3])
         yield nt.assert_equal(user_ns['y'], [1,4,9])
 
         # Now, test that paste -r works
         user_ns.pop('x', None)
         yield nt.assert_false('x' in user_ns)
         _ip.magic('paste -r')
         yield nt.assert_equal(user_ns['x'], [1,2,3])
 
         # Also test paste echoing, by temporarily faking the writer
         w = StringIO()
         writer = _ip.write
         _ip.write = w.write
         code = """
         a = 100
         b = 200"""
         try:
             paste(code,'')
             out = w.getvalue()
         finally:
             _ip.write = writer
         yield nt.assert_equal(user_ns['a'], 100)
         yield nt.assert_equal(user_ns['b'], 200)
         yield nt.assert_equal(out, code+"\n## -- End pasted text --\n")
         
     finally:
-        # This should be in a finally clause, instead of the bare except above.
         # Restore original hook
         hooks.clipboard_get = original_clip
 
 
 def test_time():
     _ip.magic('time None')
 
 
 @py3compat.doctest_refactor_print
 def doctest_time():
     """
     In [10]: %time None
     CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
     Wall time: 0.00 s
     
     In [11]: def f(kmjy):
        ....:    %time print 2*kmjy
        
     In [12]: f(3)
     6
     CPU times: user 0.00 s, sys: 0.00 s, total: 0.00 s
     Wall time: 0.00 s
     """
 
 
 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.
     nt.assert_equal(_ip.parse_options('foo', '')[1], 'foo')
     nt.assert_equal(_ip.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 = _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 check_cpaste(code, should_fail=False):
     """Execute code via 'cpaste' and ensure it was executed, unless
     should_fail is set.
     """
     _ip.user_ns['code_ran'] = False
 
     src = StringIO()
     src.write('\n')
     src.write(code)
     src.write('\n--\n')
     src.seek(0)
 
     stdin_save = sys.stdin
     sys.stdin = src
     
     try:
         _ip.magic('cpaste')
     except:
         if not should_fail:
             raise AssertionError("Failure not expected : '%s'" %
                                  code)
     else:
         assert _ip.user_ns['code_ran']
         if should_fail:
             raise AssertionError("Failure expected : '%s'" % code)
     finally:
         sys.stdin = stdin_save
 
 
 def test_cpaste():
     """Test cpaste magic"""
 
     def run():
         """Marker function: sets a flag when executed.
         """
         _ip.user_ns['code_ran'] = True
         return 'run' # return string so '+ run()' doesn't result in success
 
     tests = {'pass': ["> > > run()",
                       ">>> > run()",
                       "+++ run()",
                       "++ run()",
                       "  >>> run()"],
 
              'fail': ["+ + run()",
                       " ++ run()"]}
 
     _ip.user_ns['run'] = run
 
     for code in tests['pass']:
         check_cpaste(code)
 
     for code in tests['fail']:
         check_cpaste(code, should_fail=True)
 
 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']
     """
 
 @py3compat.u_format
 def doctest_precision():
     """doctest for %precision
     
     In [1]: f = get_ipython().shell.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'
     """
 

IPython/core/tests/test_run.py

@@ -1,210 +1,210 b''
 """Tests for code execution (%run and related), which is particularly tricky.
 
 Because of how %run manages namespaces, and the fact that we are trying here to
 verify subtle object deletion and reference counting issues, the %run tests
 will be kept in this separate file.  This makes it easier to aggregate in one
 place the tricks needed to handle it; most other magics are much easier to test
 and we do so in a common test_magic file.
 """
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 import tempfile
 
 import nose.tools as nt
 from nose import SkipTest
 
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
-from IPython.utils.py3compat import doctest_refactor_print
+from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Test functions begin
 #-----------------------------------------------------------------------------
 
 def doctest_refbug():
     """Very nasty problem with references held by multiple runs of a script.
     See: https://github.com/ipython/ipython/issues/141
 
     In [1]: _ip.clear_main_mod_cache()
     # random
 
     In [2]: %run refbug
 
     In [3]: call_f()
     lowercased: hello
 
     In [4]: %run refbug
 
     In [5]: call_f()
     lowercased: hello
     lowercased: hello
     """
 
 
 def doctest_run_builtins():
     r"""Check that %run doesn't damage __builtins__.
 
     In [1]: import tempfile
 
     In [2]: bid1 = id(__builtins__)
 
     In [3]: fname = tempfile.mkstemp('.py')[1]
 
     In [3]: f = open(fname,'w')
 
     In [4]: dummy= f.write('pass\n')
 
     In [5]: f.flush()
 
     In [6]: t1 = type(__builtins__)
 
     In [7]: %run $fname
 
     In [7]: f.close()
 
     In [8]: bid2 = id(__builtins__)
 
     In [9]: t2 = type(__builtins__)
 
     In [10]: t1 == t2
     Out[10]: True
 
     In [10]: bid1 == bid2
     Out[10]: True
 
     In [12]: try:
        ....:     os.unlink(fname)
        ....: except:
        ....:     pass
        ....:
     """
 
-@doctest_refactor_print
[e-mail]_refactor_print
 def doctest_reset_del():
     """Test that resetting doesn't cause errors in __del__ methods.
 
     In [2]: class A(object):
        ...:     def __del__(self):
        ...:         print str("Hi")
        ...:
 
     In [3]: a = A()
 
     In [4]: get_ipython().reset()
     Hi
 
     In [5]: 1+1
     Out[5]: 2
     """
 
 # For some tests, it will be handy to organize them in a class with a common
 # setup that makes a temp file
 
 class TestMagicRunPass(tt.TempFileMixin):
 
     def setup(self):
         """Make a valid python temp file."""
         self.mktmp('pass\n')
 
     def run_tmpfile(self):
         _ip = get_ipython()
         # This fails on Windows if self.tmpfile.name has spaces or "~" in it.
         # See below and ticket https://bugs.launchpad.net/bugs/366353
         _ip.magic('run %s' % self.fname)
 
     def test_builtins_id(self):
         """Check that %run doesn't damage __builtins__ """
         _ip = get_ipython()
         # Test that the id of __builtins__ is not modified by %run
         bid1 = id(_ip.user_ns['__builtins__'])
         self.run_tmpfile()
         bid2 = id(_ip.user_ns['__builtins__'])
         tt.assert_equals(bid1, bid2)
 
     def test_builtins_type(self):
         """Check that the type of __builtins__ doesn't change with %run.
 
         However, the above could pass if __builtins__ was already modified to
         be a dict (it should be a module) by a previous use of %run.  So we
         also check explicitly that it really is a module:
         """
         _ip = get_ipython()
         self.run_tmpfile()
         tt.assert_equals(type(_ip.user_ns['__builtins__']),type(sys))
 
     def test_prompts(self):
         """Test that prompts correctly generate after %run"""
         self.run_tmpfile()
         _ip = get_ipython()
         p2 = str(_ip.displayhook.prompt2).strip()
         nt.assert_equals(p2[:3], '...')
 
 
 class TestMagicRunSimple(tt.TempFileMixin):
 
     def test_simpledef(self):
         """Test that simple class definitions work."""
         src = ("class foo: pass\n"
                "def f(): return foo()")
         self.mktmp(src)
         _ip.magic('run %s' % self.fname)
         _ip.run_cell('t = isinstance(f(), foo)')
         nt.assert_true(_ip.user_ns['t'])
 
     def test_obj_del(self):
         """Test that object's __del__ methods are called on exit."""
         if sys.platform == 'win32':
             try:
                 import win32api
             except ImportError:
                 raise SkipTest("Test requires pywin32")
         src = ("class A(object):\n"
                "    def __del__(self):\n"
                "        print 'object A deleted'\n"
                "a = A()\n")
-        self.mktmp(src)
+        self.mktmp(py3compat.doctest_refactor_print(src))
         tt.ipexec_validate(self.fname, 'object A deleted')
 
     @dec.skip_known_failure
     def test_aggressive_namespace_cleanup(self):
         """Test that namespace cleanup is not too aggressive GH-238
 
         Returning from another run magic deletes the namespace"""
         # see ticket https://github.com/ipython/ipython/issues/238
         class secondtmp(tt.TempFileMixin): pass
         empty = secondtmp()
         empty.mktmp('')
         src = ("ip = get_ipython()\n"
                "for i in range(5):\n"
                "   try:\n"
                "       ip.magic('run %s')\n"
                "   except NameError, e:\n"
                "       print i;break\n" % empty.fname)
-        self.mktmp(src)
+        self.mktmp(py3compat.doctest_refactor_print(src))
         _ip.magic('run %s' % self.fname)
         _ip.run_cell('ip == get_ipython()')
         tt.assert_equals(_ip.user_ns['i'], 5)
 
     @dec.skip_win32
     def test_tclass(self):
         mydir = os.path.dirname(__file__)
         tc = os.path.join(mydir, 'tclass')
         src = ("%%run '%s' C-first\n"
                "%%run '%s' C-second\n"
                "%%run '%s' C-third\n") % (tc, tc, tc)
         self.mktmp(src, '.ipy')
         out = """\
-ARGV 1-: [u'C-first']
-ARGV 1-: [u'C-second']
+ARGV 1-: [{u}'C-first']
+ARGV 1-: [{u}'C-second']
 tclass.py: deleting object: C-first
-ARGV 1-: [u'C-third']
+ARGV 1-: [{u}'C-third']
 tclass.py: deleting object: C-second
 tclass.py: deleting object: C-third
 """
-        tt.ipexec_validate(self.fname, out)
+        tt.ipexec_validate(self.fname, py3compat.u_format(out))

IPython/testing/_paramtestpy3.py

@@ -1,69 +1,71 b''
 """Implementation of the parametric test support for Python 3.x.
 
 Thanks for the py3 version to Robert Collins, from the Testing in Python
 mailing list.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import unittest
 from unittest import TestSuite
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 
 def isgenerator(func):
     return hasattr(func,'_generator')
 
 
 class IterCallableSuite(TestSuite):
    def __init__(self, iterator, adapter):
        self._iter = iterator
        self._adapter = adapter
    def __iter__(self):
        yield self._adapter(self._iter.__next__)
 
 class ParametricTestCase(unittest.TestCase):
    """Write parametric tests in normal unittest testcase form.
 
    Limitations: the last iteration misses printing out a newline when
    running in verbose mode.
    """
 
    def run(self, result=None):
        testMethod = getattr(self, self._testMethodName)
        # For normal tests, we just call the base class and return that
        if isgenerator(testMethod):
            def adapter(next_test):
-               return unittest.FunctionTestCase(next_test,
-                                                self.setUp,
-                                                self.tearDown)
+               ftc = unittest.FunctionTestCase(next_test,
+                                               self.setUp,
+                                               self.tearDown)
+               self._nose_case = ftc   # Nose 1.0 rejects the test without this
+               return ftc
 
            return IterCallableSuite(testMethod(),adapter).run(result)
        else:
            return super(ParametricTestCase, self).run(result)
 
 
 def parametric(func):
    """Decorator to make a simple function into a normal test via
 unittest."""
    # Hack, until I figure out how to write isgenerator() for python3!!
    func._generator = True
 
    class Tester(ParametricTestCase):
        test = staticmethod(func)
 
    Tester.__name__ = func.__name__
 
    return Tester

IPython/testing/tools.py

@@ -1,345 +1,346 b''
 """Generic testing tools that do NOT depend on Twisted.
 
 In particular, this module exposes a set of top-level assert* functions that
 can be used in place of nose.tools.assert* in method generators (the ones in
 nose can not, at least as of nose 0.10.4).
 
 Note: our testing package contains testing.util, which does depend on Twisted
 and provides utilities for tests that manage Deferreds.  All testing support
 tools that only depend on nose, IPython or the standard library should go here
 instead.
 
 
 Authors
 -------
 - Fernando Perez <[e-mail]>
 """
 
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import re
 import sys
 import tempfile
 
 from contextlib import contextmanager
 
 try:
     # These tools are used by parts of the runtime, so we make the nose
     # dependency optional at this point.  Nose is a hard dependency to run the
     # test suite, but NOT to use ipython itself.
     import nose.tools as nt
     has_nose = True
 except ImportError:
     has_nose = False
 
 from IPython.config.loader import Config
 from IPython.utils.process import find_cmd, getoutputerror
 from IPython.utils.text import list_strings
 from IPython.utils.io import temp_pyfile
+from IPython.utils.py3compat import PY3
 
 from . import decorators as dec
 from . import skipdoctest
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # Make a bunch of nose.tools assert wrappers that can be used in test
 # generators.  This will expose an assert* function for each one in nose.tools.
 
 _tpl = """
 def %(name)s(*a,**kw):
     return nt.%(name)s(*a,**kw)
 """
 
 if has_nose:
     for _x in [a for a in dir(nt) if a.startswith('assert')]:
         exec _tpl % dict(name=_x)
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 # The docstring for full_path doctests differently on win32 (different path
 # separator) so just skip the doctest there.  The example remains informative.
 doctest_deco = skipdoctest.skip_doctest if sys.platform == 'win32' else dec.null_deco
 
 @doctest_deco
 def full_path(startPath,files):
     """Make full paths for all the listed files, based on startPath.
 
     Only the base part of startPath is kept, since this routine is typically
     used with a script's __file__ variable as startPath.  The base of startPath
     is then prepended to all the listed files, forming the output list.
 
     Parameters
     ----------
       startPath : string
         Initial path to use as the base for the results.  This path is split
       using os.path.split() and only its first component is kept.
 
       files : string or list
         One or more files.
 
     Examples
     --------
 
     >>> full_path('/foo/bar.py',['a.txt','b.txt'])
     ['/foo/a.txt', '/foo/b.txt']
 
     >>> full_path('/foo',['a.txt','b.txt'])
     ['/a.txt', '/b.txt']
 
     If a single file is given, the output is still a list:
     >>> full_path('/foo','a.txt')
     ['/a.txt']
     """
 
     files = list_strings(files)
     base = os.path.split(startPath)[0]
     return [ os.path.join(base,f) for f in files ]
 
 
 def parse_test_output(txt):
     """Parse the output of a test run and return errors, failures.
 
     Parameters
     ----------
     txt : str
       Text output of a test run, assumed to contain a line of one of the
       following forms::
         'FAILED (errors=1)'
         'FAILED (failures=1)'
         'FAILED (errors=1, failures=1)'
 
     Returns
     -------
     nerr, nfail: number of errors and failures.
     """
 
     err_m = re.search(r'^FAILED \(errors=(\d+)\)', txt, re.MULTILINE)
     if err_m:
         nerr = int(err_m.group(1))
         nfail = 0
         return  nerr, nfail
 
     fail_m = re.search(r'^FAILED \(failures=(\d+)\)', txt, re.MULTILINE)
     if fail_m:
         nerr = 0
         nfail = int(fail_m.group(1))
         return  nerr, nfail
 
     both_m = re.search(r'^FAILED \(errors=(\d+), failures=(\d+)\)', txt,
                        re.MULTILINE)
     if both_m:
         nerr = int(both_m.group(1))
         nfail = int(both_m.group(2))
         return  nerr, nfail
 
     # If the input didn't match any of these forms, assume no error/failures
     return 0, 0
 
 
 # So nose doesn't think this is a test
 parse_test_output.__test__ = False
 
 
 def default_argv():
     """Return a valid default argv for creating testing instances of ipython"""
 
     return ['--quick', # so no config file is loaded
             # Other defaults to minimize side effects on stdout
             '--colors=NoColor', '--no-term-title','--no-banner',
             '--autocall=0']
 
 
 def default_config():
     """Return a config object with good defaults for testing."""
     config = Config()
     config.TerminalInteractiveShell.colors = 'NoColor'
     config.TerminalTerminalInteractiveShell.term_title = False,
     config.TerminalInteractiveShell.autocall = 0
     config.HistoryManager.hist_file = tempfile.mktemp(u'test_hist.sqlite')
     config.HistoryManager.db_cache_size = 10000
     return config
 
 
 def ipexec(fname, options=None):
     """Utility to call 'ipython filename'.
 
     Starts IPython witha minimal and safe configuration to make startup as fast
     as possible.
 
     Note that this starts IPython in a subprocess!
 
     Parameters
     ----------
     fname : str
       Name of file to be executed (should have .py or .ipy extension).
 
     options : optional, list
       Extra command-line flags to be passed to IPython.
 
     Returns
     -------
     (stdout, stderr) of ipython subprocess.
     """
     if options is None: options = []
 
     # For these subprocess calls, eliminate all prompt printing so we only see
     # output from script execution
     prompt_opts = [ '--InteractiveShell.prompt_in1=""',
                     '--InteractiveShell.prompt_in2=""',
                     '--InteractiveShell.prompt_out=""'
     ]
     cmdargs = ' '.join(default_argv() + prompt_opts + options)
 
     _ip = get_ipython()
     test_dir = os.path.dirname(__file__)
 
-    ipython_cmd = find_cmd('ipython')
+    ipython_cmd = find_cmd('ipython3' if PY3 else 'ipython')
     # Absolute path for filename
     full_fname = os.path.join(test_dir, fname)
     full_cmd = '%s %s %s' % (ipython_cmd, cmdargs, full_fname)
     #print >> sys.stderr, 'FULL CMD:', full_cmd # dbg
     out = getoutputerror(full_cmd)
     # `import readline` causes 'ESC[?1034h' to be the first output sometimes,
     # so strip that off the front of the first line if it is found
     if out:
         first = out[0]
         m = re.match(r'\x1b\[[^h]+h', first)
         if m:
             # strip initial readline escape
             out = list(out)
             out[0] = first[len(m.group()):]
             out = tuple(out)
     return out
 
 
 def ipexec_validate(fname, expected_out, expected_err='',
                     options=None):
     """Utility to call 'ipython filename' and validate output/error.
 
     This function raises an AssertionError if the validation fails.
 
     Note that this starts IPython in a subprocess!
 
     Parameters
     ----------
     fname : str
       Name of the file to be executed (should have .py or .ipy extension).
 
     expected_out : str
       Expected stdout of the process.
 
     expected_err : optional, str
       Expected stderr of the process.
 
     options : optional, list
       Extra command-line flags to be passed to IPython.
 
     Returns
     -------
     None
     """
 
     import nose.tools as nt
 
     out, err = ipexec(fname)
     #print 'OUT', out  # dbg
     #print 'ERR', err  # dbg
     # If there are any errors, we must check those befor stdout, as they may be
     # more informative than simply having an empty stdout.
     if err:
         if expected_err:
             nt.assert_equals(err.strip(), expected_err.strip())
         else:
             raise ValueError('Running file %r produced error: %r' %
                              (fname, err))
     # If no errors or output on stderr was expected, match stdout
     nt.assert_equals(out.strip(), expected_out.strip())
 
 
 class TempFileMixin(object):
     """Utility class to create temporary Python/IPython files.
 
     Meant as a mixin class for test cases."""
 
     def mktmp(self, src, ext='.py'):
         """Make a valid python temp file."""
         fname, f = temp_pyfile(src, ext)
         self.tmpfile = f
         self.fname = fname
 
     def tearDown(self):
         if hasattr(self, 'tmpfile'):
             # If the tmpfile wasn't made because of skipped tests, like in
             # win32, there's nothing to cleanup.
             self.tmpfile.close()
             try:
                 os.unlink(self.fname)
             except:
                 # On Windows, even though we close the file, we still can't
                 # delete it.  I have no clue why
                 pass
 
 pair_fail_msg = ("Testing {0}\n\n"
                 "In:\n"
                 "  {1!r}\n"
                 "Expected:\n"
                 "  {2!r}\n"
                 "Got:\n"
                 "  {3!r}\n")
 def check_pairs(func, pairs):
     """Utility function for the common case of checking a function with a
     sequence of input/output pairs.
 
     Parameters
     ----------
     func : callable
       The function to be tested. Should accept a single argument.
     pairs : iterable
       A list of (input, expected_output) tuples.
 
     Returns
     -------
     None. Raises an AssertionError if any output does not match the expected
     value.
     """
     name = getattr(func, "func_name", getattr(func, "__name__", "<unknown>"))
     for inp, expected in pairs:
         out = func(inp)
         assert out == expected, pair_fail_msg.format(name, inp, expected, out)
 
 @contextmanager
 def mute_warn():
     from IPython.utils import warn
     save_warn = warn.warn
     warn.warn = lambda *a, **kw: None
     try:
         yield
     finally:
         warn.warn = save_warn
 
 @contextmanager
 def make_tempfile(name):
     """ Create an empty, named, temporary file for the duration of the context.
     """
     f = open(name, 'w')
     f.close()
     try:
         yield
     finally:
         os.unlink(name)

IPython/utils/_process_posix.py

@@ -1,191 +1,193 b''
 """Posix-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import subprocess as sp
 import sys
 
 from IPython.external import pexpect
 
 # Our own
 from .autoattr import auto_attr
 from ._process_common import getoutput
 from IPython.utils import text
+from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 def _find_cmd(cmd):
     """Find the full path to a command using which."""
 
-    return sp.Popen(['/usr/bin/env', 'which', cmd],
+    path = sp.Popen(['/usr/bin/env', 'which', cmd],
                     stdout=sp.PIPE).communicate()[0]
+    return py3compat.bytes_to_str(path)
 
 
 class ProcessHandler(object):
     """Execute subprocesses under the control of pexpect.
     """
     # Timeout in seconds to wait on each reading of the subprocess' output.
     # This should not be set too low to avoid cpu overusage from our side,
     # since we read in a loop whose period is controlled by this timeout.
     read_timeout = 0.05
 
     # Timeout to give a process if we receive SIGINT, between sending the
     # SIGINT to the process and forcefully terminating it.
     terminate_timeout = 0.2
 
     # File object where stdout and stderr of the subprocess will be written
     logfile = None
 
     # Shell to call for subprocesses to execute
     sh = None
 
     @auto_attr
     def sh(self):
         sh = pexpect.which('sh')
         if sh is None:
             raise OSError('"sh" shell not found')
         return sh
 
     def __init__(self, logfile=None, read_timeout=None, terminate_timeout=None):
         """Arguments are used for pexpect calls."""
         self.read_timeout = (ProcessHandler.read_timeout if read_timeout is
                              None else read_timeout)
         self.terminate_timeout = (ProcessHandler.terminate_timeout if
                                   terminate_timeout is None else
                                   terminate_timeout)
         self.logfile = sys.stdout if logfile is None else logfile
 
     def getoutput(self, cmd):
         """Run a command and return its stdout/stderr as a string.
 
         Parameters
         ----------
         cmd : str
           A command to be executed in the system shell.
 
         Returns
         -------
         output : str
           A string containing the combination of stdout and stderr from the
         subprocess, in whatever order the subprocess originally wrote to its
         file descriptors (so the order of the information in this string is the
         correct order as would be seen if running the command in a terminal).
         """
         pcmd = self._make_cmd(cmd)
         try:
             return pexpect.run(pcmd).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def getoutput_pexpect(self, cmd):
         """Run a command and return its stdout/stderr as a string.
 
         Parameters
         ----------
         cmd : str
           A command to be executed in the system shell.
 
         Returns
         -------
         output : str
           A string containing the combination of stdout and stderr from the
         subprocess, in whatever order the subprocess originally wrote to its
         file descriptors (so the order of the information in this string is the
         correct order as would be seen if running the command in a terminal).
         """
         pcmd = self._make_cmd(cmd)
         try:
             return pexpect.run(pcmd).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def system(self, cmd):
         """Execute a command in a subshell.
 
         Parameters
         ----------
         cmd : str
           A command to be executed in the system shell.
 
         Returns
         -------
         int : child's exitstatus
         """
         # Get likely encoding for the output.
         enc = text.getdefaultencoding()
         
         pcmd = self._make_cmd(cmd)
         # Patterns to match on the output, for pexpect.  We read input and
         # allow either a short timeout or EOF
         patterns = [pexpect.TIMEOUT, pexpect.EOF]
         # the index of the EOF pattern in the list.
         EOF_index = 1  # Fix this index if you change the list!!
         # The size of the output stored so far in the process output buffer.
         # Since pexpect only appends to this buffer, each time we print we
         # record how far we've printed, so that next time we only print *new*
         # content from the buffer.
         out_size = 0
         try:
             # Since we're not really searching the buffer for text patterns, we
             # can set pexpect's search window to be tiny and it won't matter.
             # We only search for the 'patterns' timeout or EOF, which aren't in
             # the text itself.
             #child = pexpect.spawn(pcmd, searchwindowsize=1)
             child = pexpect.spawn(pcmd)
             flush = sys.stdout.flush
             while True:
                 # res is the index of the pattern that caused the match, so we
                 # know whether we've finished (if we matched EOF) or not
                 res_idx = child.expect_list(patterns, self.read_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 flush()
                 if res_idx==EOF_index:
                     break
                 # Update the pointer to what we've already printed
                 out_size = len(child.before)
         except KeyboardInterrupt:
             # We need to send ^C to the process.  The ascii code for '^C' is 3
             # (the character is known as ETX for 'End of Text', see
             # curses.ascii.ETX).
             child.sendline(chr(3))
             # Read and print any more output the program might produce on its
             # way out.
             try:
                 out_size = len(child.before)
                 child.expect_list(patterns, self.terminate_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 sys.stdout.flush()
             except KeyboardInterrupt:
                 # Impatient users tend to type it multiple times
                 pass
             finally:
                 # Ensure the subprocess really is terminated
                 child.terminate(force=True)
         return child.exitstatus
 
     def _make_cmd(self, cmd):
         return '%s -c "%s"' % (self.sh, cmd)
 
 
 # Make system() with a functional interface for outside use.  Note that we use
 # getoutput() from the _common utils, which is built on top of popen(). Using
 # pexpect to get subprocess output produces difficult to parse output, since
 # programs think they are talking to a tty and produce highly formatted output
 # (ls is a good example) that makes them hard.
 system = ProcessHandler().system

IPython/utils/pickleshare.py

@@ -1,362 +1,362 b''
 #!/usr/bin/env python
 
 """ PickleShare - a small 'shelve' like datastore with concurrency support
 
 Like shelve, a PickleShareDB object acts like a normal dictionary. Unlike
 shelve, many processes can access the database simultaneously. Changing a
 value in database is immediately visible to other processes accessing the
 same database.
 
 Concurrency is possible because the values are stored in separate files. Hence
 the "database" is a directory where *all* files are governed by PickleShare.
 
 Example usage::
 
     from pickleshare import *
     db = PickleShareDB('~/testpickleshare')
     db.clear()
     print "Should be empty:",db.items()
     db['hello'] = 15
     db['aku ankka'] = [1,2,313]
     db['paths/are/ok/key'] = [1,(5,46)]
     print db.keys()
     del db['aku ankka']
 
 This module is certainly not ZODB, but can be used for low-load
 (non-mission-critical) situations where tiny code size trumps the
 advanced features of a "real" object database.
 
 Installation guide: easy_install pickleshare
 
 Author: Ville Vainio <[e-mail]>
 License: MIT open source license.
 
 """
 
 from IPython.external.path import path as Path
 import os,stat,time
 import collections
 import cPickle as pickle
 import glob
 
 def gethashfile(key):
     return ("%02x" % abs(hash(key) % 256))[-2:]
 
 _sentinel = object()
 
 class PickleShareDB(collections.MutableMapping):
     """ The main 'connection' object for PickleShare database """
     def __init__(self,root):
         """ Return a db object that will manage the specied directory"""
         self.root = Path(root).expanduser().abspath()
         if not self.root.isdir():
             self.root.makedirs()
         # cache has { 'key' : (obj, orig_mod_time) }
         self.cache = {}
 
 
     def __getitem__(self,key):
         """ db['key'] reading """
         fil = self.root / key
         try:
             mtime = (fil.stat()[stat.ST_MTIME])
         except OSError:
             raise KeyError(key)
 
         if fil in self.cache and mtime == self.cache[fil][1]:
             return self.cache[fil][0]
         try:
             # The cached item has expired, need to read
-            obj = pickle.load(fil.open())
+            obj = pickle.load(fil.open("rb"))
         except:
             raise KeyError(key)
 
         self.cache[fil] = (obj,mtime)
         return obj
 
     def __setitem__(self,key,value):
         """ db['key'] = 5 """
         fil = self.root / key
         parent = fil.parent
         if parent and not parent.isdir():
             parent.makedirs()
-        pickled = pickle.dump(value,fil.open('w'))
+        pickled = pickle.dump(value,fil.open('wb'))
         try:
             self.cache[fil] = (value,fil.mtime)
         except OSError,e:
             if e.errno != 2:
                 raise
 
     def hset(self, hashroot, key, value):
         """ hashed set """
         hroot = self.root / hashroot
         if not hroot.isdir():
             hroot.makedirs()
         hfile = hroot / gethashfile(key)
         d = self.get(hfile, {})
         d.update( {key : value})
         self[hfile] = d
 
 
 
     def hget(self, hashroot, key, default = _sentinel, fast_only = True):
         """ hashed get """
         hroot = self.root / hashroot
         hfile = hroot / gethashfile(key)
 
         d = self.get(hfile, _sentinel )
         #print "got dict",d,"from",hfile
         if d is _sentinel:
             if fast_only:
                 if default is _sentinel:
                     raise KeyError(key)
 
                 return default
 
             # slow mode ok, works even after hcompress()
             d = self.hdict(hashroot)
 
         return d.get(key, default)
 
     def hdict(self, hashroot):
         """ Get all data contained in hashed category 'hashroot' as dict """
         hfiles = self.keys(hashroot + "/*")
         hfiles.sort()
         last = len(hfiles) and hfiles[-1] or ''
         if last.endswith('xx'):
             # print "using xx"
             hfiles = [last] + hfiles[:-1]
 
         all = {}
 
         for f in hfiles:
             # print "using",f
             try:
                 all.update(self[f])
             except KeyError:
                 print "Corrupt",f,"deleted - hset is not threadsafe!"
                 del self[f]
 
             self.uncache(f)
 
         return all
 
     def hcompress(self, hashroot):
         """ Compress category 'hashroot', so hset is fast again
 
         hget will fail if fast_only is True for compressed items (that were
         hset before hcompress).
 
         """
         hfiles = self.keys(hashroot + "/*")
         all = {}
         for f in hfiles:
             # print "using",f
             all.update(self[f])
             self.uncache(f)
 
         self[hashroot + '/xx'] = all
         for f in hfiles:
             p = self.root / f
             if p.basename() == 'xx':
                 continue
             p.remove()
 
 
 
     def __delitem__(self,key):
         """ del db["key"] """
         fil = self.root / key
         self.cache.pop(fil,None)
         try:
             fil.remove()
         except OSError:
             # notfound and permission denied are ok - we
             # lost, the other process wins the conflict
             pass
 
     def _normalized(self, p):
         """ Make a key suitable for user's eyes """
         return str(self.root.relpathto(p)).replace('\\','/')
 
     def keys(self, globpat = None):
         """ All keys in DB, or all keys matching a glob"""
 
         if globpat is None:
             files = self.root.walkfiles()
         else:
             files = [Path(p) for p in glob.glob(self.root/globpat)]
         return [self._normalized(p) for p in files if p.isfile()]
 
     def __iter__(self):
         return iter(keys)
 
     def __len__(self):
         return len(keys)
 
     def uncache(self,*items):
         """ Removes all, or specified items from cache
 
         Use this after reading a large amount of large objects
         to free up memory, when you won't be needing the objects
         for a while.
 
         """
         if not items:
             self.cache = {}
         for it in items:
             self.cache.pop(it,None)
 
     def waitget(self,key, maxwaittime = 60 ):
         """ Wait (poll) for a key to get a value
 
         Will wait for `maxwaittime` seconds before raising a KeyError.
         The call exits normally if the `key` field in db gets a value
         within the timeout period.
 
         Use this for synchronizing different processes or for ensuring
         that an unfortunately timed "db['key'] = newvalue" operation
         in another process (which causes all 'get' operation to cause a
         KeyError for the duration of pickling) won't screw up your program
         logic.
         """
 
         wtimes = [0.2] * 3 + [0.5] * 2 + [1]
         tries = 0
         waited = 0
         while 1:
             try:
                 val = self[key]
                 return val
             except KeyError:
                 pass
 
             if waited > maxwaittime:
                 raise KeyError(key)
 
             time.sleep(wtimes[tries])
             waited+=wtimes[tries]
             if tries < len(wtimes) -1:
                 tries+=1
 
     def getlink(self,folder):
         """ Get a convenient link for accessing items  """
         return PickleShareLink(self, folder)
 
     def __repr__(self):
         return "PickleShareDB('%s')" % self.root
 
 
 
 class PickleShareLink:
     """ A shortdand for accessing nested PickleShare data conveniently.
 
     Created through PickleShareDB.getlink(), example::
 
         lnk = db.getlink('myobjects/test')
         lnk.foo = 2
         lnk.bar = lnk.foo + 5
 
     """
     def __init__(self, db, keydir ):
         self.__dict__.update(locals())
 
     def __getattr__(self,key):
         return self.__dict__['db'][self.__dict__['keydir']+'/' + key]
     def __setattr__(self,key,val):
         self.db[self.keydir+'/' + key] = val
     def __repr__(self):
         db = self.__dict__['db']
         keys = db.keys( self.__dict__['keydir'] +"/*")
         return "<PickleShareLink '%s': %s>" % (
             self.__dict__['keydir'],
             ";".join([Path(k).basename() for k in keys]))
 
 
 def test():
     db = PickleShareDB('~/testpickleshare')
     db.clear()
     print "Should be empty:",db.items()
     db['hello'] = 15
     db['aku ankka'] = [1,2,313]
     db['paths/nest/ok/keyname'] = [1,(5,46)]
     db.hset('hash', 'aku', 12)
     db.hset('hash', 'ankka', 313)
     print "12 =",db.hget('hash','aku')
     print "313 =",db.hget('hash','ankka')
     print "all hashed",db.hdict('hash')
     print db.keys()
     print db.keys('paths/nest/ok/k*')
     print dict(db) # snapsot of whole db
     db.uncache() # frees memory, causes re-reads later
 
     # shorthand for accessing deeply nested files
     lnk = db.getlink('myobjects/test')
     lnk.foo = 2
     lnk.bar = lnk.foo + 5
     print lnk.bar # 7
 
 def stress():
     db = PickleShareDB('~/fsdbtest')
     import time,sys
     for i in range(1000):
         for j in range(1000):
             if i % 15 == 0 and i < 200:
                 if str(j) in db:
                     del db[str(j)]
                 continue
 
             if j%33 == 0:
                 time.sleep(0.02)
 
             db[str(j)] = db.get(str(j), []) + [(i,j,"proc %d" % os.getpid())]
             db.hset('hash',j, db.hget('hash',j,15) + 1 )
 
         print i,
         sys.stdout.flush()
         if i % 10 == 0:
             db.uncache()
 
 def main():
     import textwrap
     usage = textwrap.dedent("""\
     pickleshare - manage PickleShare databases
 
     Usage:
 
         pickleshare dump /path/to/db > dump.txt
         pickleshare load /path/to/db < dump.txt
         pickleshare test /path/to/db
     """)
     DB = PickleShareDB
     import sys
     if len(sys.argv) < 2:
         print usage
         return
 
     cmd = sys.argv[1]
     args = sys.argv[2:]
     if cmd == 'dump':
         if not args: args= ['.']
         db = DB(args[0])
         import pprint
         pprint.pprint(db.items())
     elif cmd == 'load':
         cont = sys.stdin.read()
         db = DB(args[0])
         data = eval(cont)
         db.clear()
         for k,v in db.items():
             db[k] = v
     elif cmd == 'testwait':
         db = DB(args[0])
         db.clear()
         print db.waitget('250')
     elif cmd == 'test':
         test()
         stress()
 
 if __name__== "__main__":
     main()