Revised input transformation framework.

IPython/core/inputsplitter.py

@@ -1,651 +1,712 b''
 """Analysis of text input into executable blocks.
 
 The main class in this module, :class:`InputSplitter`, is designed to break
 input from either interactive, line-by-line environments or block-based ones,
 into standalone blocks that can be executed by Python as 'single' statements
 (thus triggering sys.displayhook).
 
 A companion, :class:`IPythonInputSplitter`, provides the same functionality but
 with full support for the extended IPython syntax (magics, system calls, etc).
 
 For more details, see the class docstring below.
 
 Syntax Transformations
 ----------------------
 
 One of the main jobs of the code in this file is to apply all syntax
 transformations that make up 'the IPython language', i.e. magics, shell
 escapes, etc.  All transformations should be implemented as *fully stateless*
 entities, that simply take one line as their input and return a line.
 Internally for implementation purposes they may be a normal function or a
 callable object, but the only input they receive will be a single line and they
 should only return a line, without holding any data-dependent state between
 calls.
 
 As an example, the EscapedTransformer is a class so we can more clearly group
 together the functionality of dispatching to individual functions based on the
 starting escape character, but the only method for public use is its call
 method.
 
 
 ToDo
 ----
 
 - Should we make push() actually raise an exception once push_accepts_more()
   returns False?
 
 - Naming cleanups.  The tr_* names aren't the most elegant, though now they are
   at least just attributes of a class so not really very exposed.
 
 - Think about the best way to support dynamic things: automagic, autocall,
   macros, etc.
 
 - Think of a better heuristic for the application of the transforms in
   IPythonInputSplitter.push() than looking at the buffer ending in ':'.  Idea:
   track indentation change events (indent, dedent, nothing) and apply them only
   if the indentation went up, but not otherwise.
 
 - Think of the cleanest way for supporting user-specified transformations (the
   user prefilters we had before).
 
 Authors
 -------
 
 * Fernando Perez
 * Brian Granger
 """
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 # stdlib
 import ast
 import codeop
 import re
 import sys
 
 # IPython modules
 from IPython.core.splitinput import split_user_input, LineInfo
 from IPython.utils.py3compat import cast_unicode
 from IPython.core.inputtransformer import (leading_indent,
                                            classic_prompt,
                                            ipy_prompt,
                                            cellmagic,
                                            assemble_logical_lines,
                                            help_end,
                                            escaped_transformer,
                                            assign_from_magic,
                                            assign_from_system,
+                                           assemble_python_lines,
                                            )
 
 # Temporary!
 from IPython.core.inputtransformer import (ESC_SHELL, ESC_SH_CAP, ESC_HELP,
                                         ESC_HELP2, ESC_MAGIC, ESC_MAGIC2,
                                         ESC_QUOTE, ESC_QUOTE2, ESC_PAREN, ESC_SEQUENCES)
 
 #-----------------------------------------------------------------------------
 # Utilities
 #-----------------------------------------------------------------------------
 
 # FIXME: These are general-purpose utilities that later can be moved to the
 # general ward.  Kept here for now because we're being very strict about test
 # coverage with this code, and this lets us ensure that we keep 100% coverage
 # while developing.
 
 # compiled regexps for autoindent management
 dedent_re = re.compile('|'.join([
     r'^\s+raise(\s.*)?$', # raise statement (+ space + other stuff, maybe)
     r'^\s+raise\([^\)]*\).*$', # wacky raise with immediate open paren
     r'^\s+return(\s.*)?$', # normal return (+ space + other stuff, maybe)
     r'^\s+return\([^\)]*\).*$', # wacky return with immediate open paren
     r'^\s+pass\s*$', # pass (optionally followed by trailing spaces)
     r'^\s+break\s*$', # break (optionally followed by trailing spaces)
     r'^\s+continue\s*$', # continue (optionally followed by trailing spaces)
 ]))
 ini_spaces_re = re.compile(r'^([ \t\r\f\v]+)')
 
 # regexp to match pure comment lines so we don't accidentally insert 'if 1:'
 # before pure comments
 comment_line_re = re.compile('^\s*\#')
 
 
 def num_ini_spaces(s):
     """Return the number of initial spaces in a string.
 
     Note that tabs are counted as a single space.  For now, we do *not* support
     mixing of tabs and spaces in the user's input.
 
     Parameters
     ----------
     s : string
 
     Returns
     -------
     n : int
     """
 
     ini_spaces = ini_spaces_re.match(s)
     if ini_spaces:
         return ini_spaces.end()
     else:
         return 0
 
 def last_blank(src):
     """Determine if the input source ends in a blank.
 
     A blank is either a newline or a line consisting of whitespace.
 
     Parameters
     ----------
     src : string
       A single or multiline string.
     """
     if not src: return False
     ll  = src.splitlines()[-1]
     return (ll == '') or ll.isspace()
 
 
 last_two_blanks_re = re.compile(r'\n\s*\n\s*$', re.MULTILINE)
 last_two_blanks_re2 = re.compile(r'.+\n\s*\n\s+$', re.MULTILINE)
 
 def last_two_blanks(src):
     """Determine if the input source ends in two blanks.
 
     A blank is either a newline or a line consisting of whitespace.
 
     Parameters
     ----------
     src : string
       A single or multiline string.
     """
     if not src: return False
     # The logic here is tricky: I couldn't get a regexp to work and pass all
     # the tests, so I took a different approach: split the source by lines,
     # grab the last two and prepend '###\n' as a stand-in for whatever was in
     # the body before the last two lines.  Then, with that structure, it's
     # possible to analyze with two regexps.  Not the most elegant solution, but
     # it works.  If anyone tries to change this logic, make sure to validate
     # the whole test suite first!
     new_src = '\n'.join(['###\n'] + src.splitlines()[-2:])
     return (bool(last_two_blanks_re.match(new_src)) or
             bool(last_two_blanks_re2.match(new_src)) )
 
 
 def remove_comments(src):
     """Remove all comments from input source.
 
     Note: comments are NOT recognized inside of strings!
 
     Parameters
     ----------
     src : string
       A single or multiline input string.
 
     Returns
     -------
     String with all Python comments removed.
     """
 
     return re.sub('#.*', '', src)
 
 
 def get_input_encoding():
     """Return the default standard input encoding.
 
     If sys.stdin has no encoding, 'ascii' is returned."""
     # There are strange environments for which sys.stdin.encoding is None. We
     # ensure that a valid encoding is returned.
     encoding = getattr(sys.stdin, 'encoding', None)
     if encoding is None:
         encoding = 'ascii'
     return encoding
 
 #-----------------------------------------------------------------------------
 # Classes and functions for normal Python syntax handling
 #-----------------------------------------------------------------------------
 
 class InputSplitter(object):
     """An object that can accumulate lines of Python source before execution.
 
     This object is designed to be fed python source line-by-line, using
        :meth:`push`. It will return on each push whether the currently pushed
        code could be executed already. In addition, it provides a method called
        :meth:`push_accepts_more` that can be used to query whether more input
        can be pushed into a single interactive block.
 
     This is a simple example of how an interactive terminal-based client can use
     this tool::
 
         isp = InputSplitter()
         while isp.push_accepts_more():
             indent = ' '*isp.indent_spaces
             prompt = '>>> ' + indent
             line = indent + raw_input(prompt)
             isp.push(line)
         print 'Input source was:\n', isp.source_reset(),
     """
     # Number of spaces of indentation computed from input that has been pushed
     # so far.  This is the attributes callers should query to get the current
     # indentation level, in order to provide auto-indent facilities.
     indent_spaces = 0
     # String, indicating the default input encoding.  It is computed by default
     # at initialization time via get_input_encoding(), but it can be reset by a
     # client with specific knowledge of the encoding.
     encoding = ''
     # String where the current full source input is stored, properly encoded.
     # Reading this attribute is the normal way of querying the currently pushed
     # source code, that has been properly encoded.
     source = ''
     # Code object corresponding to the current source.  It is automatically
     # synced to the source, so it can be queried at any time to obtain the code
     # object; it will be None if the source doesn't compile to valid Python.
     code = None
     # Input mode
     input_mode = 'line'
 
     # Private attributes
 
     # List with lines of input accumulated so far
     _buffer = None
     # Command compiler
     _compile = None
     # Mark when input has changed indentation all the way back to flush-left
     _full_dedent = False
     # Boolean indicating whether the current block is complete
     _is_complete = None
 
     def __init__(self, input_mode=None):
         """Create a new InputSplitter instance.
 
         Parameters
         ----------
         input_mode : str
 
           One of ['line', 'cell']; default is 'line'.
 
        The input_mode parameter controls how new inputs are used when fed via
        the :meth:`push` method:
 
        - 'line': meant for line-oriented clients, inputs are appended one at a
          time to the internal buffer and the whole buffer is compiled.
 
        - 'cell': meant for clients that can edit multi-line 'cells' of text at
           a time.  A cell can contain one or more blocks that can be compile in
           'single' mode by Python.  In this mode, each new input new input
           completely replaces all prior inputs.  Cell mode is thus equivalent
           to prepending a full reset() to every push() call.
         """
         self._buffer = []
         self._compile = codeop.CommandCompiler()
         self.encoding = get_input_encoding()
         self.input_mode = InputSplitter.input_mode if input_mode is None \
                           else input_mode
 
     def reset(self):
         """Reset the input buffer and associated state."""
         self.indent_spaces = 0
         self._buffer[:] = []
         self.source = ''
         self.code = None
         self._is_complete = False
         self._full_dedent = False
 
     def source_reset(self):
         """Return the input source and perform a full reset.
         """
         out = self.source
         self.reset()
         return out
 
     def push(self, lines):
         """Push one or more lines of input.
 
         This stores the given lines and returns a status code indicating
         whether the code forms a complete Python block or not.
 
         Any exceptions generated in compilation are swallowed, but if an
         exception was produced, the method returns True.
 
         Parameters
         ----------
         lines : string
           One or more lines of Python input.
 
         Returns
         -------
         is_complete : boolean
           True if the current input source (the result of the current input
           plus prior inputs) forms a complete Python execution block.  Note that
           this value is also stored as a private attribute (``_is_complete``), so it
           can be queried at any time.
         """
         if self.input_mode == 'cell':
             self.reset()
 
         self._store(lines)
         source = self.source
 
         # Before calling _compile(), reset the code object to None so that if an
         # exception is raised in compilation, we don't mislead by having
         # inconsistent code/source attributes.
         self.code, self._is_complete = None, None
 
         # Honor termination lines properly
         if source.endswith('\\\n'):
             return False
 
         self._update_indent(lines)
         try:
             self.code = self._compile(source, symbol="exec")
         # Invalid syntax can produce any of a number of different errors from
         # inside the compiler, so we have to catch them all.  Syntax errors
         # immediately produce a 'ready' block, so the invalid Python can be
         # sent to the kernel for evaluation with possible ipython
         # special-syntax conversion.
         except (SyntaxError, OverflowError, ValueError, TypeError,
                 MemoryError):
             self._is_complete = True
         else:
             # Compilation didn't produce any exceptions (though it may not have
             # given a complete code object)
             self._is_complete = self.code is not None
 
         return self._is_complete
 
     def push_accepts_more(self):
         """Return whether a block of interactive input can accept more input.
 
         This method is meant to be used by line-oriented frontends, who need to
         guess whether a block is complete or not based solely on prior and
         current input lines.  The InputSplitter considers it has a complete
         interactive block and will not accept more input only when either a
         SyntaxError is raised, or *all* of the following are true:
 
         1. The input compiles to a complete statement.
 
         2. The indentation level is flush-left (because if we are indented,
            like inside a function definition or for loop, we need to keep
            reading new input).
 
         3. There is one extra line consisting only of whitespace.
 
         Because of condition #3, this method should be used only by
         *line-oriented* frontends, since it means that intermediate blank lines
         are not allowed in function definitions (or any other indented block).
 
         If the current input produces a syntax error, this method immediately
         returns False but does *not* raise the syntax error exception, as
         typically clients will want to send invalid syntax to an execution
         backend which might convert the invalid syntax into valid Python via
         one of the dynamic IPython mechanisms.
         """
 
         # With incomplete input, unconditionally accept more
         if not self._is_complete:
             return True
 
         # If we already have complete input and we're flush left, the answer
         # depends.  In line mode, if there hasn't been any indentation,
         # that's it. If we've come back from some indentation, we need
         # the blank final line to finish.
         # In cell mode, we need to check how many blocks the input so far
         # compiles into, because if there's already more than one full
         # independent block of input, then the client has entered full
         # 'cell' mode and is feeding lines that each is complete.  In this
         # case we should then keep accepting. The Qt terminal-like console
         # does precisely this, to provide the convenience of terminal-like
         # input of single expressions, but allowing the user (with a
         # separate keystroke) to switch to 'cell' mode and type multiple
         # expressions in one shot.
         if self.indent_spaces==0:
             if self.input_mode=='line':
                 if not self._full_dedent:
                     return False
             else:
                 try:
                     code_ast = ast.parse(u''.join(self._buffer))
                 except Exception:
                     return False
                 else:
                     if len(code_ast.body) == 1:
                         return False
 
         # When input is complete, then termination is marked by an extra blank
         # line at the end.
         last_line = self.source.splitlines()[-1]
         return bool(last_line and not last_line.isspace())
 
     #------------------------------------------------------------------------
     # Private interface
     #------------------------------------------------------------------------
 
     def _find_indent(self, line):
         """Compute the new indentation level for a single line.
 
         Parameters
         ----------
         line : str
           A single new line of non-whitespace, non-comment Python input.
 
         Returns
         -------
         indent_spaces : int
           New value for the indent level (it may be equal to self.indent_spaces
         if indentation doesn't change.
 
         full_dedent : boolean
           Whether the new line causes a full flush-left dedent.
         """
         indent_spaces = self.indent_spaces
         full_dedent = self._full_dedent
 
         inisp = num_ini_spaces(line)
         if inisp < indent_spaces:
             indent_spaces = inisp
             if indent_spaces <= 0:
                 #print 'Full dedent in text',self.source # dbg
                 full_dedent = True
 
         if line.rstrip()[-1] == ':':
             indent_spaces += 4
         elif dedent_re.match(line):
             indent_spaces -= 4
             if indent_spaces <= 0:
                 full_dedent = True
 
         # Safety
         if indent_spaces < 0:
             indent_spaces = 0
             #print 'safety' # dbg
 
         return indent_spaces, full_dedent
 
     def _update_indent(self, lines):
         for line in remove_comments(lines).splitlines():
             if line and not line.isspace():
                 self.indent_spaces, self._full_dedent = self._find_indent(line)
 
     def _store(self, lines, buffer=None, store='source'):
         """Store one or more lines of input.
 
         If input lines are not newline-terminated, a newline is automatically
         appended."""
 
         if buffer is None:
             buffer = self._buffer
 
         if lines.endswith('\n'):
             buffer.append(lines)
         else:
             buffer.append(lines+'\n')
         setattr(self, store, self._set_source(buffer))
 
     def _set_source(self, buffer):
         return u''.join(buffer)
 
 
 class IPythonInputSplitter(InputSplitter):
     """An input splitter that recognizes all of IPython's special syntax."""
 
     # String with raw, untransformed input.
     source_raw = ''
     
     # Flag to track when a transformer has stored input that it hasn't given
     # back yet.
     transformer_accumulating = False
