Allow to mark transformers as having side effects

We then unly run them when actually running the code.

IPython/core/inputtransformer2.py

@@ -1,724 +1,726 b''
 """Input transformer machinery to support IPython special syntax.
 
 This includes the machinery to recognise and transform ``%magic`` commands,
 ``!system`` commands, ``help?`` querying, prompt stripping, and so forth.
 
 Added: IPython 7.0. Replaces inputsplitter and inputtransformer which were
 deprecated in 7.0.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from codeop import compile_command
 import re
 import tokenize
 from typing import List, Tuple, Optional, Any
 import warnings
 
 _indent_re = re.compile(r'^[ \t]+')
 
 def leading_empty_lines(lines):
     """Remove leading empty lines
 
     If the leading lines are empty or contain only whitespace, they will be
     removed.
     """
     if not lines:
         return lines
     for i, line in enumerate(lines):
         if line and not line.isspace():
             return lines[i:]
     return lines
 
 def leading_indent(lines):
     """Remove leading indentation.
 
     If the first line starts with a spaces or tabs, the same whitespace will be
     removed from each following line in the cell.
     """
     if not lines:
         return lines
     m = _indent_re.match(lines[0])
     if not m:
         return lines
     space = m.group(0)
     n = len(space)
     return [l[n:] if l.startswith(space) else l
             for l in lines]
 
 class PromptStripper:
     """Remove matching input prompts from a block of input.
 
     Parameters
     ----------
     prompt_re : regular expression
         A regular expression matching any input prompt (including continuation,
         e.g. ``...``)
     initial_re : regular expression, optional
         A regular expression matching only the initial prompt, but not continuation.
         If no initial expression is given, prompt_re will be used everywhere.
         Used mainly for plain Python prompts (``>>>``), where the continuation prompt
         ``...`` is a valid Python expression in Python 3, so shouldn't be stripped.
 
     If initial_re and prompt_re differ,
     only initial_re will be tested against the first line.
     If any prompt is found on the first two lines,
     prompts will be stripped from the rest of the block.
     """
     def __init__(self, prompt_re, initial_re=None):
         self.prompt_re = prompt_re
         self.initial_re = initial_re or prompt_re
 
     def _strip(self, lines):
         return [self.prompt_re.sub('', l, count=1) for l in lines]
 
     def __call__(self, lines):
         if not lines:
             return lines
         if self.initial_re.match(lines[0]) or \
                 (len(lines) > 1 and self.prompt_re.match(lines[1])):
             return self._strip(lines)
         return lines
 
 classic_prompt = PromptStripper(
     prompt_re=re.compile(r'^(>>>|\.\.\.)( |$)'),
     initial_re=re.compile(r'^>>>( |$)')
 )
 
 ipython_prompt = PromptStripper(re.compile(r'^(In \[\d+\]: |\s*\.{3,}: ?)'))
 
 def cell_magic(lines):
     if not lines or not lines[0].startswith('%%'):
         return lines
     if re.match(r'%%\w+\?', lines[0]):
         # This case will be handled by help_end
         return lines
     magic_name, _, first_line = lines[0][2:-1].partition(' ')
     body = ''.join(lines[1:])
     return ['get_ipython().run_cell_magic(%r, %r, %r)\n'
             % (magic_name, first_line, body)]
 
 
 def _find_assign_op(token_line) -> Optional[int]:
     """Get the index of the first assignment in the line ('=' not inside brackets)
 
     Note: We don't try to support multiple special assignment (a = b = %foo)
     """
     paren_level = 0
     for i, ti in enumerate(token_line):
         s = ti.string
         if s == '=' and paren_level == 0:
             return i
         if s in {'(','[','{'}:
             paren_level += 1
         elif s in {')', ']', '}'}:
             if paren_level > 0:
                 paren_level -= 1
     return None
 
 def find_end_of_continued_line(lines, start_line: int):
     """Find the last line of a line explicitly extended using backslashes.
 
     Uses 0-indexed line numbers.
     """
     end_line = start_line
     while lines[end_line].endswith('\\\n'):
         end_line += 1
         if end_line >= len(lines):
             break
     return end_line
 
 def assemble_continued_line(lines, start: Tuple[int, int], end_line: int):
     r"""Assemble a single line from multiple continued line pieces
 
     Continued lines are lines ending in ``\``, and the line following the last
     ``\`` in the block.
 
     For example, this code continues over multiple lines::
 
         if (assign_ix is not None) \
              and (len(line) >= assign_ix + 2) \
              and (line[assign_ix+1].string == '%') \
              and (line[assign_ix+2].type == tokenize.NAME):
 
     This statement contains four continued line pieces.
     Assembling these pieces into a single line would give::
 
         if (assign_ix is not None) and (len(line) >= assign_ix + 2) and (line[...
 
     This uses 0-indexed line numbers. *start* is (lineno, colno).
 
     Used to allow ``%magic`` and ``!system`` commands to be continued over
     multiple lines.
     """
     parts = [lines[start[0]][start[1]:]] + lines[start[0]+1:end_line+1]
     return ' '.join([p[:-2] for p in parts[:-1]]  # Strip backslash+newline
                     + [parts[-1][:-1]])         # Strip newline from last line
 
 class TokenTransformBase:
     """Base class for transformations which examine tokens.
 
     Special syntax should not be transformed when it occurs inside strings or
     comments. This is hard to reliably avoid with regexes. The solution is to
     tokenise the code as Python, and recognise the special syntax in the tokens.
 
     IPython's special syntax is not valid Python syntax, so tokenising may go
     wrong after the special syntax starts. These classes therefore find and
     transform *one* instance of special syntax at a time into regular Python
     syntax. After each transformation, tokens are regenerated to find the next
     piece of special syntax.
 
     Subclasses need to implement one class method (find)
     and one regular method (transform).
 
     The priority attribute can select which transformation to apply if multiple
     transformers match in the same place. Lower numbers have higher priority.
     This allows "%magic?" to be turned into a help call rather than a magic call.
     """
     # Lower numbers -> higher priority (for matches in the same location)
     priority = 10
 
     def sortby(self):
         return self.start_line, self.start_col, self.priority
 
     def __init__(self, start):
         self.start_line = start[0] - 1   # Shift from 1-index to 0-index
         self.start_col = start[1]
 
     @classmethod
     def find(cls, tokens_by_line):
         """Find one instance of special syntax in the provided tokens.
 
         Tokens are grouped into logical lines for convenience,
         so it is easy to e.g. look at the first token of each line.
         *tokens_by_line* is a list of lists of tokenize.TokenInfo objects.
 
         This should return an instance of its class, pointing to the start
         position it has found, or None if it found no match.
         """
         raise NotImplementedError
 
     def transform(self, lines: List[str]):
         """Transform one instance of special syntax found by ``find()``
 
         Takes a list of strings representing physical lines,
         returns a similar list of transformed lines.
         """
         raise NotImplementedError
 
 class MagicAssign(TokenTransformBase):
     """Transformer for assignments from magics (a = %foo)"""
     @classmethod
     def find(cls, tokens_by_line):
         """Find the first magic assignment (a = %foo) in the cell.
         """
         for line in tokens_by_line:
             assign_ix = _find_assign_op(line)
             if (assign_ix is not None) \
                     and (len(line) >= assign_ix + 2) \
                     and (line[assign_ix+1].string == '%') \
                     and (line[assign_ix+2].type == tokenize.NAME):
                 return cls(line[assign_ix+1].start)
 
     def transform(self, lines: List[str]):
         """Transform a magic assignment found by the ``find()`` classmethod.
         """
         start_line, start_col = self.start_line, self.start_col
         lhs = lines[start_line][:start_col]
         end_line = find_end_of_continued_line(lines, start_line)
         rhs = assemble_continued_line(lines, (start_line, start_col), end_line)
         assert rhs.startswith('%'), rhs
         magic_name, _, args = rhs[1:].partition(' ')
 
         lines_before = lines[:start_line]
         call = "get_ipython().run_line_magic({!r}, {!r})".format(magic_name, args)
         new_line = lhs + call + '\n'
         lines_after = lines[end_line+1:]
 
         return lines_before + [new_line] + lines_after
 
 
 class SystemAssign(TokenTransformBase):
     """Transformer for assignments from system commands (a = !foo)"""
     @classmethod
     def find(cls, tokens_by_line):
         """Find the first system assignment (a = !foo) in the cell.
         """
         for line in tokens_by_line:
             assign_ix = _find_assign_op(line)
             if (assign_ix is not None) \
                     and not line[assign_ix].line.strip().startswith('=') \
                     and (len(line) >= assign_ix + 2) \
                     and (line[assign_ix + 1].type == tokenize.ERRORTOKEN):
                 ix = assign_ix + 1
 
                 while ix < len(line) and line[ix].type == tokenize.ERRORTOKEN:
                     if line[ix].string == '!':
                         return cls(line[ix].start)
                     elif not line[ix].string.isspace():
                         break
                     ix += 1
 
     def transform(self, lines: List[str]):
         """Transform a system assignment found by the ``find()`` classmethod.
         """
         start_line, start_col = self.start_line, self.start_col
 
         lhs = lines[start_line][:start_col]
         end_line = find_end_of_continued_line(lines, start_line)
         rhs = assemble_continued_line(lines, (start_line, start_col), end_line)
         assert rhs.startswith('!'), rhs
         cmd = rhs[1:]
 
         lines_before = lines[:start_line]
         call = "get_ipython().getoutput({!r})".format(cmd)
         new_line = lhs + call + '\n'
         lines_after = lines[end_line + 1:]
 
         return lines_before + [new_line] + lines_after
 
 # 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
 
 ESCAPE_SINGLES = {'!', '?', '%', ',', ';', '/'}
 ESCAPE_DOUBLES = {'!!', '??'}  # %% (cell magic) is handled separately
 
 def _make_help_call(target, esc, 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])
     #Prepare arguments for get_ipython().run_line_magic(magic_name, magic_args)
     t_magic_name, _, t_magic_arg_s = arg.partition(' ')
     t_magic_name = t_magic_name.lstrip(ESC_MAGIC)
     if next_input is None:
         return 'get_ipython().run_line_magic(%r, %r)' % (t_magic_name, t_magic_arg_s)
     else:
         return 'get_ipython().set_next_input(%r);get_ipython().run_line_magic(%r, %r)' % \
            (next_input, t_magic_name, t_magic_arg_s)
 
 def _tr_help(content):
     """Translate lines escaped with: ?
 
     A naked help line should fire the intro help screen (shell.show_usage())
     """
     if not content:
         return 'get_ipython().show_usage()'
 
     return _make_help_call(content, '?')
 
 def _tr_help2(content):
     """Translate lines escaped with: ??
 
     A naked help line should fire the intro help screen (shell.show_usage())
     """
     if not content:
         return 'get_ipython().show_usage()'
 
     return _make_help_call(content, '??')
 
 def _tr_magic(content):
     "Translate lines escaped with a percent sign: %"
     name, _, args = content.partition(' ')
     return 'get_ipython().run_line_magic(%r, %r)' % (name, args)
 
 def _tr_quote(content):
     "Translate lines escaped with a comma: ,"
     name, _, args = content.partition(' ')
     return '%s("%s")' % (name, '", "'.join(args.split()) )
 
 def _tr_quote2(content):
     "Translate lines escaped with a semicolon: ;"
     name, _, args = content.partition(' ')
     return '%s("%s")' % (name, args)
 
 def _tr_paren(content):
     "Translate lines escaped with a slash: /"
     name, _, args = content.partition(' ')
     return '%s(%s)' % (name, ", ".join(args.split()))
 
 tr = { ESC_SHELL  : 'get_ipython().system({!r})'.format,
        ESC_SH_CAP : 'get_ipython().getoutput({!r})'.format,
        ESC_HELP   : _tr_help,
        ESC_HELP2  : _tr_help2,
        ESC_MAGIC  : _tr_magic,
        ESC_QUOTE  : _tr_quote,
        ESC_QUOTE2 : _tr_quote2,
        ESC_PAREN  : _tr_paren }
 
 class EscapedCommand(TokenTransformBase):
     """Transformer for escaped commands like %foo, !foo, or /foo"""
     @classmethod
     def find(cls, tokens_by_line):
         """Find the first escaped command (%foo, !foo, etc.) in the cell.
         """
         for line in tokens_by_line:
             if not line:
                 continue
             ix = 0
             ll = len(line)
             while ll > ix and line[ix].type in {tokenize.INDENT, tokenize.DEDENT}:
                 ix += 1
             if ix >= ll:
                 continue
             if line[ix].string in ESCAPE_SINGLES:
                 return cls(line[ix].start)
 
     def transform(self, lines):
         """Transform an escaped line found by the ``find()`` classmethod.
         """
         start_line, start_col = self.start_line, self.start_col
 
         indent = lines[start_line][:start_col]
         end_line = find_end_of_continued_line(lines, start_line)
         line = assemble_continued_line(lines, (start_line, start_col), end_line)
 
         if len(line) > 1 and line[:2] in ESCAPE_DOUBLES:
             escape, content = line[:2], line[2:]
         else:
             escape, content = line[:1], line[1:]
 
         if escape in tr:
             call = tr[escape](content)
         else:
             call = ''
 
         lines_before = lines[:start_line]
         new_line = indent + call + '\n'
         lines_after = lines[end_line + 1:]
 
         return lines_before + [new_line] + lines_after
 
 _help_end_re = re.compile(r"""(%{0,2}
                               (?!\d)[\w*]+            # Variable name
                               (\.(?!\d)[\w*]+)*       # .etc.etc
                               )
                               (\?\??)$                # ? or ??
                               """,
                               re.VERBOSE)
 
 class HelpEnd(TokenTransformBase):
     """Transformer for help syntax: obj? and obj??"""
     # This needs to be higher priority (lower number) than EscapedCommand so
     # that inspecting magics (%foo?) works.
     priority = 5
 
     def __init__(self, start, q_locn):
         super().__init__(start)
         self.q_line = q_locn[0] - 1  # Shift from 1-indexed to 0-indexed
         self.q_col = q_locn[1]
 
     @classmethod
     def find(cls, tokens_by_line):
         """Find the first help command (foo?) in the cell.
         """
         for line in tokens_by_line:
             # Last token is NEWLINE; look at last but one
             if len(line) > 2 and line[-2].string == '?':
                 # Find the first token that's not INDENT/DEDENT
                 ix = 0
                 while line[ix].type in {tokenize.INDENT, tokenize.DEDENT}:
                     ix += 1
                 return cls(line[ix].start, line[-2].start)
 
     def transform(self, lines):
         """Transform a help command found by the ``find()`` classmethod.
         """
         piece = ''.join(lines[self.start_line:self.q_line+1])
         indent, content = piece[:self.start_col], piece[self.start_col:]
         lines_before = lines[:self.start_line]
         lines_after = lines[self.q_line + 1:]
 
         m = _help_end_re.search(content)
         if not m:
             raise SyntaxError(content)
         assert m is not None, content
         target = m.group(1)
         esc = m.group(3)
 
         # If we're mid-command, put it back on the next prompt for the user.
         next_input = None
         if (not lines_before) and (not lines_after) \
                 and content.strip() != m.group(0):
             next_input = content.rstrip('?\n')
 
         call = _make_help_call(target, esc, next_input=next_input)
         new_line = indent + call + '\n'
 
         return lines_before + [new_line] + lines_after
 
 def make_tokens_by_line(lines:List[str]):
     """Tokenize a series of lines and group tokens by line.
 
     The tokens for a multiline Python string or expression are grouped as one
     line. All lines except the last lines should keep their line ending ('\\n',
     '\\r\\n') for this to properly work. Use `.splitlines(keeplineending=True)`
     for example when passing block of text to this function.
 
     """
     # NL tokens are used inside multiline expressions, but also after blank
     # lines or comments. This is intentional - see https://bugs.python.org/issue17061
     # We want to group the former case together but split the latter, so we
     # track parentheses level, similar to the internals of tokenize.
 
     #   reexported from token on 3.7+
     NEWLINE, NL = tokenize.NEWLINE, tokenize.NL  # type: ignore
     tokens_by_line:List[List[Any]] = [[]]
     if len(lines) > 1 and not lines[0].endswith(('\n', '\r', '\r\n', '\x0b', '\x0c')):
         warnings.warn("`make_tokens_by_line` received a list of lines which do not have lineending markers ('\\n', '\\r', '\\r\\n', '\\x0b', '\\x0c'), behavior will be unspecified")
     parenlev = 0
     try:
         for token in tokenize.generate_tokens(iter(lines).__next__):
             tokens_by_line[-1].append(token)
             if (token.type == NEWLINE) \
                     or ((token.type == NL) and (parenlev <= 0)):
                 tokens_by_line.append([])
             elif token.string in {'(', '[', '{'}:
                 parenlev += 1
             elif token.string in {')', ']', '}'}:
                 if parenlev > 0:
                     parenlev -= 1
     except tokenize.TokenError:
         # Input ended in a multiline string or expression. That's OK for us.
         pass
 
 
     if not tokens_by_line[-1]:
         tokens_by_line.pop()
 
 
     return tokens_by_line
 
 def show_linewise_tokens(s: str):
     """For investigation and debugging"""
     if not s.endswith('\n'):
         s += '\n'
     lines = s.splitlines(keepends=True)
     for line in make_tokens_by_line(lines):
         print("Line -------")
         for tokinfo in line:
             print(" ", tokinfo)
 
 # Arbitrary limit to prevent getting stuck in infinite loops
 TRANSFORM_LOOP_LIMIT = 500
 
 class TransformerManager:
     """Applies various transformations to a cell or code block.
 
     The key methods for external use are ``transform_cell()``
     and ``check_complete()``.
     """
     def __init__(self):
         self.cleanup_transforms = [
             leading_empty_lines,
             leading_indent,
             classic_prompt,
             ipython_prompt,
         ]
         self.line_transforms = [
             cell_magic,
         ]
         self.token_transformers = [
             MagicAssign,
             SystemAssign,
             EscapedCommand,
             HelpEnd,
         ]
 
     def do_one_token_transform(self, lines):
         """Find and run the transform earliest in the code.
 
         Returns (changed, lines).
 
         This method is called repeatedly until changed is False, indicating
         that all available transformations are complete.
 
         The tokens following IPython special syntax might not be valid, so
         the transformed code is retokenised every time to identify the next
         piece of special syntax. Hopefully long code cells are mostly valid
         Python, not using lots of IPython special syntax, so this shouldn't be
         a performance issue.
         """
         tokens_by_line = make_tokens_by_line(lines)
         candidates = []
         for transformer_cls in self.token_transformers:
             transformer = transformer_cls.find(tokens_by_line)
             if transformer:
                 candidates.append(transformer)
 
         if not candidates:
             # Nothing to transform
             return False, lines
         ordered_transformers = sorted(candidates, key=TokenTransformBase.sortby)
         for transformer in ordered_transformers:
             try:
                 return True, transformer.transform(lines)
             except SyntaxError:
                 pass
         return False, lines
 
     def do_token_transforms(self, lines):
         for _ in range(TRANSFORM_LOOP_LIMIT):
             changed, lines = self.do_one_token_transform(lines)
             if not changed:
                 return lines
 
         raise RuntimeError("Input transformation still changing after "
                            "%d iterations. Aborting." % TRANSFORM_LOOP_LIMIT)
 
     def transform_cell(self, cell: str) -> str:
         """Transforms a cell of input code"""
         if not cell.endswith('\n'):
             cell += '\n'  # Ensure the cell has a trailing newline
         lines = cell.splitlines(keepends=True)
         for transform in self.cleanup_transforms + self.line_transforms:
             lines = transform(lines)
 
         lines = self.do_token_transforms(lines)
         return ''.join(lines)
 
     def check_complete(self, cell: str):
         """Return whether a block of code is ready to execute, or should be continued
 
         Parameters
         ----------
         source : string
           Python input code, which can be multiline.
 
         Returns
         -------
         status : str
           One of 'complete', 'incomplete', or 'invalid' if source is not a
           prefix of valid code.
         indent_spaces : int or None
           The number of spaces by which to indent the next line of code. If
           status is not 'incomplete', this is None.
         """
         # Remember if the lines ends in a new line.
         ends_with_newline = False
         for character in reversed(cell):
             if character == '\n':
                 ends_with_newline = True
                 break
             elif character.strip():
                 break
             else:
                 continue
 
         if not ends_with_newline:
             # Append an newline for consistent tokenization
             # See https://bugs.python.org/issue33899
             cell += '\n'
 
         lines = cell.splitlines(keepends=True)
 
         if not lines:
             return 'complete', None
 
         if lines[-1].endswith('\\'):
             # Explicit backslash continuation
             return 'incomplete', find_last_indent(lines)
 
         try:
             for transform in self.cleanup_transforms:
-                lines = transform(lines)
+                if not getattr(transform, 'has_side_effects', False):
+                    lines = transform(lines)
         except SyntaxError:
             return 'invalid', None
 
         if lines[0].startswith('%%'):
             # Special case for cell magics - completion marked by blank line
             if lines[-1].strip():
                 return 'incomplete', find_last_indent(lines)
             else:
                 return 'complete', None
 
         try:
             for transform in self.line_transforms:
-                lines = transform(lines)
+                if not getattr(transform, 'has_side_effects', False):
+                    lines = transform(lines)
             lines = self.do_token_transforms(lines)
         except SyntaxError:
             return 'invalid', None
 
         tokens_by_line = make_tokens_by_line(lines)
 
         if not tokens_by_line:
             return 'incomplete', find_last_indent(lines)
 
         if tokens_by_line[-1][-1].type != tokenize.ENDMARKER:
             # We're in a multiline string or expression
             return 'incomplete', find_last_indent(lines)
 
         newline_types = {tokenize.NEWLINE, tokenize.COMMENT, tokenize.ENDMARKER} # type: ignore
 
         # Pop the last line which only contains DEDENTs and ENDMARKER
         last_token_line = None
         if {t.type for t in tokens_by_line[-1]} in [
             {tokenize.DEDENT, tokenize.ENDMARKER},
             {tokenize.ENDMARKER}
         ] and len(tokens_by_line) > 1:
             last_token_line = tokens_by_line.pop()
 
         while tokens_by_line[-1] and tokens_by_line[-1][-1].type in newline_types:
             tokens_by_line[-1].pop()
 
         if not tokens_by_line[-1]:
             return 'incomplete', find_last_indent(lines)
 
         if tokens_by_line[-1][-1].string == ':':
             # The last line starts a block (e.g. 'if foo:')
             ix = 0
             while tokens_by_line[-1][ix].type in {tokenize.INDENT, tokenize.DEDENT}:
                 ix += 1
 
             indent = tokens_by_line[-1][ix].start[1]
             return 'incomplete', indent + 4
 
         if tokens_by_line[-1][0].line.endswith('\\'):
             return 'incomplete', None
 
         # At this point, our checks think the code is complete (or invalid).
         # We'll use codeop.compile_command to check this with the real parser
         try:
             with warnings.catch_warnings():
                 warnings.simplefilter('error', SyntaxWarning)
                 res = compile_command(''.join(lines), symbol='exec')
         except (SyntaxError, OverflowError, ValueError, TypeError,
                 MemoryError, SyntaxWarning):
             return 'invalid', None
         else:
             if res is None:
                 return 'incomplete', find_last_indent(lines)
 
         if last_token_line and last_token_line[0].type == tokenize.DEDENT:
             if ends_with_newline:
                 return 'complete', None
             return 'incomplete', find_last_indent(lines)
 
         # If there's a blank line at the end, assume we're ready to execute
         if not lines[-1].strip():
             return 'complete', None
 
         return 'complete', None
 
 
 def find_last_indent(lines):
     m = _indent_re.match(lines[-1])
     if not m:
         return 0
     return len(m.group(0).replace('\t', ' '*4))

IPython/core/tests/test_inputtransformer2.py

@@ -1,292 +1,326 b''
 """Tests for the token-based transformers in IPython.core.inputtransformer2
 
 Line-based transformers are the simpler ones; token-based transformers are
 more complex. See test_inputtransformer2_line for tests for line-based
 transformations.
 """
 import nose.tools as nt
 import string
 
 from IPython.core import inputtransformer2 as ipt2
 from IPython.core.inputtransformer2 import make_tokens_by_line, _find_assign_op
 
 from textwrap import dedent
 
 MULTILINE_MAGIC = ("""\
 a = f()
 %foo \\
 bar
 g()
 """.splitlines(keepends=True), (2, 0), """\
 a = f()
 get_ipython().run_line_magic('foo', ' bar')
 g()
 """.splitlines(keepends=True))
 
 INDENTED_MAGIC = ("""\
 for a in range(5):
     %ls
 """.splitlines(keepends=True), (2, 4), """\
 for a in range(5):
     get_ipython().run_line_magic('ls', '')
 """.splitlines(keepends=True))
 
 MULTILINE_MAGIC_ASSIGN = ("""\
 a = f()
 b = %foo \\
   bar
 g()
 """.splitlines(keepends=True), (2, 4), """\
 a = f()
 b = get_ipython().run_line_magic('foo', '   bar')
 g()
 """.splitlines(keepends=True))
 
 MULTILINE_SYSTEM_ASSIGN = ("""\
 a = f()
 b = !foo \\
   bar
 g()
 """.splitlines(keepends=True), (2, 4), """\
 a = f()
 b = get_ipython().getoutput('foo    bar')
 g()
 """.splitlines(keepends=True))
 
 #####
 
 MULTILINE_SYSTEM_ASSIGN_AFTER_DEDENT = ("""\
 def test():
   for i in range(1):
     print(i)
   res =! ls
 """.splitlines(keepends=True), (4, 7), '''\
 def test():
   for i in range(1):
     print(i)
   res =get_ipython().getoutput(\' ls\')
 '''.splitlines(keepends=True))
 
 ######
 
 AUTOCALL_QUOTE = (
     [",f 1 2 3\n"], (1, 0),
     ['f("1", "2", "3")\n']
 )
 
 AUTOCALL_QUOTE2 = (
     [";f 1 2 3\n"], (1, 0),
     ['f("1 2 3")\n']
 )
 
 AUTOCALL_PAREN = (
     ["/f 1 2 3\n"], (1, 0),
     ['f(1, 2, 3)\n']
 )
 
 SIMPLE_HELP = (
     ["foo?\n"], (1, 0),
     ["get_ipython().run_line_magic('pinfo', 'foo')\n"]
 )
 
 DETAILED_HELP = (
     ["foo??\n"], (1, 0),
     ["get_ipython().run_line_magic('pinfo2', 'foo')\n"]
 )
 
 MAGIC_HELP = (
     ["%foo?\n"], (1, 0),
     ["get_ipython().run_line_magic('pinfo', '%foo')\n"]
 )
 
 HELP_IN_EXPR = (
     ["a = b + c?\n"], (1, 0),
     ["get_ipython().set_next_input('a = b + c');"
      "get_ipython().run_line_magic('pinfo', 'c')\n"]
 )
 
 HELP_CONTINUED_LINE = ("""\
 a = \\
 zip?
 """.splitlines(keepends=True), (1, 0),
 [r"get_ipython().set_next_input('a = \\\nzip');get_ipython().run_line_magic('pinfo', 'zip')" + "\n"]
 )
 
 HELP_MULTILINE = ("""\
 (a,
 b) = zip?
 """.splitlines(keepends=True), (1, 0),
 [r"get_ipython().set_next_input('(a,\nb) = zip');get_ipython().run_line_magic('pinfo', 'zip')" + "\n"]
 )
 
 HELP_UNICODE = (
     ["π.foo?\n"], (1, 0),
     ["get_ipython().run_line_magic('pinfo', 'π.foo')\n"]
 )
 
 
 def null_cleanup_transformer(lines):
     """
     A cleanup transform that returns an empty list.
     """
     return []
 
 def check_make_token_by_line_never_ends_empty():
     """
     Check that not sequence of single or double characters ends up leading to en empty list of tokens
     """
     from string import printable
     for c in printable:
         nt.assert_not_equal(make_tokens_by_line(c)[-1], [])
         for k in printable:
             nt.assert_not_equal(make_tokens_by_line(c+k)[-1], [])
 
 def check_find(transformer, case, match=True):
     sample, expected_start, _  = case
     tbl = make_tokens_by_line(sample)
     res = transformer.find(tbl)
     if match:
         # start_line is stored 0-indexed, expected values are 1-indexed
         nt.assert_equal((res.start_line+1, res.start_col), expected_start)
         return res
     else:
         nt.assert_is(res, None)
 
 def check_transform(transformer_cls, case):
     lines, start, expected = case
     transformer = transformer_cls(start)
     nt.assert_equal(transformer.transform(lines), expected)
 
 def test_continued_line():
     lines = MULTILINE_MAGIC_ASSIGN[0]
     nt.assert_equal(ipt2.find_end_of_continued_line(lines, 1), 2)
 
     nt.assert_equal(ipt2.assemble_continued_line(lines, (1, 5), 2), "foo    bar")
 
 def test_find_assign_magic():
     check_find(ipt2.MagicAssign, MULTILINE_MAGIC_ASSIGN)
     check_find(ipt2.MagicAssign, MULTILINE_SYSTEM_ASSIGN, match=False)
     check_find(ipt2.MagicAssign, MULTILINE_SYSTEM_ASSIGN_AFTER_DEDENT, match=False)
 
 def test_transform_assign_magic():
     check_transform(ipt2.MagicAssign, MULTILINE_MAGIC_ASSIGN)
 
 def test_find_assign_system():
     check_find(ipt2.SystemAssign, MULTILINE_SYSTEM_ASSIGN)
     check_find(ipt2.SystemAssign, MULTILINE_SYSTEM_ASSIGN_AFTER_DEDENT)
     check_find(ipt2.SystemAssign, (["a =  !ls\n"], (1, 5), None))
     check_find(ipt2.SystemAssign, (["a=!ls\n"], (1, 2), None))
     check_find(ipt2.SystemAssign, MULTILINE_MAGIC_ASSIGN, match=False)
 
 def test_transform_assign_system():
     check_transform(ipt2.SystemAssign, MULTILINE_SYSTEM_ASSIGN)
     check_transform(ipt2.SystemAssign, MULTILINE_SYSTEM_ASSIGN_AFTER_DEDENT)
 
 def test_find_magic_escape():
     check_find(ipt2.EscapedCommand, MULTILINE_MAGIC)
     check_find(ipt2.EscapedCommand, INDENTED_MAGIC)
     check_find(ipt2.EscapedCommand, MULTILINE_MAGIC_ASSIGN, match=False)
 
 def test_transform_magic_escape():
     check_transform(ipt2.EscapedCommand, MULTILINE_MAGIC)
     check_transform(ipt2.EscapedCommand, INDENTED_MAGIC)
 
 def test_find_autocalls():
     for case in [AUTOCALL_QUOTE, AUTOCALL_QUOTE2, AUTOCALL_PAREN]:
         print("Testing %r" % case[0])
         check_find(ipt2.EscapedCommand, case)
 
 def test_transform_autocall():
     for case in [AUTOCALL_QUOTE, AUTOCALL_QUOTE2, AUTOCALL_PAREN]:
         print("Testing %r" % case[0])
         check_transform(ipt2.EscapedCommand, case)
 
 def test_find_help():
     for case in [SIMPLE_HELP, DETAILED_HELP, MAGIC_HELP, HELP_IN_EXPR]:
         check_find(ipt2.HelpEnd, case)
 
     tf = check_find(ipt2.HelpEnd, HELP_CONTINUED_LINE)
     nt.assert_equal(tf.q_line, 1)
     nt.assert_equal(tf.q_col, 3)
 
     tf = check_find(ipt2.HelpEnd, HELP_MULTILINE)
     nt.assert_equal(tf.q_line, 1)
     nt.assert_equal(tf.q_col, 8)
 
     # ? in a comment does not trigger help
     check_find(ipt2.HelpEnd, (["foo # bar?\n"], None, None), match=False)
     # Nor in a string
     check_find(ipt2.HelpEnd, (["foo = '''bar?\n"], None, None), match=False)
 
 def test_transform_help():
     tf = ipt2.HelpEnd((1, 0), (1, 9))
     nt.assert_equal(tf.transform(HELP_IN_EXPR[0]), HELP_IN_EXPR[2])
 
     tf = ipt2.HelpEnd((1, 0), (2, 3))
     nt.assert_equal(tf.transform(HELP_CONTINUED_LINE[0]), HELP_CONTINUED_LINE[2])
 
     tf = ipt2.HelpEnd((1, 0), (2, 8))
     nt.assert_equal(tf.transform(HELP_MULTILINE[0]), HELP_MULTILINE[2])
 
     tf = ipt2.HelpEnd((1, 0), (1, 0))
     nt.assert_equal(tf.transform(HELP_UNICODE[0]), HELP_UNICODE[2])
 
 def test_find_assign_op_dedent():
     """
     be careful that empty token like dedent are not counted as parens
     """
     class Tk:
         def __init__(self, s):
             self.string = s
 
     nt.assert_equal(_find_assign_op([Tk(s) for s in ('','a','=','b')]), 2)
     nt.assert_equal(_find_assign_op([Tk(s) for s in ('','(', 'a','=','b', ')', '=' ,'5')]), 6)
 
 def test_check_complete():
     cc = ipt2.TransformerManager().check_complete
     nt.assert_equal(cc("a = 1"), ('complete', None))
     nt.assert_equal(cc("for a in range(5):"), ('incomplete', 4))
     nt.assert_equal(cc("for a in range(5):\n    if a > 0:"), ('incomplete', 8))
     nt.assert_equal(cc("raise = 2"), ('invalid', None))
     nt.assert_equal(cc("a = [1,\n2,"), ('incomplete', 0))
     nt.assert_equal(cc(")"), ('incomplete', 0))
     nt.assert_equal(cc("\\\r\n"), ('incomplete', 0))
     nt.assert_equal(cc("a = '''\n   hi"), ('incomplete', 3))
     nt.assert_equal(cc("def a():\n x=1\n global x"), ('invalid', None))
     nt.assert_equal(cc("a \\ "), ('invalid', None))  # Nothing allowed after backslash
     nt.assert_equal(cc("1\\\n+2"), ('complete', None))
     nt.assert_equal(cc("exit"), ('complete', None))
 
     example = dedent("""
         if True:
             a=1""" )
 
     nt.assert_equal(cc(example), ('incomplete', 4))
     nt.assert_equal(cc(example+'\n'), ('complete', None))
     nt.assert_equal(cc(example+'\n    '), ('complete', None))
 
     # no need to loop on all the letters/numbers.
     short = '12abAB'+string.printable[62:]
     for c in short:
         # test does not raise:
         cc(c)
         for k in short:
             cc(c+k)
 
     nt.assert_equal(cc("def f():\n  x=0\n  \\\n  "), ('incomplete', 2))
 
 def test_check_complete_II():
     """
     Test that multiple line strings are properly handled.
 
     Separate test function for convenience
 
     """
     cc = ipt2.TransformerManager().check_complete
     nt.assert_equal(cc('''def foo():\n    """'''), ('incomplete', 4))
 
 
 def test_null_cleanup_transformer():
     manager = ipt2.TransformerManager()
     manager.cleanup_transforms.insert(0, null_cleanup_transformer)
-    nt.assert_is(manager.transform_cell(""), "")
+    assert  manager.transform_cell("") == ""
+
+
+
+
+def test_side_effects_I():
+    count = 0
+    def counter(lines):
+        nonlocal count
+        count += 1
+        return lines
+
+    counter.has_side_effects = True
+
+    manager = ipt2.TransformerManager()
+    manager.cleanup_transforms.insert(0, counter)
+    assert manager.check_complete("a=1\n") == ('complete', None)
+    assert count == 0
+
+
+
+
+def test_side_effects_II():
+    count = 0
+    def counter(lines):
+        nonlocal count
+        count += 1
+        return lines
+
+    counter.has_side_effects = True
+
+    manager = ipt2.TransformerManager()
+    manager.line_transforms.insert(0, counter)
+    assert manager.check_complete("b=1\n") == ('complete', None)
+    assert count == 0

IPython/core/tests/test_interactiveshell.py

@@ -1,1018 +1,1042 b''
 # -*- coding: utf-8 -*-
 """Tests for the key interactiveshell module.
 
 Historically the main classes in interactiveshell have been under-tested.  This
 module should grow as many single-method tests as possible to trap many of the
 recurring bugs we seem to encounter with high-level interaction.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import asyncio
 import ast
 import os
 import signal
 import shutil
 import sys
 import tempfile
 import unittest
 from unittest import mock
 
 from os.path import join
 
 import nose.tools as nt
 
 from IPython.core.error import InputRejected
 from IPython.core.inputtransformer import InputTransformer
 from IPython.core import interactiveshell
 from IPython.testing.decorators import (
     skipif, skip_win32, onlyif_unicode_paths, onlyif_cmds_exist,
 )
 from IPython.testing import tools as tt
 from IPython.utils.process import find_cmd
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 # This is used by every single test, no point repeating it ad nauseam
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 class DerivedInterrupt(KeyboardInterrupt):
     pass
 
 class InteractiveShellTestCase(unittest.TestCase):
     def test_naked_string_cells(self):
         """Test that cells with only naked strings are fully executed"""
         # First, single-line inputs
         ip.run_cell('"a"\n')
         self.assertEqual(ip.user_ns['_'], 'a')
         # And also multi-line cells
         ip.run_cell('"""a\nb"""\n')
         self.assertEqual(ip.user_ns['_'], 'a\nb')
 
     def test_run_empty_cell(self):
         """Just make sure we don't get a horrible error with a blank
         cell of input. Yes, I did overlook that."""
         old_xc = ip.execution_count
         res = ip.run_cell('')
         self.assertEqual(ip.execution_count, old_xc)
         self.assertEqual(res.execution_count, None)
 
     def test_run_cell_multiline(self):
         """Multi-block, multi-line cells must execute correctly.
         """
         src = '\n'.join(["x=1",
                          "y=2",
                          "if 1:",
                          "    x += 1",
                          "    y += 1",])
         res = ip.run_cell(src)
         self.assertEqual(ip.user_ns['x'], 2)
         self.assertEqual(ip.user_ns['y'], 3)
         self.assertEqual(res.success, True)
         self.assertEqual(res.result, None)
 
     def test_multiline_string_cells(self):
         "Code sprinkled with multiline strings should execute (GH-306)"
         ip.run_cell('tmp=0')
         self.assertEqual(ip.user_ns['tmp'], 0)
         res = ip.run_cell('tmp=1;"""a\nb"""\n')
         self.assertEqual(ip.user_ns['tmp'], 1)
         self.assertEqual(res.success, True)
         self.assertEqual(res.result, "a\nb")
 
     def test_dont_cache_with_semicolon(self):
         "Ending a line with semicolon should not cache the returned object (GH-307)"
         oldlen = len(ip.user_ns['Out'])
         for cell in ['1;', '1;1;']:
             res = ip.run_cell(cell, store_history=True)
             newlen = len(ip.user_ns['Out'])
             self.assertEqual(oldlen, newlen)
             self.assertIsNone(res.result)
         i = 0
         #also test the default caching behavior
         for cell in ['1', '1;1']:
             ip.run_cell(cell, store_history=True)
             newlen = len(ip.user_ns['Out'])
             i += 1
             self.assertEqual(oldlen+i, newlen)
 
     def test_syntax_error(self):
         res = ip.run_cell("raise = 3")
         self.assertIsInstance(res.error_before_exec, SyntaxError)
 
     def test_In_variable(self):
         "Verify that In variable grows with user input (GH-284)"
         oldlen = len(ip.user_ns['In'])
         ip.run_cell('1;', store_history=True)
         newlen = len(ip.user_ns['In'])
         self.assertEqual(oldlen+1, newlen)
         self.assertEqual(ip.user_ns['In'][-1],'1;')
         
     def test_magic_names_in_string(self):
         ip.run_cell('a = """\n%exit\n"""')
         self.assertEqual(ip.user_ns['a'], '\n%exit\n')
     
     def test_trailing_newline(self):
         """test that running !(command) does not raise a SyntaxError"""
         ip.run_cell('!(true)\n', False)
         ip.run_cell('!(true)\n\n\n', False)
     
     def test_gh_597(self):
         """Pretty-printing lists of objects with non-ascii reprs may cause
         problems."""
         class Spam(object):
             def __repr__(self):
                 return "\xe9"*50
         import IPython.core.formatters
         f = IPython.core.formatters.PlainTextFormatter()
         f([Spam(),Spam()])
     
 
     def test_future_flags(self):
         """Check that future flags are used for parsing code (gh-777)"""
         ip.run_cell('from __future__ import barry_as_FLUFL')
         try:
             ip.run_cell('prfunc_return_val = 1 <> 2')
             assert 'prfunc_return_val' in ip.user_ns
         finally:
             # Reset compiler flags so we don't mess up other tests.
             ip.compile.reset_compiler_flags()
 
     def test_can_pickle(self):
         "Can we pickle objects defined interactively (GH-29)"
         ip = get_ipython()
         ip.reset()
         ip.run_cell(("class Mylist(list):\n"
                      "    def __init__(self,x=[]):\n"
                      "        list.__init__(self,x)"))
         ip.run_cell("w=Mylist([1,2,3])")
         
         from pickle import dumps
         
         # We need to swap in our main module - this is only necessary
         # inside the test framework, because IPython puts the interactive module
         # in place (but the test framework undoes this).
         _main = sys.modules['__main__']
         sys.modules['__main__'] = ip.user_module
         try:
             res = dumps(ip.user_ns["w"])
         finally:
             sys.modules['__main__'] = _main
         self.assertTrue(isinstance(res, bytes))
         
     def test_global_ns(self):
         "Code in functions must be able to access variables outside them."
         ip = get_ipython()
         ip.run_cell("a = 10")
         ip.run_cell(("def f(x):\n"
                      "    return x + a"))
         ip.run_cell("b = f(12)")
         self.assertEqual(ip.user_ns["b"], 22)
 
     def test_bad_custom_tb(self):
         """Check that InteractiveShell is protected from bad custom exception handlers"""
         ip.set_custom_exc((IOError,), lambda etype,value,tb: 1/0)
         self.assertEqual(ip.custom_exceptions, (IOError,))
         with tt.AssertPrints("Custom TB Handler failed", channel='stderr'):
             ip.run_cell(u'raise IOError("foo")')
         self.assertEqual(ip.custom_exceptions, ())
 
     def test_bad_custom_tb_return(self):
         """Check that InteractiveShell is protected from bad return types in custom exception handlers"""
         ip.set_custom_exc((NameError,),lambda etype,value,tb, tb_offset=None: 1)
         self.assertEqual(ip.custom_exceptions, (NameError,))
         with tt.AssertPrints("Custom TB Handler failed", channel='stderr'):
             ip.run_cell(u'a=abracadabra')
         self.assertEqual(ip.custom_exceptions, ())
 
     def test_drop_by_id(self):
         myvars = {"a":object(), "b":object(), "c": object()}
         ip.push(myvars, interactive=False)
         for name in myvars:
             assert name in ip.user_ns, name
             assert name in ip.user_ns_hidden, name
         ip.user_ns['b'] = 12
         ip.drop_by_id(myvars)
         for name in ["a", "c"]:
             assert name not in ip.user_ns, name
             assert name not in ip.user_ns_hidden, name
         assert ip.user_ns['b'] == 12
         ip.reset()
 
     def test_var_expand(self):
         ip.user_ns['f'] = u'Ca\xf1o'
         self.assertEqual(ip.var_expand(u'echo $f'), u'echo Ca\xf1o')
         self.assertEqual(ip.var_expand(u'echo {f}'), u'echo Ca\xf1o')
         self.assertEqual(ip.var_expand(u'echo {f[:-1]}'), u'echo Ca\xf1')
         self.assertEqual(ip.var_expand(u'echo {1*2}'), u'echo 2')
         
         self.assertEqual(ip.var_expand(u"grep x | awk '{print $1}'"), u"grep x | awk '{print $1}'")
 
         ip.user_ns['f'] = b'Ca\xc3\xb1o'
         # This should not raise any exception:
         ip.var_expand(u'echo $f')
    
     def test_var_expand_local(self):
         """Test local variable expansion in !system and %magic calls"""
         # !system
         ip.run_cell('def test():\n'
                     '    lvar = "ttt"\n'
                     '    ret = !echo {lvar}\n'
                     '    return ret[0]\n')
         res = ip.user_ns['test']()
         nt.assert_in('ttt', res)
         
         # %magic
         ip.run_cell('def makemacro():\n'
                     '    macroname = "macro_var_expand_locals"\n'
                     '    %macro {macroname} codestr\n')
         ip.user_ns['codestr'] = "str(12)"
         ip.run_cell('makemacro()')
         nt.assert_in('macro_var_expand_locals', ip.user_ns)
     
     def test_var_expand_self(self):
         """Test variable expansion with the name 'self', which was failing.
         
         See https://github.com/ipython/ipython/issues/1878#issuecomment-7698218
         """
         ip.run_cell('class cTest:\n'
                     '  classvar="see me"\n'
                     '  def test(self):\n'
                     '    res = !echo Variable: {self.classvar}\n'
                     '    return res[0]\n')
         nt.assert_in('see me', ip.user_ns['cTest']().test())
     
     def test_bad_var_expand(self):
         """var_expand on invalid formats shouldn't raise"""
         # SyntaxError
         self.assertEqual(ip.var_expand(u"{'a':5}"), u"{'a':5}")
         # NameError
         self.assertEqual(ip.var_expand(u"{asdf}"), u"{asdf}")
         # ZeroDivisionError
         self.assertEqual(ip.var_expand(u"{1/0}"), u"{1/0}")
     
     def test_silent_postexec(self):
         """run_cell(silent=True) doesn't invoke pre/post_run_cell callbacks"""
         pre_explicit = mock.Mock()
         pre_always = mock.Mock()
         post_explicit = mock.Mock()
         post_always = mock.Mock()
         all_mocks = [pre_explicit, pre_always, post_explicit, post_always]
         
         ip.events.register('pre_run_cell', pre_explicit)
         ip.events.register('pre_execute', pre_always)
         ip.events.register('post_run_cell', post_explicit)
         ip.events.register('post_execute', post_always)
         
         try:
             ip.run_cell("1", silent=True)
             assert pre_always.called
             assert not pre_explicit.called
             assert post_always.called
             assert not post_explicit.called
             # double-check that non-silent exec did what we expected
             # silent to avoid
             ip.run_cell("1")
             assert pre_explicit.called
             assert post_explicit.called
             info, = pre_explicit.call_args[0]
             result, = post_explicit.call_args[0]
             self.assertEqual(info, result.info)
             # check that post hooks are always called
             [m.reset_mock() for m in all_mocks]
             ip.run_cell("syntax error")
             assert pre_always.called
             assert pre_explicit.called
             assert post_always.called
             assert post_explicit.called
             info, = pre_explicit.call_args[0]
             result, = post_explicit.call_args[0]
             self.assertEqual(info, result.info)
         finally:
             # remove post-exec
             ip.events.unregister('pre_run_cell', pre_explicit)
             ip.events.unregister('pre_execute', pre_always)
             ip.events.unregister('post_run_cell', post_explicit)
             ip.events.unregister('post_execute', post_always)
     
     def test_silent_noadvance(self):
         """run_cell(silent=True) doesn't advance execution_count"""
         ec = ip.execution_count
         # silent should force store_history=False
         ip.run_cell("1", store_history=True, silent=True)
         
         self.assertEqual(ec, ip.execution_count)
         # double-check that non-silent exec did what we expected
         # silent to avoid
         ip.run_cell("1", store_history=True)
         self.assertEqual(ec+1, ip.execution_count)
     
     def test_silent_nodisplayhook(self):
         """run_cell(silent=True) doesn't trigger displayhook"""
         d = dict(called=False)
         
         trap = ip.display_trap
         save_hook = trap.hook
         
         def failing_hook(*args, **kwargs):
             d['called'] = True
         
         try:
             trap.hook = failing_hook
             res = ip.run_cell("1", silent=True)
             self.assertFalse(d['called'])
             self.assertIsNone(res.result)
             # double-check that non-silent exec did what we expected
             # silent to avoid
             ip.run_cell("1")
             self.assertTrue(d['called'])
         finally:
             trap.hook = save_hook
 
     def test_ofind_line_magic(self):
         from IPython.core.magic import register_line_magic
         
         @register_line_magic
         def lmagic(line):
             "A line magic"
 
         # Get info on line magic
         lfind = ip._ofind('lmagic')
         info = dict(found=True, isalias=False, ismagic=True,
                     namespace = 'IPython internal', obj= lmagic.__wrapped__,
                     parent = None)
         nt.assert_equal(lfind, info)
         
     def test_ofind_cell_magic(self):
         from IPython.core.magic import register_cell_magic
         
         @register_cell_magic
         def cmagic(line, cell):
             "A cell magic"
 
         # Get info on cell magic
         find = ip._ofind('cmagic')
         info = dict(found=True, isalias=False, ismagic=True,
                     namespace = 'IPython internal', obj= cmagic.__wrapped__,
                     parent = None)
         nt.assert_equal(find, info)
 
     def test_ofind_property_with_error(self):
         class A(object):
             @property
             def foo(self):
                 raise NotImplementedError()
         a = A()
 
         found = ip._ofind('a.foo', [('locals', locals())])
         info = dict(found=True, isalias=False, ismagic=False,
                     namespace='locals', obj=A.foo, parent=a)
         nt.assert_equal(found, info)
 
     def test_ofind_multiple_attribute_lookups(self):
         class A(object):
             @property
             def foo(self):
                 raise NotImplementedError()
 
         a = A()
         a.a = A()
         a.a.a = A()
 
         found = ip._ofind('a.a.a.foo', [('locals', locals())])
         info = dict(found=True, isalias=False, ismagic=False,
                     namespace='locals', obj=A.foo, parent=a.a.a)
         nt.assert_equal(found, info)
 
     def test_ofind_slotted_attributes(self):
         class A(object):
             __slots__ = ['foo']
             def __init__(self):
                 self.foo = 'bar'
 
         a = A()
         found = ip._ofind('a.foo', [('locals', locals())])
         info = dict(found=True, isalias=False, ismagic=False,
                     namespace='locals', obj=a.foo, parent=a)
         nt.assert_equal(found, info)
 
         found = ip._ofind('a.bar', [('locals', locals())])
         info = dict(found=False, isalias=False, ismagic=False,
                     namespace=None, obj=None, parent=a)
         nt.assert_equal(found, info)
 
     def test_ofind_prefers_property_to_instance_level_attribute(self):
         class A(object):
             @property
             def foo(self):
                 return 'bar'
         a = A()
         a.__dict__['foo'] = 'baz'
         nt.assert_equal(a.foo, 'bar')
         found = ip._ofind('a.foo', [('locals', locals())])
         nt.assert_is(found['obj'], A.foo)
 
     def test_custom_syntaxerror_exception(self):
         called = []
         def my_handler(shell, etype, value, tb, tb_offset=None):
             called.append(etype)
             shell.showtraceback((etype, value, tb), tb_offset=tb_offset)
 
         ip.set_custom_exc((SyntaxError,), my_handler)
         try:
             ip.run_cell("1f")
             # Check that this was called, and only once.
             self.assertEqual(called, [SyntaxError])
         finally:
             # Reset the custom exception hook
             ip.set_custom_exc((), None)
 
     def test_custom_exception(self):
         called = []
         def my_handler(shell, etype, value, tb, tb_offset=None):
             called.append(etype)
             shell.showtraceback((etype, value, tb), tb_offset=tb_offset)
         
         ip.set_custom_exc((ValueError,), my_handler)
         try:
             res = ip.run_cell("raise ValueError('test')")
             # Check that this was called, and only once.
             self.assertEqual(called, [ValueError])
             # Check that the error is on the result object
             self.assertIsInstance(res.error_in_exec, ValueError)
         finally:
             # Reset the custom exception hook
             ip.set_custom_exc((), None)
     
     def test_mktempfile(self):
         filename = ip.mktempfile()
         # Check that we can open the file again on Windows
         with open(filename, 'w') as f:
             f.write('abc')
 
         filename = ip.mktempfile(data='blah')
         with open(filename, 'r') as f:
             self.assertEqual(f.read(), 'blah')
 
     def test_new_main_mod(self):
         # Smoketest to check that this accepts a unicode module name
         name = u'jiefmw'
         mod = ip.new_main_mod(u'%s.py' % name, name)
         self.assertEqual(mod.__name__, name)
 
     def test_get_exception_only(self):
         try:
             raise KeyboardInterrupt
         except KeyboardInterrupt:
             msg = ip.get_exception_only()
         self.assertEqual(msg, 'KeyboardInterrupt\n')
 
         try:
             raise DerivedInterrupt("foo")
         except KeyboardInterrupt:
             msg = ip.get_exception_only()
         self.assertEqual(msg, 'IPython.core.tests.test_interactiveshell.DerivedInterrupt: foo\n')
 
     def test_inspect_text(self):
         ip.run_cell('a = 5')
         text = ip.object_inspect_text('a')
         self.assertIsInstance(text, str)
 
     def test_last_execution_result(self):
         """ Check that last execution result gets set correctly (GH-10702) """
         result = ip.run_cell('a = 5; a')
         self.assertTrue(ip.last_execution_succeeded)
         self.assertEqual(ip.last_execution_result.result, 5)
 
         result = ip.run_cell('a = x_invalid_id_x')
         self.assertFalse(ip.last_execution_succeeded)
         self.assertFalse(ip.last_execution_result.success)
         self.assertIsInstance(ip.last_execution_result.error_in_exec, NameError)
 
     def test_reset_aliasing(self):
         """ Check that standard posix aliases work after %reset. """
         if os.name != 'posix':
             return
 
         ip.reset()
         for cmd in ('clear', 'more', 'less', 'man'):
             res = ip.run_cell('%' + cmd)
             self.assertEqual(res.success, True)
 
 
 class TestSafeExecfileNonAsciiPath(unittest.TestCase):
 
     @onlyif_unicode_paths
     def setUp(self):
         self.BASETESTDIR = tempfile.mkdtemp()
         self.TESTDIR = join(self.BASETESTDIR, u"åäö")
         os.mkdir(self.TESTDIR)
         with open(join(self.TESTDIR, u"åäötestscript.py"), "w") as sfile:
             sfile.write("pass\n")
         self.oldpath = os.getcwd()
         os.chdir(self.TESTDIR)
         self.fname = u"åäötestscript.py"
 
     def tearDown(self):
         os.chdir(self.oldpath)
         shutil.rmtree(self.BASETESTDIR)
 
     @onlyif_unicode_paths
     def test_1(self):
         """Test safe_execfile with non-ascii path
         """
         ip.safe_execfile(self.fname, {}, raise_exceptions=True)
 
 class ExitCodeChecks(tt.TempFileMixin):
 
     def setUp(self):
         self.system = ip.system_raw
 
     def test_exit_code_ok(self):
         self.system('exit 0')
         self.assertEqual(ip.user_ns['_exit_code'], 0)
 
     def test_exit_code_error(self):
         self.system('exit 1')
         self.assertEqual(ip.user_ns['_exit_code'], 1)
     
     @skipif(not hasattr(signal, 'SIGALRM'))
     def test_exit_code_signal(self):
         self.mktmp("import signal, time\n"
                    "signal.setitimer(signal.ITIMER_REAL, 0.1)\n"
                    "time.sleep(1)\n")
         self.system("%s %s" % (sys.executable, self.fname))
         self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGALRM)
     
     @onlyif_cmds_exist("csh")
     def test_exit_code_signal_csh(self):
         SHELL = os.environ.get('SHELL', None)
         os.environ['SHELL'] = find_cmd("csh")
         try:
             self.test_exit_code_signal()
         finally:
             if SHELL is not None:
                 os.environ['SHELL'] = SHELL
             else:
                 del os.environ['SHELL']
 
 
 class TestSystemRaw(ExitCodeChecks):
 
     def setUp(self):
         super().setUp()
         self.system = ip.system_raw
 
     @onlyif_unicode_paths
     def test_1(self):
         """Test system_raw with non-ascii cmd
         """
         cmd = u'''python -c "'åäö'"   '''
         ip.system_raw(cmd)
 
     @mock.patch('subprocess.call', side_effect=KeyboardInterrupt)
     @mock.patch('os.system', side_effect=KeyboardInterrupt)
     def test_control_c(self, *mocks):
         try:
             self.system("sleep 1 # wont happen")
         except KeyboardInterrupt:
             self.fail("system call should intercept "
                       "keyboard interrupt from subprocess.call")
         self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGINT)
 
 # TODO: Exit codes are currently ignored on Windows.
 class TestSystemPipedExitCode(ExitCodeChecks):
 
     def setUp(self):
         super().setUp()
         self.system = ip.system_piped
 
     @skip_win32
     def test_exit_code_ok(self):
         ExitCodeChecks.test_exit_code_ok(self)
 
     @skip_win32
     def test_exit_code_error(self):
         ExitCodeChecks.test_exit_code_error(self)
 
     @skip_win32
     def test_exit_code_signal(self):
         ExitCodeChecks.test_exit_code_signal(self)
 
 class TestModules(tt.TempFileMixin):
     def test_extraneous_loads(self):
         """Test we're not loading modules on startup that we shouldn't.
         """
         self.mktmp("import sys\n"
                    "print('numpy' in sys.modules)\n"
                    "print('ipyparallel' in sys.modules)\n"
                    "print('ipykernel' in sys.modules)\n"
                    )
         out = "False\nFalse\nFalse\n"
         tt.ipexec_validate(self.fname, out)
 
 class Negator(ast.NodeTransformer):
     """Negates all number literals in an AST."""
 
     # for python 3.7 and earlier
     def visit_Num(self, node):
         node.n = -node.n
         return node
 
     # for python 3.8+
     def visit_Constant(self, node):
         if isinstance(node.value, int):
             return self.visit_Num(node)
         return node
 
 class TestAstTransform(unittest.TestCase):
     def setUp(self):
         self.negator = Negator()
         ip.ast_transformers.append(self.negator)
     
     def tearDown(self):
         ip.ast_transformers.remove(self.negator)
     
     def test_run_cell(self):
         with tt.AssertPrints('-34'):
             ip.run_cell('print (12 + 22)')
         
         # A named reference to a number shouldn't be transformed.
         ip.user_ns['n'] = 55
         with tt.AssertNotPrints('-55'):
             ip.run_cell('print (n)')
     
     def test_timeit(self):
         called = set()
         def f(x):
             called.add(x)
         ip.push({'f':f})
         
         with tt.AssertPrints("std. dev. of"):
             ip.run_line_magic("timeit", "-n1 f(1)")
         self.assertEqual(called, {-1})
         called.clear()
 
         with tt.AssertPrints("std. dev. of"):
             ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
         self.assertEqual(called, {-2, -3})
     
     def test_time(self):
         called = []
         def f(x):
             called.append(x)
         ip.push({'f':f})
         
         # Test with an expression
         with tt.AssertPrints("Wall time: "):
             ip.run_line_magic("time", "f(5+9)")
         self.assertEqual(called, [-14])
         called[:] = []
         
         # Test with a statement (different code path)
         with tt.AssertPrints("Wall time: "):
             ip.run_line_magic("time", "a = f(-3 + -2)")
         self.assertEqual(called, [5])
     
     def test_macro(self):
         ip.push({'a':10})
         # The AST transformation makes this do a+=-1
         ip.define_macro("amacro", "a+=1\nprint(a)")
         
         with tt.AssertPrints("9"):
             ip.run_cell("amacro")
         with tt.AssertPrints("8"):
             ip.run_cell("amacro")
 
+class TestMiscTransform(unittest.TestCase):
+
+
+    def test_transform_only_once(self):
+        cleanup = 0
+        line_t = 0
+        def count_cleanup(lines):
+            nonlocal cleanup
+            cleanup += 1
+            return lines
+
+        def count_line_t(lines):
+            nonlocal line_t
+            line_t += 1
+            return lines
+
+        ip.input_transformer_manager.cleanup_transforms.append(count_cleanup)
+        ip.input_transformer_manager.line_transforms.append(count_line_t)
+
+        ip.run_cell('1')
+
+        assert cleanup == 1
+        assert line_t == 1
+
 class IntegerWrapper(ast.NodeTransformer):
     """Wraps all integers in a call to Integer()"""
 
     # for Python 3.7 and earlier
 
     # for Python 3.7 and earlier
     def visit_Num(self, node):
         if isinstance(node.n, int):
             return ast.Call(func=ast.Name(id='Integer', ctx=ast.Load()),
                             args=[node], keywords=[])
         return node
 
     # For Python 3.8+
     def visit_Constant(self, node):
         if isinstance(node.value, int):
             return self.visit_Num(node)
         return node
 
 
 class TestAstTransform2(unittest.TestCase):
     def setUp(self):
         self.intwrapper = IntegerWrapper()
         ip.ast_transformers.append(self.intwrapper)
         
         self.calls = []
         def Integer(*args):
             self.calls.append(args)
             return args
         ip.push({"Integer": Integer})
     
     def tearDown(self):
         ip.ast_transformers.remove(self.intwrapper)
         del ip.user_ns['Integer']
     
     def test_run_cell(self):
         ip.run_cell("n = 2")
         self.assertEqual(self.calls, [(2,)])
         
         # This shouldn't throw an error
         ip.run_cell("o = 2.0")
         self.assertEqual(ip.user_ns['o'], 2.0)
     
     def test_timeit(self):
         called = set()
         def f(x):
             called.add(x)
         ip.push({'f':f})
 
         with tt.AssertPrints("std. dev. of"):
             ip.run_line_magic("timeit", "-n1 f(1)")
         self.assertEqual(called, {(1,)})
         called.clear()
 
         with tt.AssertPrints("std. dev. of"):
             ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
         self.assertEqual(called, {(2,), (3,)})
 
 class ErrorTransformer(ast.NodeTransformer):
     """Throws an error when it sees a number."""
 
     # for Python 3.7 and earlier
     def visit_Num(self, node):
         raise ValueError("test")
 
     # for Python 3.8+
     def visit_Constant(self, node):
         if isinstance(node.value, int):
             return self.visit_Num(node)
         return node
 
 
 class TestAstTransformError(unittest.TestCase):
     def test_unregistering(self):
         err_transformer = ErrorTransformer()
         ip.ast_transformers.append(err_transformer)
         
         with self.assertWarnsRegex(UserWarning, "It will be unregistered"):
             ip.run_cell("1 + 2")
         
         # This should have been removed.
         nt.assert_not_in(err_transformer, ip.ast_transformers)
 
 
 class StringRejector(ast.NodeTransformer):
     """Throws an InputRejected when it sees a string literal.
 
     Used to verify that NodeTransformers can signal that a piece of code should
     not be executed by throwing an InputRejected.
     """
     
     #for python 3.7 and earlier
     def visit_Str(self, node):
         raise InputRejected("test")
 
     # 3.8 only
     def visit_Constant(self, node):
         if isinstance(node.value, str):
             raise InputRejected("test")
         return node
 
 
 class TestAstTransformInputRejection(unittest.TestCase):
 
     def setUp(self):
         self.transformer = StringRejector()
         ip.ast_transformers.append(self.transformer)
 
     def tearDown(self):
         ip.ast_transformers.remove(self.transformer)
 
     def test_input_rejection(self):
         """Check that NodeTransformers can reject input."""
 
         expect_exception_tb = tt.AssertPrints("InputRejected: test")
         expect_no_cell_output = tt.AssertNotPrints("'unsafe'", suppress=False)
 
         # Run the same check twice to verify that the transformer is not
         # disabled after raising.
         with expect_exception_tb, expect_no_cell_output:
             ip.run_cell("'unsafe'")
 
         with expect_exception_tb, expect_no_cell_output:
             res = ip.run_cell("'unsafe'")
 
         self.assertIsInstance(res.error_before_exec, InputRejected)
 
 def test__IPYTHON__():
     # This shouldn't raise a NameError, that's all
     __IPYTHON__
 
 
 class DummyRepr(object):
     def __repr__(self):
         return "DummyRepr"
     
     def _repr_html_(self):
         return "<b>dummy</b>"
     
     def _repr_javascript_(self):
         return "console.log('hi');", {'key': 'value'}
     
 
 def test_user_variables():
     # enable all formatters
     ip.display_formatter.active_types = ip.display_formatter.format_types
     
     ip.user_ns['dummy'] = d = DummyRepr()
     keys = {'dummy', 'doesnotexist'}
     r = ip.user_expressions({ key:key for key in keys})
 
     nt.assert_equal(keys, set(r.keys()))
     dummy = r['dummy']
     nt.assert_equal({'status', 'data', 'metadata'}, set(dummy.keys()))
     nt.assert_equal(dummy['status'], 'ok')
     data = dummy['data']
     metadata = dummy['metadata']
     nt.assert_equal(data.get('text/html'), d._repr_html_())
     js, jsmd = d._repr_javascript_()
     nt.assert_equal(data.get('application/javascript'), js)
     nt.assert_equal(metadata.get('application/javascript'), jsmd)
     
     dne = r['doesnotexist']
     nt.assert_equal(dne['status'], 'error')
     nt.assert_equal(dne['ename'], 'NameError')
     
     # back to text only
     ip.display_formatter.active_types = ['text/plain']
     
 def test_user_expression():
     # enable all formatters
     ip.display_formatter.active_types = ip.display_formatter.format_types
     query = {
         'a' : '1 + 2',
         'b' : '1/0',
     }
     r = ip.user_expressions(query)
     import pprint
     pprint.pprint(r)
     nt.assert_equal(set(r.keys()), set(query.keys()))
     a = r['a']
     nt.assert_equal({'status', 'data', 'metadata'}, set(a.keys()))
     nt.assert_equal(a['status'], 'ok')
     data = a['data']
     metadata = a['metadata']
     nt.assert_equal(data.get('text/plain'), '3')
     
     b = r['b']
     nt.assert_equal(b['status'], 'error')
     nt.assert_equal(b['ename'], 'ZeroDivisionError')
     
     # back to text only
     ip.display_formatter.active_types = ['text/plain']
 
 
 class TestSyntaxErrorTransformer(unittest.TestCase):
     """Check that SyntaxError raised by an input transformer is handled by run_cell()"""
 
     @staticmethod
     def transformer(lines):
         for line in lines:
             pos = line.find('syntaxerror')
             if pos >= 0:
                 e = SyntaxError('input contains "syntaxerror"')
                 e.text = line
                 e.offset = pos + 1
                 raise e
         return lines
 
     def setUp(self):
         ip.input_transformers_post.append(self.transformer)
 
     def tearDown(self):
         ip.input_transformers_post.remove(self.transformer)
 
     def test_syntaxerror_input_transformer(self):
         with tt.AssertPrints('1234'):
             ip.run_cell('1234')
         with tt.AssertPrints('SyntaxError: invalid syntax'):
             ip.run_cell('1 2 3')   # plain python syntax error
         with tt.AssertPrints('SyntaxError: input contains "syntaxerror"'):
             ip.run_cell('2345  # syntaxerror')  # input transformer syntax error
         with tt.AssertPrints('3456'):
             ip.run_cell('3456')
 
 
 class TestWarningSuppression(unittest.TestCase):
     def test_warning_suppression(self):
         ip.run_cell("import warnings")
         try:
             with self.assertWarnsRegex(UserWarning, "asdf"):
                 ip.run_cell("warnings.warn('asdf')")
             # Here's the real test -- if we run that again, we should get the
             # warning again. Traditionally, each warning was only issued once per
             # IPython session (approximately), even if the user typed in new and
             # different code that should have also triggered the warning, leading
             # to much confusion.
             with self.assertWarnsRegex(UserWarning, "asdf"):
                 ip.run_cell("warnings.warn('asdf')")
         finally:
             ip.run_cell("del warnings")
 
 
     def test_deprecation_warning(self):
         ip.run_cell("""
 import warnings
 def wrn():
     warnings.warn(
         "I AM  A WARNING",
         DeprecationWarning
     )
         """)
         try:
             with self.assertWarnsRegex(DeprecationWarning, "I AM  A WARNING"):
                 ip.run_cell("wrn()")
         finally:
             ip.run_cell("del warnings")
             ip.run_cell("del wrn")
 
 
 class TestImportNoDeprecate(tt.TempFileMixin):
 
     def setUp(self):
         """Make a valid python temp file."""
         self.mktmp("""
 import warnings
 def wrn():
     warnings.warn(
         "I AM  A WARNING",
         DeprecationWarning
     )
 """)
         super().setUp()
 
     def test_no_dep(self):
         """
         No deprecation warning should be raised from imported functions
         """
         ip.run_cell("from {} import wrn".format(self.fname))
 
         with tt.AssertNotPrints("I AM  A WARNING"):
             ip.run_cell("wrn()")
         ip.run_cell("del wrn")
 
 
 def test_custom_exc_count():
     hook = mock.Mock(return_value=None)
     ip.set_custom_exc((SyntaxError,), hook)
     before = ip.execution_count
     ip.run_cell("def foo()", store_history=True)
     # restore default excepthook
     ip.set_custom_exc((), None)
     nt.assert_equal(hook.call_count, 1)
     nt.assert_equal(ip.execution_count, before + 1)
 
 
 def test_run_cell_async():
     loop = asyncio.get_event_loop()
     ip.run_cell("import asyncio")
     coro = ip.run_cell_async("await asyncio.sleep(0.01)\n5")
     assert asyncio.iscoroutine(coro)
     result = loop.run_until_complete(coro)
     assert isinstance(result, interactiveshell.ExecutionResult)
     assert result.result == 5
 
 
 def test_should_run_async():
     assert not ip.should_run_async("a = 5")
     assert ip.should_run_async("await x")
     assert ip.should_run_async("import asyncio; await asyncio.sleep(1)")
 
 
 def test_set_custom_completer():
     num_completers = len(ip.Completer.matchers)
 
     def foo(*args, **kwargs):
         return "I'm a completer!"
 
     ip.set_custom_completer(foo, 0)
 
     # check that we've really added a new completer
     assert len(ip.Completer.matchers) == num_completers + 1
 
     # check that the first completer is the function we defined
     assert ip.Completer.matchers[0]() == "I'm a completer!"
 
     # clean up
     ip.Completer.custom_matchers.pop()