+    
+    # Flag to track when assemble_python_lines has stored input that it hasn't
+    # given back yet.
+    within_python_line = False
 
     # Private attributes
 
     # List with lines of raw input accumulated so far.
     _buffer_raw = None
 
-    def __init__(self, input_mode=None, transforms=None):
+    def __init__(self, input_mode=None, physical_line_transforms=None,
+                    logical_line_transforms=None, python_line_transforms=None):
         super(IPythonInputSplitter, self).__init__(input_mode)
         self._buffer_raw = []
         self._validate = True
-        if transforms is not None:
-            self.transforms = transforms
-        else:
-            self.transforms = [leading_indent(),
-                               classic_prompt(),
-                               ipy_prompt(),
-                               cellmagic(),
-                               assemble_logical_lines(),
-                               help_end(),
-                               escaped_transformer(),
-                               assign_from_magic(),
-                               assign_from_system(),
-                              ]
+        
+        self.physical_line_transforms = physical_line_transforms or \
+                [leading_indent(),
+                 classic_prompt(),
+                 ipy_prompt(),
+                ]
+        
+        self.assemble_logical_lines = assemble_logical_lines()
+        self.logical_line_transforms = logical_line_transforms or \
+                [cellmagic(),
+                 help_end(),
+                 escaped_transformer(),
+                 assign_from_magic(),
+                 assign_from_system(),
+                ]
+        
+        self.assemble_python_lines = assemble_python_lines()
+        self.python_line_transforms = python_line_transforms or []
+    
+    @property
+    def transforms(self):
+        "Quick access to all transformers."
+        return self.physical_line_transforms + \
+            [self.assemble_logical_lines] + self.logical_line_transforms + \
+            [self.assemble_python_lines]  + self.python_line_transforms
+    
+    @property
+    def transforms_in_use(self):
+        """Transformers, excluding logical line transformers if we're in a
+        Python line."""
+        t = self.physical_line_transforms + [self.assemble_logical_lines]
+        if not self.within_python_line:
+            t += self.logical_line_transforms
+        return t + [self.assemble_python_lines] + self.python_line_transforms
 
     def reset(self):
         """Reset the input buffer and associated state."""
         super(IPythonInputSplitter, self).reset()
         self._buffer_raw[:] = []
         self.source_raw = ''
         self.transformer_accumulating = False
         for t in self.transforms:
             t.reset()
     
     def flush_transformers(self):
+        def _flush(transform, out):
+            if out is not None:
+                tmp = transform.push(out)
+                return tmp or transform.reset() or None
+            else:
+                return transform.reset() or None
+        
         out = None
-        for t in self.transforms:
-            tmp = t.reset()
-            if tmp:
-                out = tmp
-        if out:
+        for t in self.transforms_in_use:
+            out = _flush(t, out)
+        
+        if out is not None:
             self._store(out)
 
     def source_raw_reset(self):
         """Return input and raw source and perform a full reset.
         """
         self.flush_transformers()
         out = self.source
         out_r = self.source_raw
         self.reset()
         return out, out_r
     
     def source_reset(self):
         self.flush_transformers()
         return super(IPythonInputSplitter, self).source_reset()
 
     def push_accepts_more(self):
         if self.transformer_accumulating:
             return True
         else:
             return super(IPythonInputSplitter, self).push_accepts_more()
 
     def transform_cell(self, cell):
         """Process and translate a cell of input.
         """
         self.reset()
         self.push(cell)
         return self.source_reset()
 
     def push(self, lines):
         """Push one or more lines of IPython input.
 
         This stores the given lines and returns a status code indicating
         whether the code forms a complete Python block or not, after processing
         all input lines for special IPython syntax.
 
         Any exceptions generated in compilation are swallowed, but if an
         exception was produced, the method returns True.
 
         Parameters
         ----------
         lines : string
           One or more lines of Python input.
 
         Returns
         -------
         is_complete : boolean
           True if the current input source (the result of the current input
         plus prior inputs) forms a complete Python execution block.  Note that
         this value is also stored as a private attribute (_is_complete), so it
         can be queried at any time.
         """
 
         # We must ensure all input is pure unicode
         lines = cast_unicode(lines, self.encoding)
         
         # ''.splitlines() --> [], but we need to push the empty line to transformers
         lines_list = lines.splitlines()
         if not lines_list:
             lines_list = ['']
 
         # Transform logic
         #
         # We only apply the line transformers to the input if we have either no
         # input yet, or complete input, or if the last line of the buffer ends
         # with ':' (opening an indented block).  This prevents the accidental
         # transformation of escapes inside multiline expressions like
         # triple-quoted strings or parenthesized expressions.
         #
         # The last heuristic, while ugly, ensures that the first line of an
         # indented block is correctly transformed.
         #
         # FIXME: try to find a cleaner approach for this last bit.
 
         # If we were in 'block' mode, since we're going to pump the parent
         # class by hand line by line, we need to temporarily switch out to
         # 'line' mode, do a single manual reset and then feed the lines one
         # by one.  Note that this only matters if the input has more than one
         # line.
         changed_input_mode = False
 
         if self.input_mode == 'cell':
             self.reset()
             changed_input_mode = True
             saved_input_mode = 'cell'
             self.input_mode = 'line'
 
         # Store raw source before applying any transformations to it.  Note
         # that this must be done *after* the reset() call that would otherwise
         # flush the buffer.
         self._store(lines, self._buffer_raw, 'source_raw')
 
         try:
             for line in lines_list:
                 out = self.push_line(line)
         finally:
             if changed_input_mode:
                 self.input_mode = saved_input_mode
         
         return out
     
     def push_line(self, line):
         buf = self._buffer
-        for transformer in self.transforms:
+        
+        def _accumulating(dbg):
+            #print(dbg)
+            self.transformer_accumulating = True
+            return False
+        
+        for transformer in self.physical_line_transforms:
+            line = transformer.push(line)
+            if line is None:
+                return _accumulating(transformer)
+        
+        line = self.assemble_logical_lines.push(line)
+        if line is None:
+            return _accumulating('acc logical line')
+        
+        if not self.within_python_line:
+            for transformer in self.logical_line_transforms:
+                line = transformer.push(line)
+                if line is None:
+                    return _accumulating(transformer)
+        
+        line = self.assemble_python_lines.push(line)
+        if line is None:
+            self.within_python_line = True
+            return _accumulating('acc python line')
+        else:
+            self.within_python_line = False
+        
+        for transformer in self.python_line_transforms:
             line = transformer.push(line)
             if line is None:
-                self.transformer_accumulating = True
-                return False                
+                return _accumulating(transformer)
 
+        #print("transformers clear") #debug
         self.transformer_accumulating = False
         return super(IPythonInputSplitter, self).push(line)

IPython/core/inputtransformer.py

@@ -1,454 +1,487 b''
 import abc
 import functools
 import re
 from StringIO import StringIO
 import tokenize
 
 try:
     generate_tokens = tokenize.generate_tokens
 except AttributeError:
     # Python 3. Note that we use the undocumented _tokenize because it expects
     # strings, not bytes. See also Python issue #9969.
     generate_tokens = tokenize._tokenize
 
 from IPython.core.splitinput import split_user_input, LineInfo
 from IPython.utils.untokenize import untokenize
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # The escape sequences that define the syntax transformations IPython will
 # apply to user input.  These can NOT be just changed here: many regular
 # expressions and other parts of the code may use their hardcoded values, and
 # for all intents and purposes they constitute the 'IPython syntax', so they
 # should be considered fixed.
 
 ESC_SHELL  = '!'     # Send line to underlying system shell
 ESC_SH_CAP = '!!'    # Send line to system shell and capture output
 ESC_HELP   = '?'     # Find information about object
 ESC_HELP2  = '??'    # Find extra-detailed information about object
 ESC_MAGIC  = '%'     # Call magic function
 ESC_MAGIC2 = '%%'    # Call cell-magic function
 ESC_QUOTE  = ','     # Split args on whitespace, quote each as string and call
 ESC_QUOTE2 = ';'     # Quote all args as a single string, call
 ESC_PAREN  = '/'     # Call first argument with rest of line as arguments
 
 ESC_SEQUENCES = [ESC_SHELL, ESC_SH_CAP, ESC_HELP ,\
                  ESC_HELP2, ESC_MAGIC, ESC_MAGIC2,\
                  ESC_QUOTE, ESC_QUOTE2, ESC_PAREN ]
 
 
 class InputTransformer(object):
     """Abstract base class for line-based input transformers."""
     __metaclass__ = abc.ABCMeta
     
     @abc.abstractmethod
     def push(self, line):
         """Send a line of input to the transformer, returning the transformed
         input or None if the transformer is waiting for more input.
         
         Must be overridden by subclasses.
         """
         pass
     
     @abc.abstractmethod
     def reset(self):
         """Return, transformed any lines that the transformer has accumulated,
         and reset its internal state.
         
         Must be overridden by subclasses.
         """
         pass
     
     # Set this to True to allow the transformer to act on lines inside strings.
     look_in_string = False
     
     @classmethod
     def wrap(cls, func):
         """Can be used by subclasses as a decorator, to return a factory that
         will allow instantiation with the decorated object.
         """
         @functools.wraps(func)
         def transformer_factory():
             transformer = cls(func)
             if getattr(transformer_factory, 'look_in_string', False):
                 transformer.look_in_string = True
             return transformer
         
         return transformer_factory
 
 class StatelessInputTransformer(InputTransformer):
     """Wrapper for a stateless input transformer implemented as a function."""
     def __init__(self, func):
         self.func = func
     
     def __repr__(self):
         return "StatelessInputTransformer(func={!r})".format(self.func)
     
     def push(self, line):
         """Send a line of input to the transformer, returning the
         transformed input."""
         return self.func(line)
     
     def reset(self):
         """No-op - exists for compatibility."""
         pass
 
 class CoroutineInputTransformer(InputTransformer):
     """Wrapper for an input transformer implemented as a coroutine."""
     def __init__(self, coro):
         # Prime it
         self.coro = coro()
         next(self.coro)
     
     def __repr__(self):
         return "CoroutineInputTransformer(coro={!r})".format(self.coro)
     
     def push(self, line):
         """Send a line of input to the transformer, returning the
         transformed input or None if the transformer is waiting for more
         input.
         """
         return self.coro.send(line)
     
     def reset(self):
         """Return, transformed any lines that the transformer has
         accumulated, and reset its internal state.
         """
         return self.coro.send(None)
 
 class TokenInputTransformer(InputTransformer):
     """Wrapper for a token-based input transformer.
     
     func should accept a list of tokens (5-tuples, see tokenize docs), and
     return an iterable which can be passed to tokenize.untokenize().
     """
     def __init__(self, func):
         self.func = func
         self.current_line = ""
-        self.line_used= False
+        self.line_used = False
         self.reset_tokenizer()
     
     def reset_tokenizer(self):
         self.tokenizer = generate_tokens(self.get_line)
     
     def get_line(self):
         if self.line_used:
             raise tokenize.TokenError
         self.line_used = True
         return self.current_line
     
     def push(self, line):
         self.current_line += line + "\n"
+        if self.current_line.isspace():
+            return self.reset()
+        
         self.line_used = False
         tokens = []
+        stop_at_NL = False
         try:
             for intok in self.tokenizer:
                 tokens.append(intok)
-                if intok[0] in (tokenize.NEWLINE, tokenize.NL):
+                t = intok[0]
+                if t == tokenize.NEWLINE or (stop_at_NL and t == tokenize.NL):
                     # Stop before we try to pull a line we don't have yet
                     break
+                elif t in (tokenize.COMMENT, tokenize.ERRORTOKEN):
+                    stop_at_NL = True
         except tokenize.TokenError:
             # Multi-line statement - stop and try again with the next line
             self.reset_tokenizer()
             return None
         
+        return self.output(tokens)
+    
+    def output(self, tokens):
         self.current_line = ""
         self.reset_tokenizer()
         return untokenize(self.func(tokens)).rstrip('\n')
     
     def reset(self):
         l = self.current_line
         self.current_line = ""
+        self.reset_tokenizer()
         if l:
             return l.rstrip('\n')
 
[e-mail]
-def assemble_logical_lines(tokens):
-    return tokens
+class assemble_python_lines(TokenInputTransformer):
+    def __init__(self):
+        super(assemble_python_lines, self).__init__(None)
+    
+    def output(self, tokens):
+        return self.reset()
+
[e-mail]
+def assemble_logical_lines():
+    """Join lines following explicit line continuations (\)"""
+    line = ''
+    while True:
+        line = (yield line)
+        if not line or line.isspace():
+            continue
+        
+        parts = []
+        while line is not None:
+            parts.append(line.rstrip('\\'))
+            if not line.endswith('\\'):
+                break
+            line = (yield None)
+        
+        # Output
+        line = ' '.join(parts)
 
 # Utilities
 def _make_help_call(target, esc, lspace, next_input=None):
     """Prepares a pinfo(2)/psearch call from a target name and the escape
     (i.e. ? or ??)"""
     method  = 'pinfo2' if esc == '??' \
                 else 'psearch' if '*' in target \
                 else 'pinfo'
     arg = " ".join([method, target])
     if next_input is None:
         return '%sget_ipython().magic(%r)' % (lspace, arg)
     else:
         return '%sget_ipython().set_next_input(%r);get_ipython().magic(%r)' % \
            (lspace, next_input, arg)
 
 @CoroutineInputTransformer.wrap
 def escaped_transformer():
     """Translate lines beginning with one of IPython's escape characters.
     
     This is stateful to allow magic commands etc. to be continued over several
     lines using explicit line continuations (\ at the end of a line).
     """
     
     # These define the transformations for the different escape characters.
     def _tr_system(line_info):
         "Translate lines escaped with: !"
         cmd = line_info.line.lstrip().lstrip(ESC_SHELL)
         return '%sget_ipython().system(%r)' % (line_info.pre, cmd)
 
     def _tr_system2(line_info):
         "Translate lines escaped with: !!"
         cmd = line_info.line.lstrip()[2:]
         return '%sget_ipython().getoutput(%r)' % (line_info.pre, cmd)
 
     def _tr_help(line_info):
         "Translate lines escaped with: ?/??"
         # A naked help line should just fire the intro help screen
         if not line_info.line[1:]:
             return 'get_ipython().show_usage()'
 
         return _make_help_call(line_info.ifun, line_info.esc, line_info.pre)
 
     def _tr_magic(line_info):
         "Translate lines escaped with: %"
         tpl = '%sget_ipython().magic(%r)'
         cmd = ' '.join([line_info.ifun, line_info.the_rest]).strip()
         return tpl % (line_info.pre, cmd)
 
     def _tr_quote(line_info):
         "Translate lines escaped with: ,"
         return '%s%s("%s")' % (line_info.pre, line_info.ifun,
                              '", "'.join(line_info.the_rest.split()) )
 
     def _tr_quote2(line_info):
         "Translate lines escaped with: ;"
         return '%s%s("%s")' % (line_info.pre, line_info.ifun,
                                line_info.the_rest)
 
     def _tr_paren(line_info):
         "Translate lines escaped with: /"
         return '%s%s(%s)' % (line_info.pre, line_info.ifun,
                              ", ".join(line_info.the_rest.split()))
     
     tr = { ESC_SHELL  : _tr_system,
            ESC_SH_CAP : _tr_system2,
            ESC_HELP   : _tr_help,
            ESC_HELP2  : _tr_help,
            ESC_MAGIC  : _tr_magic,
            ESC_QUOTE  : _tr_quote,
            ESC_QUOTE2 : _tr_quote2,
            ESC_PAREN  : _tr_paren }
     
     line = ''
     while True:
         line = (yield line)
         if not line or line.isspace():
             continue
         lineinf = LineInfo(line)
         if lineinf.esc not in tr:
             continue
         
         parts = []
         while line is not None:
             parts.append(line.rstrip('\\'))
             if not line.endswith('\\'):
                 break
             line = (yield None)
         
         # Output
         lineinf = LineInfo(' '.join(parts))
         line = tr[lineinf.esc](lineinf)
 
 _initial_space_re = re.compile(r'\s*')
 
 _help_end_re = re.compile(r"""(%{0,2}
                               [a-zA-Z_*][\w*]*        # Variable name
                               (\.[a-zA-Z_*][\w*]*)*   # .etc.etc
                               )
                               (\?\??)$                # ? or ??""",
                               re.VERBOSE)
 
 def has_comment(src):
     """Indicate whether an input line has (i.e. ends in, or is) a comment.
 
     This uses tokenize, so it can distinguish comments from # inside strings.
 
     Parameters
     ----------
     src : string
       A single line input string.
 
     Returns
     -------
     Boolean: True if source has a comment.
     """
     readline = StringIO(src).readline
     toktypes = set()
     try:
         for t in tokenize.generate_tokens(readline):
             toktypes.add(t[0])
     except tokenize.TokenError:
         pass
     return(tokenize.COMMENT in toktypes)
 
 
 @StatelessInputTransformer.wrap
 def help_end(line):
     """Translate lines with ?/?? at the end"""
     m = _help_end_re.search(line)
     if m is None or has_comment(line):
         return line
     target = m.group(1)
     esc = m.group(3)
     lspace = _initial_space_re.match(line).group(0)
 
     # If we're mid-command, put it back on the next prompt for the user.
     next_input = line.rstrip('?') if line.strip() != m.group(0) else None
 
     return _make_help_call(target, esc, lspace, next_input)
 
 
 @CoroutineInputTransformer.wrap
 def cellmagic():
     """Captures & transforms cell magics.
     
     After a cell magic is started, this stores up any lines it gets until it is
     reset (sent None).
     """
     tpl = 'get_ipython().run_cell_magic(%r, %r, %r)'
     cellmagic_help_re = re.compile('%%\w+\?')
     line = ''
     while True:
         line = (yield line)
         if (not line) or (not line.startswith(ESC_MAGIC2)):
             continue
         
         if cellmagic_help_re.match(line):
             # This case will be handled by help_end
             continue
         
         first = line
         body = []
         line = (yield None)
         while (line is not None) and (line.strip() != ''):
             body.append(line)
             line = (yield None)
         
         # Output
         magic_name, _, first = first.partition(' ')
         magic_name = magic_name.lstrip(ESC_MAGIC2)
         line = tpl % (magic_name, first, u'\n'.join(body))
 
 
 def _strip_prompts(prompt1_re, prompt2_re):
     """Remove matching input prompts from a block of input."""
     line = ''
     while True:
         line = (yield line)
         
         if line is None:
             continue
         
         m = prompt1_re.match(line)
         if m:
             while m:
                 line = (yield line[len(m.group(0)):])
                 if line is None:
                     break
                 m = prompt2_re.match(line)
         else:
             # Prompts not in input - wait for reset
             while line is not None:
                 line = (yield line)
 
 @CoroutineInputTransformer.wrap
 def classic_prompt():
     """Strip the >>>/... prompts of the Python interactive shell."""
     prompt1_re = re.compile(r'^(>>> )')
     prompt2_re = re.compile(r'^(>>> |^\.\.\. )')
     return _strip_prompts(prompt1_re, prompt2_re)
 
 classic_prompt.look_in_string = True
 
 @CoroutineInputTransformer.wrap
 def ipy_prompt():
     """Strip IPython's In [1]:/...: prompts."""
     prompt1_re = re.compile(r'^In \[\d+\]: ')
     prompt2_re = re.compile(r'^(In \[\d+\]: |^\ \ \ \.\.\.+: )')
     return _strip_prompts(prompt1_re, prompt2_re)
 
 ipy_prompt.look_in_string = True
 
 
 @CoroutineInputTransformer.wrap
 def leading_indent():
     """Remove leading indentation.
     
     If the first line starts with a spaces or tabs, the same whitespace will be
     removed from each following line until it is reset.
     """
     space_re = re.compile(r'^[ \t]+')
     line = ''
     while True:
         line = (yield line)
         
         if line is None:
             continue
         
         m = space_re.match(line)
         if m:
             space = m.group(0)
             while line is not None:
                 if line.startswith(space):
                     line = line[len(space):]
                 line = (yield line)
         else:
             # No leading spaces - wait for reset
             while line is not None:
                 line = (yield line)
 
 leading_indent.look_in_string = True
 
 
 def _special_assignment(assignment_re, template):
     """Transform assignment from system & magic commands.
     
     This is stateful so that it can handle magic commands continued on several
     lines.
     """
     line = ''
     while True:
         line = (yield line)
         if not line or line.isspace():
             continue
         
         m = assignment_re.match(line)
         if not m:
             continue
         
         parts = []
         while line is not None:
             parts.append(line.rstrip('\\'))
             if not line.endswith('\\'):
                 break
             line = (yield None)
         
         # Output
         whole = assignment_re.match(' '.join(parts))
         line = template % (whole.group('lhs'), whole.group('cmd'))
 
 @CoroutineInputTransformer.wrap
 def assign_from_system():
     """Transform assignment from system commands (e.g. files = !ls)"""
     assignment_re = re.compile(r'(?P<lhs>(\s*)([\w\.]+)((\s*,\s*[\w\.]+)*))'
                                r'\s*=\s*!\s*(?P<cmd>.*)')
     template = '%s = get_ipython().getoutput(%r)'
     return _special_assignment(assignment_re, template)
 
 @CoroutineInputTransformer.wrap
 def assign_from_magic():
     """Transform assignment from magic commands (e.g. a = %who_ls)"""
     assignment_re = re.compile(r'(?P<lhs>(\s*)([\w\.]+)((\s*,\s*[\w\.]+)*))'
                                r'\s*=\s*%\s*(?P<cmd>.*)')
     template = '%s = get_ipython().magic(%r)'
     return _special_assignment(assignment_re, template)

IPython/core/tests/test_inputsplitter.py

@@ -1,625 +1,626 b''
 # -*- coding: utf-8 -*-
 """Tests for the inputsplitter module.
 
 Authors
 -------
 * Fernando Perez
 * Robert Kern
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 # stdlib
 import unittest
 import sys
 
 # Third party
 import nose.tools as nt
 
 # Our own
 from IPython.core import inputsplitter as isp
 from IPython.core.tests.test_inputtransformer import syntax, syntax_ml
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Semi-complete examples (also used as tests)
 #-----------------------------------------------------------------------------
 
 # Note: at the bottom, there's a slightly more complete version of this that
 # can be useful during development of code here.
 
 def mini_interactive_loop(input_func):
     """Minimal example of the logic of an interactive interpreter loop.
 
     This serves as an example, and it is used by the test system with a fake
     raw_input that simulates interactive input."""
 
     from IPython.core.inputsplitter import InputSplitter
 
     isp = InputSplitter()
     # In practice, this input loop would be wrapped in an outside loop to read
     # input indefinitely, until some exit/quit command was issued.  Here we
     # only illustrate the basic inner loop.
     while isp.push_accepts_more():
         indent = ' '*isp.indent_spaces
         prompt = '>>> ' + indent
         line = indent + input_func(prompt)
         isp.push(line)
 
     # Here we just return input so we can use it in a test suite, but a real
     # interpreter would instead send it for execution somewhere.
     src = isp.source_reset()
     #print 'Input source was:\n', src  # dbg
     return src
 
 #-----------------------------------------------------------------------------
 # Test utilities, just for local use
 #-----------------------------------------------------------------------------
 
 def assemble(block):
     """Assemble a block into multi-line sub-blocks."""
     return ['\n'.join(sub_block)+'\n' for sub_block in block]
 
 
 def pseudo_input(lines):
     """Return a function that acts like raw_input but feeds the input list."""
     ilines = iter(lines)
     def raw_in(prompt):
         try:
             return next(ilines)
         except StopIteration:
             return ''
     return raw_in
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 def test_spaces():
     tests = [('', 0),
              (' ', 1),
              ('\n', 0),
              (' \n', 1),
              ('x', 0),
              (' x', 1),
              ('  x',2),
              ('    x',4),
              # Note: tabs are counted as a single whitespace!
              ('\tx', 1),
              ('\t x', 2),
              ]
     tt.check_pairs(isp.num_ini_spaces, tests)
 
 
 def test_remove_comments():
     tests = [('text', 'text'),
              ('text # comment', 'text '),
              ('text # comment\n', 'text \n'),
              ('text # comment \n', 'text \n'),
              ('line # c \nline\n','line \nline\n'),
              ('line # c \nline#c2  \nline\nline #c\n\n',
               'line \nline\nline\nline \n\n'),
              ]
     tt.check_pairs(isp.remove_comments, tests)
 
 
 def test_get_input_encoding():
     encoding = isp.get_input_encoding()
     nt.assert_true(isinstance(encoding, basestring))
     # simple-minded check that at least encoding a simple string works with the
     # encoding we got.
     nt.assert_equal(u'test'.encode(encoding), b'test')
 
 
 class NoInputEncodingTestCase(unittest.TestCase):
     def setUp(self):
         self.old_stdin = sys.stdin
         class X: pass
         fake_stdin = X()
         sys.stdin = fake_stdin
 
     def test(self):
         # Verify that if sys.stdin has no 'encoding' attribute we do the right
         # thing
         enc = isp.get_input_encoding()
         self.assertEqual(enc, 'ascii')
 
     def tearDown(self):
         sys.stdin = self.old_stdin
 
 
 class InputSplitterTestCase(unittest.TestCase):
     def setUp(self):
         self.isp = isp.InputSplitter()
 
     def test_reset(self):
         isp = self.isp
         isp.push('x=1')
         isp.reset()
         self.assertEqual(isp._buffer, [])
         self.assertEqual(isp.indent_spaces, 0)
         self.assertEqual(isp.source, '')
         self.assertEqual(isp.code, None)
         self.assertEqual(isp._is_complete, False)
 
     def test_source(self):
         self.isp._store('1')
         self.isp._store('2')
         self.assertEqual(self.isp.source, '1\n2\n')
         self.assertTrue(len(self.isp._buffer)>0)
         self.assertEqual(self.isp.source_reset(), '1\n2\n')
         self.assertEqual(self.isp._buffer, [])
         self.assertEqual(self.isp.source, '')
 
     def test_indent(self):
         isp = self.isp # shorthand
         isp.push('x=1')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n    x=1')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('y=2\n')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_indent2(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         isp.push('if 1:')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('    x=1')
         self.assertEqual(isp.indent_spaces, 4)
         # Blank lines shouldn't change the indent level
         isp.push(' '*2)
         self.assertEqual(isp.indent_spaces, 4)
 
     def test_indent3(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         # When a multiline statement contains parens or multiline strings, we
         # shouldn't get confused.
         isp.push("if 1:")
         isp.push("    x = (1+\n    2)")
         self.assertEqual(isp.indent_spaces, 4)
 
     def test_indent4(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         # whitespace after ':' should not screw up indent level
         isp.push('if 1: \n    x=1')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('y=2\n')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\t\n    x=1')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('y=2\n')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_dedent_pass(self):
         isp = self.isp # shorthand
         # should NOT cause dedent
         isp.push('if 1:\n    passes = 5')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('if 1:\n     pass')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     pass   ')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_dedent_break(self):
         isp = self.isp # shorthand
         # should NOT cause dedent
         isp.push('while 1:\n    breaks = 5')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('while 1:\n     break')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('while 1:\n     break   ')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_dedent_continue(self):
         isp = self.isp # shorthand
         # should NOT cause dedent
         isp.push('while 1:\n    continues = 5')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('while 1:\n     continue')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('while 1:\n     continue   ')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_dedent_raise(self):
         isp = self.isp # shorthand
         # should NOT cause dedent
         isp.push('if 1:\n    raised = 4')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('if 1:\n     raise TypeError()')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     raise')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     raise      ')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_dedent_return(self):
         isp = self.isp # shorthand
         # should NOT cause dedent
         isp.push('if 1:\n    returning = 4')
         self.assertEqual(isp.indent_spaces, 4)
         isp.push('if 1:\n     return 5 + 493')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     return')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     return      ')
         self.assertEqual(isp.indent_spaces, 0)
         isp.push('if 1:\n     return(0)')
         self.assertEqual(isp.indent_spaces, 0)
 
     def test_push(self):
         isp = self.isp
         self.assertTrue(isp.push('x=1'))
 
     def test_push2(self):
         isp = self.isp
         self.assertFalse(isp.push('if 1:'))
         for line in ['  x=1', '# a comment', '  y=2']:
+            print(line)
             self.assertTrue(isp.push(line))
 
     def test_push3(self):
         isp = self.isp
         isp.push('if True:')
         isp.push('  a = 1')
         self.assertFalse(isp.push('b = [1,'))
 
     def test_replace_mode(self):
         isp = self.isp
         isp.input_mode = 'cell'
         isp.push('x=1')
         self.assertEqual(isp.source, 'x=1\n')
         isp.push('x=2')
         self.assertEqual(isp.source, 'x=2\n')
 
     def test_push_accepts_more(self):
         isp = self.isp
         isp.push('x=1')
         self.assertFalse(isp.push_accepts_more())
 
     def test_push_accepts_more2(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         isp.push('if 1:')
         self.assertTrue(isp.push_accepts_more())
         isp.push('  x=1')
         self.assertTrue(isp.push_accepts_more())
         isp.push('')
         self.assertFalse(isp.push_accepts_more())
 
     def test_push_accepts_more3(self):
         isp = self.isp
         isp.push("x = (2+\n3)")
         self.assertFalse(isp.push_accepts_more())
 
     def test_push_accepts_more4(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         # When a multiline statement contains parens or multiline strings, we
         # shouldn't get confused.
         # FIXME: we should be able to better handle de-dents in statements like
         # multiline strings and multiline expressions (continued with \ or
         # parens).  Right now we aren't handling the indentation tracking quite
         # correctly with this, though in practice it may not be too much of a
         # problem.  We'll need to see.
         isp.push("if 1:")
         isp.push("    x = (2+")
         isp.push("    3)")
         self.assertTrue(isp.push_accepts_more())
         isp.push("    y = 3")
         self.assertTrue(isp.push_accepts_more())
         isp.push('')
         self.assertFalse(isp.push_accepts_more())
 
     def test_push_accepts_more5(self):
         # In cell mode, inputs must be fed in whole blocks, so skip this test
         if self.isp.input_mode == 'cell': return
 
         isp = self.isp
         isp.push('try:')
         isp.push('    a = 5')
         isp.push('except:')
         isp.push('    raise')
         self.assertTrue(isp.push_accepts_more())
 
     def test_continuation(self):
         isp = self.isp
         isp.push("import os, \\")
         self.assertTrue(isp.push_accepts_more())
         isp.push("sys")
         self.assertFalse(isp.push_accepts_more())
 
     def test_syntax_error(self):
         isp = self.isp
         # Syntax errors immediately produce a 'ready' block, so the invalid
         # Python can be sent to the kernel for evaluation with possible ipython
         # special-syntax conversion.
         isp.push('run foo')
         self.assertFalse(isp.push_accepts_more())
 
     def test_unicode(self):
         self.isp.push(u"Pérez")
         self.isp.push(u'\xc3\xa9')
         self.isp.push(u"u'\xc3\xa9'")
 
     def test_line_continuation(self):
         """ Test issue #2108."""
         isp = self.isp
         # A blank line after a line continuation should not accept more
         isp.push("1 \\\n\n")
         self.assertFalse(isp.push_accepts_more())
         # Whitespace after a \ is a SyntaxError.  The only way to test that
         # here is to test that push doesn't accept more (as with
         # test_syntax_error() above).
         isp.push(r"1 \ ")
         self.assertFalse(isp.push_accepts_more())
         # Even if the line is continuable (c.f. the regular Python
         # interpreter)
         isp.push(r"(1 \ ")
         self.assertFalse(isp.push_accepts_more())
 
 class InteractiveLoopTestCase(unittest.TestCase):
     """Tests for an interactive loop like a python shell.
     """
     def check_ns(self, lines, ns):
         """Validate that the given input lines produce the resulting namespace.
 
         Note: the input lines are given exactly as they would be typed in an
         auto-indenting environment, as mini_interactive_loop above already does
         auto-indenting and prepends spaces to the input.
         """
         src = mini_interactive_loop(pseudo_input(lines))
         test_ns = {}
         exec src in test_ns
         # We can't check that the provided ns is identical to the test_ns,
         # because Python fills test_ns with extra keys (copyright, etc).  But
         # we can check that the given dict is *contained* in test_ns
         for k,v in ns.iteritems():
             self.assertEqual(test_ns[k], v)
 
     def test_simple(self):
         self.check_ns(['x=1'], dict(x=1))
 
     def test_simple2(self):
         self.check_ns(['if 1:', 'x=2'], dict(x=2))
 
     def test_xy(self):
         self.check_ns(['x=1; y=2'], dict(x=1, y=2))
 
     def test_abc(self):
         self.check_ns(['if 1:','a=1','b=2','c=3'], dict(a=1, b=2, c=3))
 
     def test_multi(self):
         self.check_ns(['x =(1+','1+','2)'], dict(x=4))
 
 
 def test_LineInfo():
     """Simple test for LineInfo construction and str()"""
     linfo = isp.LineInfo('  %cd /home')
     nt.assert_equal(str(linfo), 'LineInfo [  |%|cd|/home]')
 
 
 
 
 class IPythonInputTestCase(InputSplitterTestCase):
     """By just creating a new class whose .isp is a different instance, we
     re-run the same test battery on the new input splitter.
 
     In addition, this runs the tests over the syntax and syntax_ml dicts that
     were tested by individual functions, as part of the OO interface.
 
     It also makes some checks on the raw buffer storage.
     """
 
     def setUp(self):
         self.isp = isp.IPythonInputSplitter(input_mode='line')
 
     def test_syntax(self):
         """Call all single-line syntax tests from the main object"""
         isp = self.isp
         for example in syntax.itervalues():
             for raw, out_t in example:
                 if raw.startswith(' '):
                     continue
 
                 isp.push(raw+'\n')
                 out, out_raw = isp.source_raw_reset()
                 self.assertEqual(out.rstrip(), out_t,
                         tt.pair_fail_msg.format("inputsplitter",raw, out_t, out))
                 self.assertEqual(out_raw.rstrip(), raw.rstrip())
 
     def test_syntax_multiline(self):
         isp = self.isp
         for example in syntax_ml.itervalues():
             for line_pairs in example:
                 out_t_parts = []
                 raw_parts = []
                 for lraw, out_t_part in line_pairs:
                     if out_t_part is not None:
                         out_t_parts.append(out_t_part)
                     
                     if lraw is not None:
                         isp.push(lraw)
                         raw_parts.append(lraw)
 
                 out, out_raw = isp.source_raw_reset()
                 out_t = '\n'.join(out_t_parts).rstrip()
                 raw = '\n'.join(raw_parts).rstrip()
                 self.assertEqual(out.rstrip(), out_t)
                 self.assertEqual(out_raw.rstrip(), raw)
 
 
 class BlockIPythonInputTestCase(IPythonInputTestCase):
 
     # Deactivate tests that don't make sense for the block mode
     test_push3 = test_split = lambda s: None
 
     def setUp(self):
         self.isp = isp.IPythonInputSplitter(input_mode='cell')
 
     def test_syntax_multiline(self):
         isp = self.isp
         for example in syntax_ml.itervalues():
             raw_parts = []
             out_t_parts = []
             for line_pairs in example:
                 raw_parts, out_t_parts = zip(*line_pairs)
 
                 raw = '\n'.join(r for r in raw_parts if r is not None)
                 out_t = '\n'.join(o for o in out_t_parts if o is not None)
 
                 isp.push(raw)
                 out, out_raw = isp.source_raw_reset()
                 # Match ignoring trailing whitespace
                 self.assertEqual(out.rstrip(), out_t.rstrip())
                 self.assertEqual(out_raw.rstrip(), raw.rstrip())
 
     def test_syntax_multiline_cell(self):
         isp = self.isp
         for example in syntax_ml.itervalues():
 
             out_t_parts = []
             for line_pairs in example:
                 raw = '\n'.join(r for r, _ in line_pairs if r is not None)
                 out_t = '\n'.join(t for _,t in line_pairs if t is not None)
                 out = isp.transform_cell(raw)
                 # Match ignoring trailing whitespace
                 self.assertEqual(out.rstrip(), out_t.rstrip())
 
 #-----------------------------------------------------------------------------
 # Main - use as a script, mostly for developer experiments
 #-----------------------------------------------------------------------------
 
 if __name__ == '__main__':
     # A simple demo for interactive experimentation.  This code will not get
     # picked up by any test suite.
     from IPython.core.inputsplitter import InputSplitter, IPythonInputSplitter
 
     # configure here the syntax to use, prompt and whether to autoindent
     #isp, start_prompt = InputSplitter(), '>>> '
     isp, start_prompt = IPythonInputSplitter(), 'In> '
 
     autoindent = True
     #autoindent = False
 
     try:
         while True:
             prompt = start_prompt
             while isp.push_accepts_more():
                 indent = ' '*isp.indent_spaces
                 if autoindent:
                     line = indent + raw_input(prompt+indent)
                 else:
                     line = raw_input(prompt)
                 isp.push(line)
                 prompt = '... '
 
             # Here we just return input so we can use it in a test suite, but a
             # real interpreter would instead send it for execution somewhere.
             #src = isp.source; raise EOFError # dbg
             src, raw = isp.source_raw_reset()
             print 'Input source was:\n', src
             print 'Raw source was:\n', raw
     except EOFError:
         print 'Bye'
 
 # Tests for cell magics support
 
 def test_last_blank():
     nt.assert_false(isp.last_blank(''))
     nt.assert_false(isp.last_blank('abc'))
     nt.assert_false(isp.last_blank('abc\n'))
     nt.assert_false(isp.last_blank('abc\na'))
 
     nt.assert_true(isp.last_blank('\n'))
     nt.assert_true(isp.last_blank('\n '))
     nt.assert_true(isp.last_blank('abc\n '))
     nt.assert_true(isp.last_blank('abc\n\n'))
     nt.assert_true(isp.last_blank('abc\nd\n\n'))
     nt.assert_true(isp.last_blank('abc\nd\ne\n\n'))
     nt.assert_true(isp.last_blank('abc \n \n \n\n'))
 
 
 def test_last_two_blanks():
     nt.assert_false(isp.last_two_blanks(''))
     nt.assert_false(isp.last_two_blanks('abc'))
     nt.assert_false(isp.last_two_blanks('abc\n'))
     nt.assert_false(isp.last_two_blanks('abc\n\na'))
     nt.assert_false(isp.last_two_blanks('abc\n \n'))
     nt.assert_false(isp.last_two_blanks('abc\n\n'))
 
     nt.assert_true(isp.last_two_blanks('\n\n'))
     nt.assert_true(isp.last_two_blanks('\n\n '))
     nt.assert_true(isp.last_two_blanks('\n \n'))
     nt.assert_true(isp.last_two_blanks('abc\n\n '))
     nt.assert_true(isp.last_two_blanks('abc\n\n\n'))
     nt.assert_true(isp.last_two_blanks('abc\n\n \n'))
     nt.assert_true(isp.last_two_blanks('abc\n\n \n '))
     nt.assert_true(isp.last_two_blanks('abc\n\n \n \n'))
     nt.assert_true(isp.last_two_blanks('abc\nd\n\n\n'))
     nt.assert_true(isp.last_two_blanks('abc\nd\ne\nf\n\n\n'))
 
 
 class CellMagicsCommon(object):
 
     def test_whole_cell(self):
         src = "%%cellm line\nbody\n"
         sp = self.sp
         sp.push(src)
         out = sp.source_reset()
         ref = u"get_ipython().run_cell_magic({u}'cellm', {u}'line', {u}'body')\n"
         nt.assert_equal(out, py3compat.u_format(ref))
 
     def tearDown(self):
         self.sp.reset()
 
 
 class CellModeCellMagics(CellMagicsCommon, unittest.TestCase):
     sp = isp.IPythonInputSplitter(input_mode='cell')
 
     def test_incremental(self):
         sp = self.sp
         src = '%%cellm line2\n'
         sp.push(src)
         nt.assert_true(sp.push_accepts_more()) #1
         src += '\n'
         sp.push(src)
         # Note: if we ever change the logic to allow full blank lines (see
         # _handle_cell_magic), then the following test should change to true
         nt.assert_false(sp.push_accepts_more()) #2
         # By now, even with full blanks allowed, a second blank should signal
         # the end.  For now this test is only a redundancy safety, but don't
         # delete it in case we change our mind and the previous one goes to
         # true.
         src += '\n'
         sp.push(src)
         nt.assert_false(sp.push_accepts_more()) #3
 
 
 class LineModeCellMagics(CellMagicsCommon, unittest.TestCase):
     sp = isp.IPythonInputSplitter(input_mode='line')
 
     def test_incremental(self):
         sp = self.sp
         sp.push('%%cellm line2\n')
         nt.assert_true(sp.push_accepts_more()) #1
         sp.push('\n')
         nt.assert_false(sp.push_accepts_more()) #2

IPython/core/tests/test_inputtransformer.py

@@ -1,368 +1,368 b''
 import tokenize
 import unittest
 import nose.tools as nt
 
 from IPython.testing import tools as tt
 from IPython.utils import py3compat
 u_fmt = py3compat.u_format
 
 from IPython.core import inputtransformer as ipt
 
 def transform_and_reset(transformer):
     transformer = transformer()
     def transform(inp):
         try:
             return transformer.push(inp)
         finally:
             transformer.reset()
     
     return transform
 
 # Transformer tests
 def transform_checker(tests, transformer):
     """Utility to loop over test inputs"""
     transformer = transformer()
     try:
         for inp, tr in tests:
             if inp is None:
                 out = transformer.reset()
             else:
                 out = transformer.push(inp)
             nt.assert_equal(out, tr)
     finally:
         transformer.reset()
 
 # Data for all the syntax tests in the form of lists of pairs of
 # raw/transformed input.  We store it here as a global dict so that we can use
 # it both within single-function tests and also to validate the behavior of the
 # larger objects
 
 syntax = \
   dict(assign_system =
        [(i,py3compat.u_format(o)) for i,o in \
        [(u'a =! ls', "a = get_ipython().getoutput({u}'ls')"),
         (u'b = !ls', "b = get_ipython().getoutput({u}'ls')"),
         ('x=1', 'x=1'), # normal input is unmodified
         ('    ','    '),  # blank lines are kept intact
         ]],
 
        assign_magic =
        [(i,py3compat.u_format(o)) for i,o in \
        [(u'a =% who', "a = get_ipython().magic({u}'who')"),
         (u'b = %who', "b = get_ipython().magic({u}'who')"),
         ('x=1', 'x=1'), # normal input is unmodified
         ('    ','    '),  # blank lines are kept intact
         ]],
 
        classic_prompt =
        [('>>> x=1', 'x=1'),
         ('x=1', 'x=1'), # normal input is unmodified
         ('    ', '    '),  # blank lines are kept intact
         ],
 
        ipy_prompt =
        [('In [1]: x=1', 'x=1'),
         ('x=1', 'x=1'), # normal input is unmodified
         ('    ','    '),  # blank lines are kept intact
         ],
 
        # Tests for the escape transformer to leave normal code alone
        escaped_noesc =
        [ ('    ', '    '),
          ('x=1', 'x=1'),
          ],
 
        # System calls
        escaped_shell =
        [(i,py3compat.u_format(o)) for i,o in \
        [ (u'!ls', "get_ipython().system({u}'ls')"),
          # Double-escape shell, this means to capture the output of the
          # subprocess and return it
          (u'!!ls', "get_ipython().getoutput({u}'ls')"),
          ]],
 
        # Help/object info
        escaped_help =
        [(i,py3compat.u_format(o)) for i,o in \
        [ (u'?', 'get_ipython().show_usage()'),
          (u'?x1', "get_ipython().magic({u}'pinfo x1')"),
          (u'??x2', "get_ipython().magic({u}'pinfo2 x2')"),
          (u'?a.*s', "get_ipython().magic({u}'psearch a.*s')"),
          (u'?%hist1', "get_ipython().magic({u}'pinfo %hist1')"),
          (u'?%%hist2', "get_ipython().magic({u}'pinfo %%hist2')"),
          (u'?abc = qwe', "get_ipython().magic({u}'pinfo abc')"),
          ]],
 
       end_help =
       [(i,py3compat.u_format(o)) for i,o in \
       [ (u'x3?', "get_ipython().magic({u}'pinfo x3')"),
         (u'x4??', "get_ipython().magic({u}'pinfo2 x4')"),
         (u'%hist1?', "get_ipython().magic({u}'pinfo %hist1')"),
         (u'%hist2??', "get_ipython().magic({u}'pinfo2 %hist2')"),
         (u'%%hist3?', "get_ipython().magic({u}'pinfo %%hist3')"),
         (u'%%hist4??', "get_ipython().magic({u}'pinfo2 %%hist4')"),
         (u'f*?', "get_ipython().magic({u}'psearch f*')"),
         (u'ax.*aspe*?', "get_ipython().magic({u}'psearch ax.*aspe*')"),
         (u'a = abc?', "get_ipython().set_next_input({u}'a = abc');"
                       "get_ipython().magic({u}'pinfo abc')"),
         (u'a = abc.qe??', "get_ipython().set_next_input({u}'a = abc.qe');"
                           "get_ipython().magic({u}'pinfo2 abc.qe')"),
         (u'a = *.items?', "get_ipython().set_next_input({u}'a = *.items');"
                           "get_ipython().magic({u}'psearch *.items')"),
         (u'plot(a?', "get_ipython().set_next_input({u}'plot(a');"
                      "get_ipython().magic({u}'pinfo a')"),
         (u'a*2 #comment?', 'a*2 #comment?'),
         ]],
 
        # Explicit magic calls
        escaped_magic =
        [(i,py3compat.u_format(o)) for i,o in \
        [ (u'%cd', "get_ipython().magic({u}'cd')"),
          (u'%cd /home', "get_ipython().magic({u}'cd /home')"),
          # Backslashes need to be escaped.
          (u'%cd C:\\User', "get_ipython().magic({u}'cd C:\\\\User')"),
          (u'    %magic', "    get_ipython().magic({u}'magic')"),
          ]],
 
        # Quoting with separate arguments
        escaped_quote =
        [ (',f', 'f("")'),
          (',f x', 'f("x")'),
          ('  ,f y', '  f("y")'),
          (',f a b', 'f("a", "b")'),
          ],
 
        # Quoting with single argument
        escaped_quote2 =
        [ (';f', 'f("")'),
          (';f x', 'f("x")'),
          ('  ;f y', '  f("y")'),
          (';f a b', 'f("a b")'),
          ],
 
        # Simply apply parens
        escaped_paren =
        [ ('/f', 'f()'),
          ('/f x', 'f(x)'),
          ('  /f y', '  f(y)'),
          ('/f a b', 'f(a, b)'),
          ],
 
        # Check that we transform prompts before other transforms
        mixed =
        [(i,py3compat.u_format(o)) for i,o in \
        [ (u'In [1]: %lsmagic', "get_ipython().magic({u}'lsmagic')"),
          (u'>>> %lsmagic', "get_ipython().magic({u}'lsmagic')"),
          (u'In [2]: !ls', "get_ipython().system({u}'ls')"),
          (u'In [3]: abs?', "get_ipython().magic({u}'pinfo abs')"),
          (u'In [4]: b = %who', "b = get_ipython().magic({u}'who')"),
          ]],
        )
 
 # multiline syntax examples.  Each of these should be a list of lists, with
 # each entry itself having pairs of raw/transformed input.  The union (with
 # '\n'.join() of the transformed inputs is what the splitter should produce
 # when fed the raw lines one at a time via push.
 syntax_ml = \
   dict(classic_prompt =
        [ [('>>> for i in range(10):','for i in range(10):'),
           ('...     print i','    print i'),
           ('... ', ''),
           ],
          [('>>> a="""','a="""'),
           ('... 123"""','123"""'),
           ],
          [('a="""','a="""'),
           ('... 123"""','... 123"""'),
           ],
         ],
 
        ipy_prompt =
        [ [('In [24]: for i in range(10):','for i in range(10):'),
           ('   ....:     print i','    print i'),
           ('   ....: ', ''),
           ],
          [('In [2]: a="""','a="""'),
           ('   ...: 123"""','123"""'),
           ],
          [('a="""','a="""'),
           ('   ...: 123"""','   ...: 123"""'),
           ],
          ],
 
        multiline_datastructure =
        [ [('>>> a = [1,','a = [1,'),
           ('... 2]','2]'),
          ],
        ],
        
        leading_indent =
        [ [('    print "hi"','print "hi"'),
           ],
          [('  for a in range(5):','for a in range(5):'),
           ('    a*2','  a*2'),
           ],
          [('    a="""','a="""'),
           ('    123"""','123"""'),
            ],
          [('a="""','a="""'),
           ('    123"""','    123"""'),
           ],
        ],
        
        cellmagic =
        [ [(u'%%foo a', None),
           (None, u_fmt("get_ipython().run_cell_magic({u}'foo', {u}'a', {u}'')")),
           ],
          [(u'%%bar 123', None),
           (u'hello', None),
           (u'', u_fmt("get_ipython().run_cell_magic({u}'bar', {u}'123', {u}'hello')")),
           ],
        ],
        
        escaped =
        [ [('%abc def \\', None),
           ('ghi', u_fmt("get_ipython().magic({u}'abc def  ghi')")),
           ],
          [('%abc def \\', None),
           ('ghi\\', None),
           (None, u_fmt("get_ipython().magic({u}'abc def  ghi')")),
           ],
        ],
        
        assign_magic =
        [ [(u'a = %bc de \\', None),
           (u'fg', u_fmt("a = get_ipython().magic({u}'bc de  fg')")),
           ],
          [(u'a = %bc de \\', None),
           (u'fg\\', None),
           (None, u_fmt("a = get_ipython().magic({u}'bc de  fg')")),
           ],
        ],
        
        assign_system =
        [ [(u'a = !bc de \\', None),
           (u'fg', u_fmt("a = get_ipython().getoutput({u}'bc de  fg')")),
           ],
          [(u'a = !bc de \\', None),
           (u'fg\\', None),
           (None, u_fmt("a = get_ipython().getoutput({u}'bc de  fg')")),
           ],
        ],
        )
 
 
 def test_assign_system():
     tt.check_pairs(transform_and_reset(ipt.assign_from_system), syntax['assign_system'])
     for example in syntax_ml['assign_system']:
         transform_checker(example, ipt.assign_from_system)
 
 def test_assign_magic():
     tt.check_pairs(transform_and_reset(ipt.assign_from_magic), syntax['assign_magic'])
     for example in syntax_ml['assign_magic']:
         transform_checker(example, ipt.assign_from_magic)
 
 
 def test_classic_prompt():
     tt.check_pairs(transform_and_reset(ipt.classic_prompt), syntax['classic_prompt'])
     for example in syntax_ml['classic_prompt']:
         transform_checker(example, ipt.classic_prompt)
     for example in syntax_ml['multiline_datastructure']:
         transform_checker(example, ipt.classic_prompt)
 
 
 def test_ipy_prompt():
     tt.check_pairs(transform_and_reset(ipt.ipy_prompt), syntax['ipy_prompt'])
     for example in syntax_ml['ipy_prompt']:
         transform_checker(example, ipt.ipy_prompt)
 
 def test_help_end():
     tt.check_pairs(transform_and_reset(ipt.help_end), syntax['end_help'])
 
 def test_escaped_noesc():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_noesc'])
 
 
 def test_escaped_shell():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_shell'])
 
 
 def test_escaped_help():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_help'])
 
 
 def test_escaped_magic():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_magic'])
 
 
 def test_escaped_quote():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_quote'])
 
 
 def test_escaped_quote2():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_quote2'])
 
 
 def test_escaped_paren():
     tt.check_pairs(transform_and_reset(ipt.escaped_transformer), syntax['escaped_paren'])
 
 def test_escaped_multiline():
     for example in syntax_ml['escaped']:
         transform_checker(example, ipt.escaped_transformer)
 
 def test_cellmagic():
     for example in syntax_ml['cellmagic']:
         transform_checker(example, ipt.cellmagic)
 
 def test_has_comment():
     tests = [('text', False),
              ('text #comment', True),
              ('text #comment\n', True),
              ('#comment', True),
              ('#comment\n', True),
              ('a = "#string"', False),
              ('a = "#string" # comment', True),
              ('a #comment not "string"', True),
              ]
     tt.check_pairs(ipt.has_comment, tests)
 
 @ipt.TokenInputTransformer.wrap
 def decistmt(tokens):
     """Substitute Decimals for floats in a string of statements.
 
     Based on an example from the tokenize module docs.
     """
     result = []
     for toknum, tokval, _, _, _  in tokens:
         if toknum == tokenize.NUMBER and '.' in tokval:  # replace NUMBER tokens
             for newtok in [
                 (tokenize.NAME, 'Decimal'),
                 (tokenize.OP, '('),
                 (tokenize.STRING, repr(tokval)),
                 (tokenize.OP, ')')
             ]:
                 yield newtok
         else:
             yield (toknum, tokval)
 
 
 
 def test_token_input_transformer():
     tests = [(u'1.2', u_fmt(u"Decimal ({u}'1.2')")),
              (u'"1.2"', u'"1.2"'),
              ]
     tt.check_pairs(transform_and_reset(decistmt), tests)
     ml_tests = \
     [ [(u"a = 1.2; b = '''x", None),
        (u"y'''", u_fmt(u"a =Decimal ({u}'1.2');b ='''x\ny'''")),
       ],
-      [(u"a = [1.2,", u_fmt(u"a =[Decimal ({u}'1.2'),")),
-       (u"3]", u"3 ]"),
+      [(u"a = [1.2,", None),
+       (u"3]", u_fmt(u"a =[Decimal ({u}'1.2'),\n3 ]")),
       ],
       [(u"a = '''foo", None),  # Test resetting when within a multi-line string
        (u"bar", None),
        (None, u"a = '''foo\nbar"),
       ],
     ]
     for example in ml_tests:
         transform_checker(example, decistmt)