FIX: Typing annotations (#12683)

Running `mypy -p IPython.sphinxext` raised a few errors. Fix them and
strip some trailing whitespace from a few places.

IPython/core/inputsplitter.py

@@ -1,798 +1,799
 """DEPRECATED: Input handling and transformation machinery.
 
 This module was deprecated in IPython 7.0, in favour of inputtransformer2.
 
 The first class in this module, :class:`InputSplitter`, is designed to tell when
 input from a line-oriented frontend is complete and should be executed, and when
 the user should be prompted for another line of code instead. The name 'input
 splitter' is largely for historical reasons.
 
 A companion, :class:`IPythonInputSplitter`, provides the same functionality but
 with full support for the extended IPython syntax (magics, system calls, etc).
 The code to actually do these transformations is in :mod:`IPython.core.inputtransformer`.
 :class:`IPythonInputSplitter` feeds the raw code to the transformers in order
 and stores the results.
 
 For more details, see the class docstrings below.
 """
+
 from __future__ import annotations
 
 from warnings import warn
 
 warn('IPython.core.inputsplitter is deprecated since IPython 7 in favor of `IPython.core.inputtransformer2`',
      DeprecationWarning)
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 import ast
 import codeop
 import io
 import re
 import sys
 import tokenize
 import warnings
 
 from typing import List, Tuple, Union, Optional, TYPE_CHECKING
 from types import CodeType
 
 from IPython.core.inputtransformer import (leading_indent,
                                            classic_prompt,
                                            ipy_prompt,
                                            cellmagic,
                                            assemble_logical_lines,
                                            help_end,
                                            escaped_commands,
                                            assign_from_magic,
                                            assign_from_system,
                                            assemble_python_lines,
                                            )
 from IPython.utils import tokenutil
 
 # These are available in this module for backwards compatibility.
 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)
 
 if TYPE_CHECKING:
     from typing_extensions import Self
 #-----------------------------------------------------------------------------
 # 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(r'^\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
     """
     warnings.warn(
         "`num_ini_spaces` is Pending Deprecation since IPython 8.17."
         "It is considered for removal in in future version. "
         "Please open an issue if you believe it should be kept.",
         stacklevel=2,
         category=PendingDeprecationWarning,
     )
     ini_spaces = ini_spaces_re.match(s)
     if ini_spaces:
         return ini_spaces.end()
     else:
         return 0
 
 # Fake token types for partial_tokenize:
 INCOMPLETE_STRING = tokenize.N_TOKENS
 IN_MULTILINE_STATEMENT = tokenize.N_TOKENS + 1
 
 # The 2 classes below have the same API as TokenInfo, but don't try to look up
 # a token type name that they won't find.
 class IncompleteString:
     type = exact_type = INCOMPLETE_STRING
     def __init__(self, s, start, end, line):
         self.s = s
         self.start = start
         self.end = end
         self.line = line
 
 class InMultilineStatement:
     type = exact_type = IN_MULTILINE_STATEMENT
     def __init__(self, pos, line):
         self.s = ''
         self.start = self.end = pos
         self.line = line
 
 def partial_tokens(s):
     """Iterate over tokens from a possibly-incomplete string of code.
 
     This adds two special token types: INCOMPLETE_STRING and
     IN_MULTILINE_STATEMENT. These can only occur as the last token yielded, and
     represent the two main ways for code to be incomplete.
     """
     readline = io.StringIO(s).readline
     token = tokenize.TokenInfo(tokenize.NEWLINE, '', (1, 0), (1, 0), '')
     try:
         for token in tokenutil.generate_tokens_catch_errors(readline):
             yield token
     except tokenize.TokenError as e:
         # catch EOF error
         lines = s.splitlines(keepends=True)
         end = len(lines), len(lines[-1])
         if 'multi-line string' in e.args[0]:
             l, c = start = token.end
             s = lines[l-1][c:] + ''.join(lines[l:])
             yield IncompleteString(s, start, end, lines[-1])
         elif 'multi-line statement' in e.args[0]:
             yield InMultilineStatement(end, lines[-1])
         else:
             raise
 
 def find_next_indent(code) -> int:
     """Find the number of spaces for the next line of indentation"""
     tokens = list(partial_tokens(code))
     if tokens[-1].type == tokenize.ENDMARKER:
         tokens.pop()
     if not tokens:
         return 0
 
     while tokens[-1].type in {
         tokenize.DEDENT,
         tokenize.NEWLINE,
         tokenize.COMMENT,
         tokenize.ERRORTOKEN,
     }:
         tokens.pop()
 
     # Starting in Python 3.12, the tokenize module adds implicit newlines at the end
     # of input. We need to remove those if we're in a multiline statement
     if tokens[-1].type == IN_MULTILINE_STATEMENT:
         while tokens[-2].type in {tokenize.NL}:
             tokens.pop(-2)
 
 
     if tokens[-1].type == INCOMPLETE_STRING:
         # Inside a multiline string
         return 0
 
     # Find the indents used before
     prev_indents = [0]
     def _add_indent(n):
         if n != prev_indents[-1]:
             prev_indents.append(n)
 
     tokiter = iter(tokens)
     for tok in tokiter:
         if tok.type in {tokenize.INDENT, tokenize.DEDENT}:
             _add_indent(tok.end[1])
         elif (tok.type == tokenize.NL):
             try:
                 _add_indent(next(tokiter).start[1])
             except StopIteration:
                 break
 
     last_indent = prev_indents.pop()
 
     # If we've just opened a multiline statement (e.g. 'a = ['), indent more
     if tokens[-1].type == IN_MULTILINE_STATEMENT:
         if tokens[-2].exact_type in {tokenize.LPAR, tokenize.LSQB, tokenize.LBRACE}:
             return last_indent + 4
         return last_indent
 
     if tokens[-1].exact_type == tokenize.COLON:
         # Line ends with colon - indent
         return last_indent + 4
 
     if last_indent:
         # Examine the last line for dedent cues - statements like return or
         # raise which normally end a block of code.
         last_line_starts = 0
         for i, tok in enumerate(tokens):
             if tok.type == tokenize.NEWLINE:
                 last_line_starts = i + 1
 
         last_line_tokens = tokens[last_line_starts:]
         names = [t.string for t in last_line_tokens if t.type == tokenize.NAME]
         if names and names[0] in {'raise', 'return', 'pass', 'break', 'continue'}:
             # Find the most recent indentation less than the current level
             for indent in reversed(prev_indents):
                 if indent < last_indent:
                     return indent
 
     return last_indent
 
 
 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):
     r"""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())
     """
     # A cache for storing the current indentation
     # The first value stores the most recently processed source input
     # The second value is the number of spaces for the current indentation 
     # If self.source matches the first value, the second value is a valid
     # current indentation. Otherwise, the cache is invalid and the indentation
     # must be recalculated.
     _indent_spaces_cache: Union[Tuple[None, None], Tuple[str, int]] = None, None
     # 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: str = ""
     # 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: Optional[CodeType] = None
 
     # Private attributes
 
     # List with lines of input accumulated so far
     _buffer: List[str]
     # Command compiler
     _compile: codeop.CommandCompiler
     # Boolean indicating whether the current block is complete
     _is_complete: Optional[bool] = None
     # Boolean indicating whether the current block has an unrecoverable syntax error
     _is_invalid: bool = False
 
     def __init__(self) -> None:
         """Create a new InputSplitter instance."""
         self._buffer = []
         self._compile = codeop.CommandCompiler()
         self.encoding = get_input_encoding()
 
     def reset(self):
         """Reset the input buffer and associated state."""
         self._buffer[:] = []
         self.source = ''
         self.code = None
         self._is_complete = False
         self._is_invalid = False
 
     def source_reset(self):
         """Return the input source and perform a full reset.
         """
         out = self.source
         self.reset()
         return out
 
     def check_complete(self, source):
         """Return whether a block of code is ready to execute, or should be continued
 
         This is a non-stateful API, and will reset the state of this InputSplitter.
 
         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.
         """
         self.reset()
         try:
             self.push(source)
         except SyntaxError:
             # Transformers in IPythonInputSplitter can raise SyntaxError,
             # which push() will not catch.
             return 'invalid', None
         else:
             if self._is_invalid:
                 return 'invalid', None
             elif self.push_accepts_more():
                 return 'incomplete', self.get_indent_spaces()
             else:
                 return 'complete', None
         finally:
             self.reset()
 
     def push(self, lines:str) -> bool:
         """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.
         """
         assert isinstance(lines, str)
         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
         self._is_invalid = False
 
         # Honor termination lines properly
         if source.endswith('\\\n'):
             return False
 
         try:
             with warnings.catch_warnings():
                 warnings.simplefilter('error', SyntaxWarning)
                 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, SyntaxWarning):
             self._is_complete = True
             self._is_invalid = 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 when either:
 
         * A SyntaxError is raised
 
         * The code is complete and consists of a single line or a single
           non-compound statement
 
         * The code is complete and has a blank line at the end
 
         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
         # A syntax error also sets _is_complete to True - see push()
         if not self._is_complete:
             #print("Not complete")  # debug
             return True
         
         # The user can make any (complete) input execute by leaving a blank line
         last_line = self.source.splitlines()[-1]
         if (not last_line) or last_line.isspace():
             #print("Blank line")  # debug
             return False
         
         # If there's just a single line or AST node, and we're flush left, as is
         # the case after a simple statement such as 'a=1', we want to execute it
         # straight away.
         if self.get_indent_spaces() == 0:
             if len(self.source.splitlines()) <= 1:
                 return False
             
             try:
                 code_ast = ast.parse("".join(self._buffer))
             except Exception:
                 #print("Can't parse AST")  # debug
                 return False
             else:
                 if len(code_ast.body) == 1 and \
                                     not hasattr(code_ast.body[0], 'body'):
                     #print("Simple statement")  # debug
                     return False
 
         # General fallback - accept more code
         return True
 
     def get_indent_spaces(self) -> int:
         sourcefor, n = self._indent_spaces_cache
         if sourcefor == self.source:
             assert n is not None
             return n
 
         # self.source always has a trailing newline
         n = find_next_indent(self.source[:-1])
         self._indent_spaces_cache = (self.source, n)
         return n
 
     # Backwards compatibility. I think all code that used .indent_spaces was
     # inside IPython, but we can leave this here until IPython 7 in case any
     # other modules are using it. -TK, November 2017
     indent_spaces = property(get_indent_spaces)
 
     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: List[str]
 
     def __init__(self, line_input_checker=True, physical_line_transforms=None,
                     logical_line_transforms=None, python_line_transforms=None):
         super(IPythonInputSplitter, self).__init__()
         self._buffer_raw = []
         self._validate = True
         
         if physical_line_transforms is not None:
             self.physical_line_transforms = physical_line_transforms
         else:
             self.physical_line_transforms = [
                                              leading_indent(),
                                              classic_prompt(),
                                              ipy_prompt(),
                                              cellmagic(end_on_blank_line=line_input_checker),
                                             ]
         
         self.assemble_logical_lines = assemble_logical_lines()
         if logical_line_transforms is not None:
             self.logical_line_transforms = logical_line_transforms
         else:
             self.logical_line_transforms = [
                                             help_end(),
                                             escaped_commands(),
                                             assign_from_magic(),
                                             assign_from_system(),
                                            ]
         
         self.assemble_python_lines = assemble_python_lines()
         if python_line_transforms is not None:
             self.python_line_transforms = python_line_transforms
         else:
             # We don't use any of these at present
             self.python_line_transforms = []
     
     @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[:]
         if not self.within_python_line:
             t += [self.assemble_logical_lines] + 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
         self.within_python_line = False
 
         for t in self.transforms:
             try:
                 t.reset()
             except SyntaxError:
                 # Nothing that calls reset() expects to handle transformer
                 # errors
                 pass
 
     def flush_transformers(self: Self):
         def _flush(transform, outs: List[str]):
             """yield transformed lines
 
             always strings, never None
 
             transform: the current transform
             outs: an iterable of previously transformed inputs.
                  Each may be multiline, which will be passed
                  one line at a time to transform.
             """
             for out in outs:
                 for line in out.splitlines():
                     # push one line at a time
                     tmp = transform.push(line)
                     if tmp is not None:
                         yield tmp
             
             # reset the transform
             tmp = transform.reset()
             if tmp is not None:
                 yield tmp
 
         out: List[str] = []
         for t in self.transforms_in_use:
             out = _flush(t, out)
         
         out = list(out)
         if out:
             self._store('\n'.join(out))
 
     def raw_reset(self):
         """Return raw input only and perform a full reset.
         """
         out = self.source_raw
         self.reset()
         return out
     
     def source_reset(self):
         try:
             self.flush_transformers()
             return self.source
         finally:
             self.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()
         try:
             self.push(cell)
             self.flush_transformers()
             return self.source
         finally:
             self.reset()
 
     def push(self, lines:str) -> bool:
         """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.
         """
         assert isinstance(lines, str)
         # We must ensure all input is pure unicode
         # ''.splitlines() --> [], but we need to push the empty line to transformers
         lines_list = lines.splitlines()
         if not lines_list:
             lines_list = ['']
 
         # 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')
 
         transformed_lines_list = []
         for line in lines_list:
             transformed = self._transform_line(line)
             if transformed is not None:
                 transformed_lines_list.append(transformed)
 
         if transformed_lines_list:
             transformed_lines = '\n'.join(transformed_lines_list)
             return super(IPythonInputSplitter, self).push(transformed_lines)
         else:
             # Got nothing back from transformers - they must be waiting for
             # more input.
             return False
 
     def _transform_line(self, line):
         """Push a line of input code through the various transformers.
 
         Returns any output from the transformers, or None if a transformer
         is accumulating lines.
 
         Sets self.transformer_accumulating as a side effect.
         """
         def _accumulating(dbg):
             #print(dbg)
             self.transformer_accumulating = True
             return None
 
         for transformer in self.physical_line_transforms:
             line = transformer.push(line)
             if line is None:
                 return _accumulating(transformer)
 
         if not self.within_python_line:
             line = self.assemble_logical_lines.push(line)
             if line is None:
                 return _accumulating('acc logical 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:
                 return _accumulating(transformer)
 
         #print("transformers clear") #debug
         self.transformer_accumulating = False
         return line
 

IPython/core/tests/test_inputtransformer2.py

@@ -1,447 +1,448
 """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 platform
 import string
 import sys
 from textwrap import dedent
 
 import pytest
 
 from IPython.core import inputtransformer2 as ipt2
 from IPython.core.inputtransformer2 import _find_assign_op, make_tokens_by_line
 
 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
     ),
 )
 
 CRLF_MAGIC = (
     ["a = f()\n", "%ls\r\n", "g()\n"],
     (2, 0),
     ["a = f()\n", "get_ipython().run_line_magic('ls', '')\n", "g()\n"],
 )
 
 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().run_line_magic('pinfo', 'c')\n"],
 )
 
 HELP_CONTINUED_LINE = (
     """\
 a = \\
 zip?
 """.splitlines(
         keepends=True
     ),
     (1, 0),
     [r"get_ipython().run_line_magic('pinfo', 'zip')" + "\n"],
 )
 
 HELP_MULTILINE = (
     """\
 (a,
 b) = zip?
 """.splitlines(
         keepends=True
     ),
     (1, 0),
     [r"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 test_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:
         assert make_tokens_by_line(c)[-1] != []
         for k in printable:
             assert 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
         assert (res.start_line + 1, res.start_col) == expected_start
         return res
     else:
         assert res is None
 
 
 def check_transform(transformer_cls, case):
     lines, start, expected = case
     transformer = transformer_cls(start)
     assert transformer.transform(lines) == expected
 
 
 def test_continued_line():
     lines = MULTILINE_MAGIC_ASSIGN[0]
     assert ipt2.find_end_of_continued_line(lines, 1) == 2
 
     assert 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)
     check_transform(ipt2.EscapedCommand, CRLF_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)
     assert tf.q_line == 1
     assert tf.q_col == 3
 
     tf = check_find(ipt2.HelpEnd, HELP_MULTILINE)
     assert tf.q_line == 1
     assert 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))
     assert tf.transform(HELP_IN_EXPR[0]) == HELP_IN_EXPR[2]
 
     tf = ipt2.HelpEnd((1, 0), (2, 3))
     assert tf.transform(HELP_CONTINUED_LINE[0]) == HELP_CONTINUED_LINE[2]
 
     tf = ipt2.HelpEnd((1, 0), (2, 8))
     assert tf.transform(HELP_MULTILINE[0]) == HELP_MULTILINE[2]
 
     tf = ipt2.HelpEnd((1, 0), (1, 0))
     assert 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
 
     assert _find_assign_op([Tk(s) for s in ("", "a", "=", "b")]) == 2
     assert (
         _find_assign_op([Tk(s) for s in ("", "(", "a", "=", "b", ")", "=", "5")]) == 6
     )
 
 
 extra_closing_paren_param = (
     pytest.param("(\n))", "invalid", None)
     if sys.version_info >= (3, 12)
     else pytest.param("(\n))", "incomplete", 0)
 )
 examples = [
     pytest.param("a = 1", "complete", None),
     pytest.param("for a in range(5):", "incomplete", 4),
     pytest.param("for a in range(5):\n    if a > 0:", "incomplete", 8),
     pytest.param("raise = 2", "invalid", None),
     pytest.param("a = [1,\n2,", "incomplete", 0),
     extra_closing_paren_param,
     pytest.param("\\\r\n", "incomplete", 0),
     pytest.param("a = '''\n   hi", "incomplete", 3),
     pytest.param("def a():\n x=1\n global x", "invalid", None),
     pytest.param(
         "a \\ ",
         "invalid",
         None,
         marks=pytest.mark.xfail(
             reason="Bug in python 3.9.8 – bpo 45738",
             condition=sys.version_info in [(3, 11, 0, "alpha", 2)],
             raises=SystemError,
             strict=True,
         ),
     ),  # Nothing allowed after backslash,
     pytest.param("1\\\n+2", "complete", None),
 ]
 
 
 @pytest.mark.parametrize("code, expected, number", examples)
 def test_check_complete_param(code, expected, number):
     cc = ipt2.TransformerManager().check_complete
     assert cc(code) == (expected, number)
 
 
 @pytest.mark.xfail(platform.python_implementation() == "PyPy", reason="fail on pypy")
 @pytest.mark.xfail(
     reason="Bug in python 3.9.8 – bpo 45738",
     condition=sys.version_info in [(3, 11, 0, "alpha", 2)],
     raises=SystemError,
     strict=True,
 )
 def test_check_complete():
     cc = ipt2.TransformerManager().check_complete
 
     example = dedent(
         """
         if True:
             a=1"""
     )
 
     assert cc(example) == ("incomplete", 4)
     assert cc(example + "\n") == ("complete", None)
     assert 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)
 
     assert cc("def f():\n  x=0\n  \\\n  ") == ("incomplete", 2)
 
 
 @pytest.mark.xfail(platform.python_implementation() == "PyPy", reason="fail on pypy")
 @pytest.mark.parametrize(
     "value, expected",
     [
         ('''def foo():\n    """''', ("incomplete", 4)),
         ("""async with example:\n    pass""", ("incomplete", 4)),
         ("""async with example:\n    pass\n    """, ("complete", None)),
     ],
 )
 def test_check_complete_II(value, expected):
     """
     Test that multiple line strings are properly handled.
 
     Separate test function for convenience
 
     """
     cc = ipt2.TransformerManager().check_complete
     assert cc(value) == expected
 
 
 @pytest.mark.parametrize(
     "value, expected",
     [
         (")", ("invalid", None)),
         ("]", ("invalid", None)),
         ("}", ("invalid", None)),
         (")(", ("invalid", None)),
         ("][", ("invalid", None)),
         ("}{", ("invalid", None)),
         ("]()(", ("invalid", None)),
         ("())(", ("invalid", None)),
         (")[](", ("invalid", None)),
         ("()](", ("invalid", None)),
     ],
 )
 def test_check_complete_invalidates_sunken_brackets(value, expected):
     """
     Test that a single line with more closing brackets than the opening ones is
     interpreted as invalid
     """
     cc = ipt2.TransformerManager().check_complete
     assert cc(value) == expected
 
 
 def test_null_cleanup_transformer():
     manager = ipt2.TransformerManager()
     manager.cleanup_transforms.insert(0, null_cleanup_transformer)
     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_paths.py

@@ -1,200 +1,202
 import errno
 import os
 import shutil
 import tempfile
 import warnings
 from unittest.mock import patch
 
 from tempfile import TemporaryDirectory
 from testpath import assert_isdir, assert_isfile, modified_env
 
 from IPython import paths
 from IPython.testing.decorators import skip_win32
 
 TMP_TEST_DIR = os.path.realpath(tempfile.mkdtemp())
 HOME_TEST_DIR = os.path.join(TMP_TEST_DIR, "home_test_dir")
 XDG_TEST_DIR = os.path.join(HOME_TEST_DIR, "xdg_test_dir")
 XDG_CACHE_DIR = os.path.join(HOME_TEST_DIR, "xdg_cache_dir")
 IP_TEST_DIR = os.path.join(HOME_TEST_DIR,'.ipython')
 
 def setup_module():
     """Setup testenvironment for the module:
 
             - Adds dummy home dir tree
     """
     # Do not mask exceptions here.  In particular, catching WindowsError is a
     # problem because that exception is only defined on Windows...
     os.makedirs(IP_TEST_DIR)
     os.makedirs(os.path.join(XDG_TEST_DIR, 'ipython'))
     os.makedirs(os.path.join(XDG_CACHE_DIR, 'ipython'))
 
 
 def teardown_module():
     """Teardown testenvironment for the module:
 
             - Remove dummy home dir tree
     """
     # Note: we remove the parent test dir, which is the root of all test
     # subdirs we may have created.  Use shutil instead of os.removedirs, so
     # that non-empty directories are all recursively removed.
     shutil.rmtree(TMP_TEST_DIR)
 
 def patch_get_home_dir(dirpath):
     return patch.object(paths, 'get_home_dir', return_value=dirpath)
 
 
 def test_get_ipython_dir_1():
     """test_get_ipython_dir_1, Testcase to see if we can call get_ipython_dir without Exceptions."""
     env_ipdir = os.path.join("someplace", ".ipython")
     with patch.object(paths, '_writable_dir', return_value=True), \
             modified_env({'IPYTHONDIR': env_ipdir}):
         ipdir = paths.get_ipython_dir()
 
     assert ipdir == env_ipdir
 
 def test_get_ipython_dir_2():
     """test_get_ipython_dir_2, Testcase to see if we can call get_ipython_dir without Exceptions."""
     with patch_get_home_dir('someplace'), \
             patch.object(paths, 'get_xdg_dir', return_value=None), \
             patch.object(paths, '_writable_dir', return_value=True), \
             patch('os.name', "posix"), \
             modified_env({'IPYTHON_DIR': None,
                           'IPYTHONDIR': None,
                           'XDG_CONFIG_HOME': None
                          }):
         ipdir = paths.get_ipython_dir()
 
     assert ipdir == os.path.join("someplace", ".ipython")
 
 def test_get_ipython_dir_3():
     """test_get_ipython_dir_3, use XDG if defined and exists, and .ipython doesn't exist."""
     tmphome = TemporaryDirectory()
     try:
         with patch_get_home_dir(tmphome.name), \
                 patch('os.name', 'posix'), \
                 modified_env({
                     'IPYTHON_DIR': None,
                     'IPYTHONDIR': None,
                     'XDG_CONFIG_HOME': XDG_TEST_DIR,
                 }), warnings.catch_warnings(record=True) as w:
             ipdir = paths.get_ipython_dir()
 
         assert ipdir == os.path.join(tmphome.name, XDG_TEST_DIR, "ipython")
         assert len(w) == 0
     finally:
         tmphome.cleanup()
 
 def test_get_ipython_dir_4():
     """test_get_ipython_dir_4, warn if XDG and home both exist."""
     with patch_get_home_dir(HOME_TEST_DIR), \
             patch('os.name', 'posix'):
         try:
             os.mkdir(os.path.join(XDG_TEST_DIR, 'ipython'))
         except OSError as e:
             if e.errno != errno.EEXIST:
                 raise
 
 
         with modified_env({
             'IPYTHON_DIR': None,
             'IPYTHONDIR': None,
             'XDG_CONFIG_HOME': XDG_TEST_DIR,
         }), warnings.catch_warnings(record=True) as w:
             ipdir = paths.get_ipython_dir()
 
         assert len(w) == 1
         assert "Ignoring" in str(w[0])
 
 
 def test_get_ipython_dir_5():
     """test_get_ipython_dir_5, use .ipython if exists and XDG defined, but doesn't exist."""
     with patch_get_home_dir(HOME_TEST_DIR), \
             patch('os.name', 'posix'):
         try:
             os.rmdir(os.path.join(XDG_TEST_DIR, 'ipython'))
         except OSError as e:
             if e.errno != errno.ENOENT:
                 raise
 
         with modified_env({
             'IPYTHON_DIR': None,
             'IPYTHONDIR': None,
             'XDG_CONFIG_HOME': XDG_TEST_DIR,
         }):
             ipdir = paths.get_ipython_dir()
 
         assert ipdir == IP_TEST_DIR
 
 def test_get_ipython_dir_6():
     """test_get_ipython_dir_6, use home over XDG if defined and neither exist."""
     xdg = os.path.join(HOME_TEST_DIR, 'somexdg')
     os.mkdir(xdg)
     shutil.rmtree(os.path.join(HOME_TEST_DIR, '.ipython'))
     print(paths._writable_dir)
     with patch_get_home_dir(HOME_TEST_DIR), \
             patch.object(paths, 'get_xdg_dir', return_value=xdg), \
             patch('os.name', 'posix'), \
             modified_env({
                 'IPYTHON_DIR': None,
                 'IPYTHONDIR': None,
                 'XDG_CONFIG_HOME': None,
             }), warnings.catch_warnings(record=True) as w:
         ipdir = paths.get_ipython_dir()
 
     assert ipdir == os.path.join(HOME_TEST_DIR, ".ipython")
     assert len(w) == 0
 
 def test_get_ipython_dir_7():
     """test_get_ipython_dir_7, test home directory expansion on IPYTHONDIR"""
     home_dir = os.path.normpath(os.path.expanduser('~'))
     with modified_env({'IPYTHONDIR': os.path.join('~', 'somewhere')}), \
             patch.object(paths, '_writable_dir', return_value=True):
         ipdir = paths.get_ipython_dir()
     assert ipdir == os.path.join(home_dir, "somewhere")
 
 
 @skip_win32
 def test_get_ipython_dir_8():
     """test_get_ipython_dir_8, test / home directory"""
     if not os.access("/", os.W_OK):
         # test only when HOME directory actually writable
         return
 
-    with patch.object(paths, "_writable_dir", lambda path: bool(path)), patch.object(
-        paths, "get_xdg_dir", return_value=None
-    ), modified_env(
-        {
-            "IPYTHON_DIR": None,
-            "IPYTHONDIR": None,
-            "HOME": "/",
-        }
+    with (
+        patch.object(paths, "_writable_dir", lambda path: bool(path)),
+        patch.object(paths, "get_xdg_dir", return_value=None),
+        modified_env(
+            {
+                "IPYTHON_DIR": None,
+                "IPYTHONDIR": None,
+                "HOME": "/",
+            }
+        ),
     ):
         assert paths.get_ipython_dir() == "/.ipython"
 
 
 def test_get_ipython_cache_dir():
     with modified_env({'HOME': HOME_TEST_DIR}):
         if os.name == "posix":
             # test default
             os.makedirs(os.path.join(HOME_TEST_DIR, ".cache"))
             with modified_env({'XDG_CACHE_HOME': None}):
                 ipdir = paths.get_ipython_cache_dir()
             assert os.path.join(HOME_TEST_DIR, ".cache", "ipython") == ipdir
             assert_isdir(ipdir)
 
             # test env override
             with modified_env({"XDG_CACHE_HOME": XDG_CACHE_DIR}):
                 ipdir = paths.get_ipython_cache_dir()
             assert_isdir(ipdir)
             assert ipdir == os.path.join(XDG_CACHE_DIR, "ipython")
         else:
             assert paths.get_ipython_cache_dir() == paths.get_ipython_dir()
 
 def test_get_ipython_package_dir():
     ipdir = paths.get_ipython_package_dir()
     assert_isdir(ipdir)
 
 
 def test_get_ipython_module_path():
     ipapp_path = paths.get_ipython_module_path('IPython.terminal.ipapp')
     assert_isfile(ipapp_path)

IPython/external/qt_loaders.py

@@ -1,422 +1,423
 """
 This module contains factory functions that attempt
 to return Qt submodules from the various python Qt bindings.
 
 It also protects against double-importing Qt with different
 bindings, which is unstable and likely to crash
 
 This is used primarily by qt and qt_for_kernel, and shouldn't
 be accessed directly from the outside
 """
+
 import importlib.abc
 import sys
 import os
 import types
 from functools import partial, lru_cache
 import operator
 
 # ### Available APIs.
 # Qt6
 QT_API_PYQT6 = "pyqt6"
 QT_API_PYSIDE6 = "pyside6"
 
 # Qt5
 QT_API_PYQT5 = 'pyqt5'
 QT_API_PYSIDE2 = 'pyside2'
 
 # Qt4
 # NOTE: Here for legacy matplotlib compatibility, but not really supported on the IPython side.
 QT_API_PYQT = "pyqt"  # Force version 2
 QT_API_PYQTv1 = "pyqtv1"  # Force version 2
 QT_API_PYSIDE = "pyside"
 
 QT_API_PYQT_DEFAULT = "pyqtdefault"  # use system default for version 1 vs. 2
 
 api_to_module = {
     # Qt6
     QT_API_PYQT6: "PyQt6",
     QT_API_PYSIDE6: "PySide6",
     # Qt5
     QT_API_PYQT5: "PyQt5",
     QT_API_PYSIDE2: "PySide2",
     # Qt4
     QT_API_PYSIDE: "PySide",
     QT_API_PYQT: "PyQt4",
     QT_API_PYQTv1: "PyQt4",
     # default
     QT_API_PYQT_DEFAULT: "PyQt6",
 }
 
 
 class ImportDenier(importlib.abc.MetaPathFinder):
     """Import Hook that will guard against bad Qt imports
     once IPython commits to a specific binding
     """
 
     def __init__(self):
         self.__forbidden = set()
 
     def forbid(self, module_name):
         sys.modules.pop(module_name, None)
         self.__forbidden.add(module_name)
 
     def find_spec(self, fullname, path, target=None):
         if path:
             return
         if fullname in self.__forbidden:
             raise ImportError(
                 """
     Importing %s disabled by IPython, which has
     already imported an Incompatible QT Binding: %s
     """
                 % (fullname, loaded_api())
             )
 
 
 ID = ImportDenier()
 sys.meta_path.insert(0, ID)
 
 
 def commit_api(api):
     """Commit to a particular API, and trigger ImportErrors on subsequent
     dangerous imports"""
     modules = set(api_to_module.values())
 
     modules.remove(api_to_module[api])
     for mod in modules:
         ID.forbid(mod)
 
 
 def loaded_api():
     """Return which API is loaded, if any
 
     If this returns anything besides None,
     importing any other Qt binding is unsafe.
 
     Returns
     -------
     None, 'pyside6', 'pyqt6', 'pyside2', 'pyside', 'pyqt', 'pyqt5', 'pyqtv1'
     """
     if sys.modules.get("PyQt6.QtCore"):
         return QT_API_PYQT6
     elif sys.modules.get("PySide6.QtCore"):
         return QT_API_PYSIDE6
     elif sys.modules.get("PyQt5.QtCore"):
         return QT_API_PYQT5
     elif sys.modules.get("PySide2.QtCore"):
         return QT_API_PYSIDE2
     elif sys.modules.get("PyQt4.QtCore"):
         if qtapi_version() == 2:
             return QT_API_PYQT
         else:
             return QT_API_PYQTv1
     elif sys.modules.get("PySide.QtCore"):
         return QT_API_PYSIDE
 
     return None
 
 
 def has_binding(api):
     """Safely check for PyQt4/5, PySide or PySide2, without importing submodules
 
     Parameters
     ----------
     api : str [ 'pyqtv1' | 'pyqt' | 'pyqt5' | 'pyside' | 'pyside2' | 'pyqtdefault']
         Which module to check for
 
     Returns
     -------
     True if the relevant module appears to be importable
     """
     module_name = api_to_module[api]
     from importlib.util import find_spec
 
     required = ['QtCore', 'QtGui', 'QtSvg']
     if api in (QT_API_PYQT5, QT_API_PYSIDE2, QT_API_PYQT6, QT_API_PYSIDE6):
         # QT5 requires QtWidgets too
         required.append('QtWidgets')
 
     for submod in required:
         try:
             spec = find_spec('%s.%s' % (module_name, submod))
         except ImportError:
             # Package (e.g. PyQt5) not found
             return False
         else:
             if spec is None:
                 # Submodule (e.g. PyQt5.QtCore) not found
                 return False
 
     if api == QT_API_PYSIDE:
         # We can also safely check PySide version
         import PySide
 
         return PySide.__version_info__ >= (1, 0, 3)
 
     return True
 
 
 def qtapi_version():
     """Return which QString API has been set, if any
 
     Returns
     -------
     The QString API version (1 or 2), or None if not set
     """
     try:
         import sip
     except ImportError:
         # as of PyQt5 5.11, sip is no longer available as a top-level
         # module and needs to be imported from the PyQt5 namespace
         try:
             from PyQt5 import sip
         except ImportError:
             return
     try:
         return sip.getapi('QString')
     except ValueError:
         return
 
 
 def can_import(api):
     """Safely query whether an API is importable, without importing it"""
     if not has_binding(api):
         return False
 
     current = loaded_api()
     if api == QT_API_PYQT_DEFAULT:
         return current in [QT_API_PYQT6, None]
     else:
         return current in [api, None]
 
 
 def import_pyqt4(version=2):
     """
     Import PyQt4
 
     Parameters
     ----------
     version : 1, 2, or None
         Which QString/QVariant API to use. Set to None to use the system
         default
     ImportErrors raised within this function are non-recoverable
     """
     # The new-style string API (version=2) automatically
     # converts QStrings to Unicode Python strings. Also, automatically unpacks
     # QVariants to their underlying objects.
     import sip
 
     if version is not None:
         sip.setapi('QString', version)
         sip.setapi('QVariant', version)
 
     from PyQt4 import QtGui, QtCore, QtSvg
 
     if QtCore.PYQT_VERSION < 0x040700:
         raise ImportError("IPython requires PyQt4 >= 4.7, found %s" %
                           QtCore.PYQT_VERSION_STR)
 
     # Alias PyQt-specific functions for PySide compatibility.
     QtCore.Signal = QtCore.pyqtSignal
     QtCore.Slot = QtCore.pyqtSlot
 
     # query for the API version (in case version == None)
     version = sip.getapi('QString')
     api = QT_API_PYQTv1 if version == 1 else QT_API_PYQT
     return QtCore, QtGui, QtSvg, api
 
 
 def import_pyqt5():
     """
     Import PyQt5
 
     ImportErrors raised within this function are non-recoverable
     """
 
     from PyQt5 import QtCore, QtSvg, QtWidgets, QtGui
 
     # Alias PyQt-specific functions for PySide compatibility.
     QtCore.Signal = QtCore.pyqtSignal
     QtCore.Slot = QtCore.pyqtSlot
 
     # Join QtGui and QtWidgets for Qt4 compatibility.
     QtGuiCompat = types.ModuleType('QtGuiCompat')
     QtGuiCompat.__dict__.update(QtGui.__dict__)
     QtGuiCompat.__dict__.update(QtWidgets.__dict__)
 
     api = QT_API_PYQT5
     return QtCore, QtGuiCompat, QtSvg, api
 
 
 def import_pyqt6():
     """
     Import PyQt6
 
     ImportErrors raised within this function are non-recoverable
     """
 
     from PyQt6 import QtCore, QtSvg, QtWidgets, QtGui
 
     # Alias PyQt-specific functions for PySide compatibility.
     QtCore.Signal = QtCore.pyqtSignal
     QtCore.Slot = QtCore.pyqtSlot
 
     # Join QtGui and QtWidgets for Qt4 compatibility.
     QtGuiCompat = types.ModuleType("QtGuiCompat")
     QtGuiCompat.__dict__.update(QtGui.__dict__)
     QtGuiCompat.__dict__.update(QtWidgets.__dict__)
 
     api = QT_API_PYQT6
     return QtCore, QtGuiCompat, QtSvg, api
 
 
 def import_pyside():
     """
     Import PySide
 
     ImportErrors raised within this function are non-recoverable
     """
     from PySide import QtGui, QtCore, QtSvg
     return QtCore, QtGui, QtSvg, QT_API_PYSIDE
 
 def import_pyside2():
     """
     Import PySide2
 
     ImportErrors raised within this function are non-recoverable
     """
     from PySide2 import QtGui, QtCore, QtSvg, QtWidgets, QtPrintSupport
 
     # Join QtGui and QtWidgets for Qt4 compatibility.
     QtGuiCompat = types.ModuleType('QtGuiCompat')
     QtGuiCompat.__dict__.update(QtGui.__dict__)
     QtGuiCompat.__dict__.update(QtWidgets.__dict__)
     QtGuiCompat.__dict__.update(QtPrintSupport.__dict__)
 
     return QtCore, QtGuiCompat, QtSvg, QT_API_PYSIDE2
 
 
 def import_pyside6():
     """
     Import PySide6
 
     ImportErrors raised within this function are non-recoverable
     """
 
     def get_attrs(module):
         return {
             name: getattr(module, name)
             for name in dir(module)
             if not name.startswith("_")
         }
 
     from PySide6 import QtGui, QtCore, QtSvg, QtWidgets, QtPrintSupport
 
     # Join QtGui and QtWidgets for Qt4 compatibility.
     QtGuiCompat = types.ModuleType("QtGuiCompat")
     QtGuiCompat.__dict__.update(QtGui.__dict__)
     if QtCore.__version_info__ < (6, 7):
         QtGuiCompat.__dict__.update(QtWidgets.__dict__)
         QtGuiCompat.__dict__.update(QtPrintSupport.__dict__)
     else:
         QtGuiCompat.__dict__.update(get_attrs(QtWidgets))
         QtGuiCompat.__dict__.update(get_attrs(QtPrintSupport))
 
     return QtCore, QtGuiCompat, QtSvg, QT_API_PYSIDE6
 
 
 def load_qt(api_options):
     """
     Attempt to import Qt, given a preference list
     of permissible bindings
 
     It is safe to call this function multiple times.
 
     Parameters
     ----------
     api_options : List of strings
         The order of APIs to try. Valid items are 'pyside', 'pyside2',
         'pyqt', 'pyqt5', 'pyqtv1' and 'pyqtdefault'
 
     Returns
     -------
     A tuple of QtCore, QtGui, QtSvg, QT_API
     The first three are the Qt modules. The last is the
     string indicating which module was loaded.
 
     Raises
     ------
     ImportError, if it isn't possible to import any requested
     bindings (either because they aren't installed, or because
     an incompatible library has already been installed)
     """
     loaders = {
         # Qt6
         QT_API_PYQT6: import_pyqt6,
         QT_API_PYSIDE6: import_pyside6,
         # Qt5
         QT_API_PYQT5: import_pyqt5,
         QT_API_PYSIDE2: import_pyside2,
         # Qt4
         QT_API_PYSIDE: import_pyside,
         QT_API_PYQT: import_pyqt4,
         QT_API_PYQTv1: partial(import_pyqt4, version=1),
         # default
         QT_API_PYQT_DEFAULT: import_pyqt6,
     }
 
     for api in api_options:
 
         if api not in loaders:
             raise RuntimeError(
                 "Invalid Qt API %r, valid values are: %s" %
                 (api, ", ".join(["%r" % k for k in loaders.keys()])))
 
         if not can_import(api):
             continue
 
         #cannot safely recover from an ImportError during this
         result = loaders[api]()
         api = result[-1]  # changed if api = QT_API_PYQT_DEFAULT
         commit_api(api)
         return result
     else:
         # Clear the environment variable since it doesn't work.
         if "QT_API" in os.environ:
             del os.environ["QT_API"]
 
         raise ImportError(
             """
     Could not load requested Qt binding. Please ensure that
     PyQt4 >= 4.7, PyQt5, PyQt6, PySide >= 1.0.3, PySide2, or
     PySide6 is available, and only one is imported per session.
 
     Currently-imported Qt library:                              %r
     PyQt5 available (requires QtCore, QtGui, QtSvg, QtWidgets): %s
     PyQt6 available (requires QtCore, QtGui, QtSvg, QtWidgets): %s
     PySide2 installed:                                          %s
     PySide6 installed:                                          %s
     Tried to load:                                              %r
     """
             % (
                 loaded_api(),
                 has_binding(QT_API_PYQT5),
                 has_binding(QT_API_PYQT6),
                 has_binding(QT_API_PYSIDE2),
                 has_binding(QT_API_PYSIDE6),
                 api_options,
             )
         )
 
 
 def enum_factory(QT_API, QtCore):
     """Construct an enum helper to account for PyQt5 <-> PyQt6 changes."""
 
     @lru_cache(None)
     def _enum(name):
         # foo.bar.Enum.Entry (PyQt6) <=> foo.bar.Entry (non-PyQt6).
         return operator.attrgetter(
             name if QT_API == QT_API_PYQT6 else name.rpartition(".")[0]
         )(sys.modules[QtCore.__package__])
 
     return _enum

IPython/lib/clipboard.py

@@ -1,101 +1,102
 """ Utilities for accessing the platform's clipboard.
 """
+
 import os
 import subprocess
 
 from IPython.core.error import TryNext
 import IPython.utils.py3compat as py3compat
 
 
 class ClipboardEmpty(ValueError):
     pass
 
 
 def win32_clipboard_get():
     """ Get the current clipboard's text on Windows.
 
     Requires Mark Hammond's pywin32 extensions.
     """
     try:
         import win32clipboard
     except ImportError as e:
         raise TryNext("Getting text from the clipboard requires the pywin32 "
                       "extensions: http://sourceforge.net/projects/pywin32/") from e
     win32clipboard.OpenClipboard()
     try:
         text = win32clipboard.GetClipboardData(win32clipboard.CF_UNICODETEXT)
     except (TypeError, win32clipboard.error):
         try:
             text = win32clipboard.GetClipboardData(win32clipboard.CF_TEXT)
             text = py3compat.cast_unicode(text, py3compat.DEFAULT_ENCODING)
         except (TypeError, win32clipboard.error) as e:
             raise ClipboardEmpty from e
     finally:
         win32clipboard.CloseClipboard()
     return text
 
 
 def osx_clipboard_get() -> str:
     """ Get the clipboard's text on OS X.
     """
     p = subprocess.Popen(['pbpaste', '-Prefer', 'ascii'],
         stdout=subprocess.PIPE)
     bytes_, stderr = p.communicate()
     # Text comes in with old Mac \r line endings. Change them to \n.
     bytes_ = bytes_.replace(b'\r', b'\n')
     text = py3compat.decode(bytes_)
     return text
 
 
 def tkinter_clipboard_get():
     """ Get the clipboard's text using Tkinter.
 
     This is the default on systems that are not Windows or OS X. It may
     interfere with other UI toolkits and should be replaced with an
     implementation that uses that toolkit.
     """
     try:
         from tkinter import Tk, TclError 
     except ImportError as e:
         raise TryNext("Getting text from the clipboard on this platform requires tkinter.") from e
         
     root = Tk()
     root.withdraw()
     try:
         text = root.clipboard_get()
     except TclError as e:
         raise ClipboardEmpty from e
     finally:
         root.destroy()
     text = py3compat.cast_unicode(text, py3compat.DEFAULT_ENCODING)
     return text
 
 
 def wayland_clipboard_get():
     """Get the clipboard's text under Wayland using wl-paste command.
 
     This requires Wayland and wl-clipboard installed and running.
     """
     if os.environ.get("XDG_SESSION_TYPE") != "wayland":
         raise TryNext("wayland is not detected")
 
     try:
         with subprocess.Popen(["wl-paste"], stdout=subprocess.PIPE) as p:
             raw, err = p.communicate()
             if p.wait():
                 raise TryNext(err)
     except FileNotFoundError as e:
         raise TryNext(
             "Getting text from the clipboard under Wayland requires the wl-clipboard "
             "extension: https://github.com/bugaevc/wl-clipboard"
         ) from e
 
     if not raw:
         raise ClipboardEmpty
 
     try:
         text = py3compat.decode(raw)
     except UnicodeDecodeError as e:
         raise ClipboardEmpty from e
 
     return text

IPython/sphinxext/custom_doctests.py

@@ -1,155 +1,155
 """
 Handlers for IPythonDirective's @doctest pseudo-decorator.
 
 The Sphinx extension that provides support for embedded IPython code provides
 a pseudo-decorator @doctest, which treats the input/output block as a
 doctest, raising a RuntimeError during doc generation if the actual output
 (after running the input) does not match the expected output.
 
 An example usage is:
 
 .. code-block:: rst
 
    .. ipython::
 
         In [1]: x = 1
 
         @doctest
         In [2]: x + 2
         Out[3]: 3
 
 One can also provide arguments to the decorator. The first argument should be
 the name of a custom handler. The specification of any other arguments is
 determined by the handler. For example,
 
 .. code-block:: rst
 
       .. ipython::
 
          @doctest float
          In [154]: 0.1 + 0.2
          Out[154]: 0.3
 
 allows the actual output ``0.30000000000000004`` to match the expected output
 due to a comparison with `np.allclose`.
 
 This module contains handlers for the @doctest pseudo-decorator. Handlers
 should have the following function signature::
 
     handler(sphinx_shell, args, input_lines, found, submitted)
 
 where `sphinx_shell` is the embedded Sphinx shell, `args` contains the list
 of arguments that follow: '@doctest handler_name', `input_lines` contains
 a list of the lines relevant to the current doctest, `found` is a string
 containing the output from the IPython shell, and `submitted` is a string
 containing the expected output from the IPython shell.
 
 Handlers must be registered in the `doctests` dict at the end of this module.
 
 """
 
 def str_to_array(s):
     """
     Simplistic converter of strings from repr to float NumPy arrays.
 
     If the repr representation has ellipsis in it, then this will fail.
 
     Parameters
     ----------
     s : str
         The repr version of a NumPy array.
 
     Examples
     --------
     >>> s = "array([ 0.3,  inf,  nan])"
     >>> a = str_to_array(s)
 
     """
     import numpy as np
 
     # Need to make sure eval() knows about inf and nan.
     # This also assumes default printoptions for NumPy.
     from numpy import inf, nan
 
     if s.startswith(u'array'):
         # Remove array( and )
         s = s[6:-1]
 
     if s.startswith(u'['):
         a = np.array(eval(s), dtype=float)
     else:
         # Assume its a regular float. Force 1D so we can index into it.
         a = np.atleast_1d(float(s))
     return a
 
 def float_doctest(sphinx_shell, args, input_lines, found, submitted):
     """
     Doctest which allow the submitted output to vary slightly from the input.
 
     Here is how it might appear in an rst file:
 
     .. code-block:: rst
 
        .. ipython::
 
           @doctest float
           In [1]: 0.1 + 0.2
           Out[1]: 0.3
 
     """
     import numpy as np
 
     if len(args) == 2:
         rtol = 1e-05
         atol = 1e-08
     else:
         # Both must be specified if any are specified.
         try:
             rtol = float(args[2])
             atol = float(args[3])
-        except IndexError as e:
+        except IndexError:
             e = ("Both `rtol` and `atol` must be specified "
                  "if either are specified: {0}".format(args))
             raise IndexError(e) from e
 
     try:
         submitted = str_to_array(submitted)
         found = str_to_array(found)
     except:
         # For example, if the array is huge and there are ellipsis in it.
         error = True
     else:
         found_isnan = np.isnan(found)
         submitted_isnan = np.isnan(submitted)
         error = not np.allclose(found_isnan, submitted_isnan)
         error |= not np.allclose(found[~found_isnan],
                                  submitted[~submitted_isnan],
                                  rtol=rtol, atol=atol)
 
     TAB = ' ' * 4
     directive = sphinx_shell.directive
     if directive is None:
         source = 'Unavailable'
         content = 'Unavailable'
     else:
         source = directive.state.document.current_source
         # Add tabs and make into a single string.
         content = '\n'.join([TAB + line for line in directive.content])
 
     if error:
 
         e = ('doctest float comparison failure\n\n'
              'Document source: {0}\n\n'
              'Raw content: \n{1}\n\n'
              'On input line(s):\n{TAB}{2}\n\n'
              'we found output:\n{TAB}{3}\n\n'
              'instead of the expected:\n{TAB}{4}\n\n')
         e = e.format(source, content, '\n'.join(input_lines), repr(found),
                      repr(submitted), TAB=TAB)
         raise RuntimeError(e)
 
 # dict of allowable doctest handlers. The key represents the first argument
 # that must be given to @doctest in order to activate the handler.
 doctests = {
     'float': float_doctest,
 }

IPython/sphinxext/ipython_directive.py

@@ -1,1276 +1,1278
 # -*- coding: utf-8 -*-
 """
 Sphinx directive to support embedded IPython code.
 
 IPython provides an extension for `Sphinx <http://www.sphinx-doc.org/>`_ to
 highlight and run code.
 
 This directive allows pasting of entire interactive IPython sessions, prompts
 and all, and their code will actually get re-executed at doc build time, with
 all prompts renumbered sequentially. It also allows you to input code as a pure
 python input by giving the argument python to the directive. The output looks
 like an interactive ipython section.
 
 Here is an example of how the IPython directive can
 **run** python code, at build time.
 
 .. ipython::
 
    In [1]: 1+1
 
    In [1]: import datetime
       ...: datetime.date.fromisoformat('2022-02-22')
 
 It supports IPython construct that plain
 Python does not understand (like magics):
 
 .. ipython::
 
    In [0]: import time
 
    In [0]: %pdoc time.sleep
 
 This will also support top-level async when using IPython 7.0+
 
 .. ipython::
 
    In [2]: import asyncio
       ...: print('before')
       ...: await asyncio.sleep(1)
       ...: print('after')
 
 
 The namespace will persist across multiple code chucks, Let's define a variable:
 
 .. ipython::
 
    In [0]: who = "World"
 
 And now say hello:
 
 .. ipython::
 
    In [0]: print('Hello,', who)
 
 If the current section raises an exception, you can add the ``:okexcept:`` flag
 to the current block, otherwise the build will fail.
 
 .. ipython::
    :okexcept:
 
    In [1]: 1/0
 
 IPython Sphinx directive module
 ===============================
 
 To enable this directive, simply list it in your Sphinx ``conf.py`` file
 (making sure the directory where you placed it is visible to sphinx, as is
 needed for all Sphinx directives). For example, to enable syntax highlighting
 and the IPython directive::
 
     extensions = ['IPython.sphinxext.ipython_console_highlighting',
                   'IPython.sphinxext.ipython_directive']
 
 The IPython directive outputs code-blocks with the language 'ipython'. So
 if you do not have the syntax highlighting extension enabled as well, then
 all rendered code-blocks will be uncolored. By default this directive assumes
 that your prompts are unchanged IPython ones, but this can be customized.
 The configurable options that can be placed in conf.py are:
 
 ipython_savefig_dir:
     The directory in which to save the figures. This is relative to the
     Sphinx source directory. The default is `html_static_path`.
 ipython_rgxin:
     The compiled regular expression to denote the start of IPython input
     lines. The default is ``re.compile('In \\[(\\d+)\\]:\\s?(.*)\\s*')``. You
     shouldn't need to change this.
 ipython_warning_is_error: [default to True]
     Fail the build if something unexpected happen, for example if a block raise
     an exception but does not have the `:okexcept:` flag. The exact behavior of
     what is considered strict, may change between the sphinx directive version.
 ipython_rgxout:
     The compiled regular expression to denote the start of IPython output
     lines. The default is ``re.compile('Out\\[(\\d+)\\]:\\s?(.*)\\s*')``. You
     shouldn't need to change this.
 ipython_promptin:
     The string to represent the IPython input prompt in the generated ReST.
     The default is ``'In [%d]:'``. This expects that the line numbers are used
     in the prompt.
 ipython_promptout:
     The string to represent the IPython prompt in the generated ReST. The
     default is ``'Out [%d]:'``. This expects that the line numbers are used
     in the prompt.
 ipython_mplbackend:
     The string which specifies if the embedded Sphinx shell should import
     Matplotlib and set the backend. The value specifies a backend that is
     passed to `matplotlib.use()` before any lines in `ipython_execlines` are
     executed. If not specified in conf.py, then the default value of 'agg' is
     used. To use the IPython directive without matplotlib as a dependency, set
     the value to `None`. It may end up that matplotlib is still imported
     if the user specifies so in `ipython_execlines` or makes use of the
     @savefig pseudo decorator.
 ipython_execlines:
     A list of strings to be exec'd in the embedded Sphinx shell. Typical
     usage is to make certain packages always available. Set this to an empty
     list if you wish to have no imports always available. If specified in
     ``conf.py`` as `None`, then it has the effect of making no imports available.
     If omitted from conf.py altogether, then the default value of
     ['import numpy as np', 'import matplotlib.pyplot as plt'] is used.
 ipython_holdcount
     When the @suppress pseudo-decorator is used, the execution count can be
     incremented or not. The default behavior is to hold the execution count,
     corresponding to a value of `True`. Set this to `False` to increment
     the execution count after each suppressed command.
 
 As an example, to use the IPython directive when `matplotlib` is not available,
 one sets the backend to `None`::
 
     ipython_mplbackend = None
 
 An example usage of the directive is:
 
 .. code-block:: rst
 
     .. ipython::
 
         In [1]: x = 1
 
         In [2]: y = x**2
 
         In [3]: print(y)
 
 See http://matplotlib.org/sampledoc/ipython_directive.html for additional
 documentation.
 
 Pseudo-Decorators
 =================
 
 Note: Only one decorator is supported per input. If more than one decorator
 is specified, then only the last one is used.
 
 In addition to the Pseudo-Decorators/options described at the above link,
 several enhancements have been made. The directive will emit a message to the
 console at build-time if code-execution resulted in an exception or warning.
 You can suppress these on a per-block basis by specifying the :okexcept:
 or :okwarning: options:
 
 .. code-block:: rst
 
     .. ipython::
         :okexcept:
         :okwarning:
 
         In [1]: 1/0
         In [2]: # raise warning.
 
 To Do
 =====
 
 - Turn the ad-hoc test() function into a real test suite.
 - Break up ipython-specific functionality from matplotlib stuff into better
   separated code.
 
 """
 
 # Authors
 # =======
-# 
+#
 # - John D Hunter: original author.
 # - Fernando Perez: refactoring, documentation, cleanups, port to 0.11.
 # - VáclavŠmilauer <eudoxos-AT-arcig.cz>: Prompt generalizations.
 # - Skipper Seabold, refactoring, cleanups, pure python addition
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Stdlib
 import atexit
 import errno
 import os
 import pathlib
 import re
 import sys
 import tempfile
 import ast
 import warnings
 import shutil
 from io import StringIO
+from typing import Any, Dict, Set
 
 # Third-party
 from docutils.parsers.rst import directives
 from docutils.parsers.rst import Directive
 from sphinx.util import logging
 
 # Our own
 from traitlets.config import Config
 from IPython import InteractiveShell
 from IPython.core.profiledir import ProfileDir
 
 use_matplotlib = False
 try:
     import matplotlib
     use_matplotlib = True
 except Exception:
     pass
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 # for tokenizing blocks
 COMMENT, INPUT, OUTPUT =  range(3)
 
 PSEUDO_DECORATORS = ["suppress", "verbatim", "savefig", "doctest"]
 
 #-----------------------------------------------------------------------------
 # Functions and class declarations
 #-----------------------------------------------------------------------------
 
 def block_parser(part, rgxin, rgxout, fmtin, fmtout):
     """
     part is a string of ipython text, comprised of at most one
     input, one output, comments, and blank lines.  The block parser
     parses the text into a list of::
 
       blocks = [ (TOKEN0, data0), (TOKEN1, data1), ...]
 
     where TOKEN is one of [COMMENT | INPUT | OUTPUT ] and
     data is, depending on the type of token::
 
       COMMENT : the comment string
 
       INPUT: the (DECORATOR, INPUT_LINE, REST) where
          DECORATOR: the input decorator (or None)
          INPUT_LINE: the input as string (possibly multi-line)
          REST : any stdout generated by the input line (not OUTPUT)
 
       OUTPUT: the output string, possibly multi-line
 
     """
     block = []
     lines = part.split('\n')
     N = len(lines)
     i = 0
     decorator = None
     while 1:
 
         if i==N:
             # nothing left to parse -- the last line
             break
 
         line = lines[i]
         i += 1
         line_stripped = line.strip()
         if line_stripped.startswith('#'):
             block.append((COMMENT, line))
             continue
 
         if any(
             line_stripped.startswith("@" + pseudo_decorator)
             for pseudo_decorator in PSEUDO_DECORATORS
         ):
             if decorator:
                 raise RuntimeError(
                     "Applying multiple pseudo-decorators on one line is not supported"
                 )
             else:
                 decorator = line_stripped
                 continue
 
         # does this look like an input line?
         matchin = rgxin.match(line)
         if matchin:
             lineno, inputline = int(matchin.group(1)), matchin.group(2)
 
             # the ....: continuation string
             continuation = '   %s:'%''.join(['.']*(len(str(lineno))+2))
             Nc = len(continuation)
             # input lines can continue on for more than one line, if
             # we have a '\' line continuation char or a function call
             # echo line 'print'.  The input line can only be
             # terminated by the end of the block or an output line, so
             # we parse out the rest of the input line if it is
             # multiline as well as any echo text
 
             rest = []
             while i<N:
 
                 # look ahead; if the next line is blank, or a comment, or
                 # an output line, we're done
 
                 nextline = lines[i]
                 matchout = rgxout.match(nextline)
                 # print("nextline=%s, continuation=%s, starts=%s"%(nextline, continuation, nextline.startswith(continuation)))
                 if matchout or nextline.startswith('#'):
                     break
                 elif nextline.startswith(continuation):
                     # The default ipython_rgx* treat the space following the colon as optional.
                     # However, If the space is there we must consume it or code
                     # employing the cython_magic extension will fail to execute.
                     #
                     # This works with the default ipython_rgx* patterns,
                     # If you modify them, YMMV.
                     nextline = nextline[Nc:]
                     if nextline and nextline[0] == ' ':
                         nextline = nextline[1:]
 
                     inputline += '\n' +  nextline
                 else:
                     rest.append(nextline)
                 i+= 1
 
             block.append((INPUT, (decorator, inputline, '\n'.join(rest))))
             continue
 
         # if it looks like an output line grab all the text to the end
         # of the block
         matchout = rgxout.match(line)
         if matchout:
             lineno, output = int(matchout.group(1)), matchout.group(2)
             if i<N-1:
                 output = '\n'.join([output] + lines[i:])
 
             block.append((OUTPUT, output))
             break
 
     return block
 
 
 class EmbeddedSphinxShell(object):
     """An embedded IPython instance to run inside Sphinx"""
 
     def __init__(self, exec_lines=None):
 
         self.cout = StringIO()
 
         if exec_lines is None:
             exec_lines = []
 
         # Create config object for IPython
         config = Config()
         config.HistoryManager.hist_file = ':memory:'
         config.InteractiveShell.autocall = False
         config.InteractiveShell.autoindent = False
         config.InteractiveShell.colors = 'NoColor'
 
         # create a profile so instance history isn't saved
         tmp_profile_dir = tempfile.mkdtemp(prefix='profile_')
         profname = 'auto_profile_sphinx_build'
         pdir = os.path.join(tmp_profile_dir,profname)
         profile = ProfileDir.create_profile_dir(pdir)
 
         # Create and initialize global ipython, but don't start its mainloop.
         # This will persist across different EmbeddedSphinxShell instances.
         IP = InteractiveShell.instance(config=config, profile_dir=profile)
         atexit.register(self.cleanup)
 
         # Store a few parts of IPython we'll need.
         self.IP = IP
         self.user_ns = self.IP.user_ns
         self.user_global_ns = self.IP.user_global_ns
 
         self.input = ''
         self.output = ''
         self.tmp_profile_dir = tmp_profile_dir
 
         self.is_verbatim = False
         self.is_doctest = False
         self.is_suppress = False
 
         # Optionally, provide more detailed information to shell.
         # this is assigned by the SetUp method of IPythonDirective
         # to point at itself.
         #
         # So, you can access handy things at self.directive.state
         self.directive = None
 
         # on the first call to the savefig decorator, we'll import
         # pyplot as plt so we can make a call to the plt.gcf().savefig
         self._pyplot_imported = False
 
         # Prepopulate the namespace.
         for line in exec_lines:
             self.process_input_line(line, store_history=False)
 
     def cleanup(self):
         shutil.rmtree(self.tmp_profile_dir, ignore_errors=True)
 
     def clear_cout(self):
         self.cout.seek(0)
         self.cout.truncate(0)
 
     def process_input_line(self, line, store_history):
         return self.process_input_lines([line], store_history=store_history)
 
     def process_input_lines(self, lines, store_history=True):
         """process the input, capturing stdout"""
         stdout = sys.stdout
         source_raw = '\n'.join(lines)
         try:
             sys.stdout = self.cout
             self.IP.run_cell(source_raw, store_history=store_history)
         finally:
             sys.stdout = stdout
 
     def process_image(self, decorator):
         """
         # build out an image directive like
         # .. image:: somefile.png
         #    :width 4in
         #
         # from an input like
         # savefig somefile.png width=4in
         """
         savefig_dir = self.savefig_dir
         source_dir = self.source_dir
         saveargs = decorator.split(' ')
         filename = saveargs[1]
-        # insert relative path to image file in source 
+        # insert relative path to image file in source
         # as absolute path for Sphinx
         # sphinx expects a posix path, even on Windows
         path = pathlib.Path(savefig_dir, filename)
         outfile = '/' + path.relative_to(source_dir).as_posix()
 
         imagerows = ['.. image:: %s' % outfile]
 
         for kwarg in saveargs[2:]:
             arg, val = kwarg.split('=')
             arg = arg.strip()
             val = val.strip()
             imagerows.append('   :%s: %s'%(arg, val))
 
         image_file = os.path.basename(outfile) # only return file name
         image_directive = '\n'.join(imagerows)
         return image_file, image_directive
 
     # Callbacks for each type of token
     def process_input(self, data, input_prompt, lineno):
         """
         Process data block for INPUT token.
 
         """
         decorator, input, rest = data
         image_file = None
         image_directive = None
 
         is_verbatim = decorator=='@verbatim' or self.is_verbatim
         is_doctest = (decorator is not None and \
                      decorator.startswith('@doctest')) or self.is_doctest
         is_suppress = decorator=='@suppress' or self.is_suppress
         is_okexcept = decorator=='@okexcept' or self.is_okexcept
         is_okwarning = decorator=='@okwarning' or self.is_okwarning
         is_savefig = decorator is not None and \
                      decorator.startswith('@savefig')
 
         input_lines = input.split('\n')
         if len(input_lines) > 1:
             if input_lines[-1] != "":
                 input_lines.append('') # make sure there's a blank line
                                        # so splitter buffer gets reset
 
         continuation = '   %s:'%''.join(['.']*(len(str(lineno))+2))
 
         if is_savefig:
             image_file, image_directive = self.process_image(decorator)
 
         ret = []
         is_semicolon = False
 
         # Hold the execution count, if requested to do so.
         if is_suppress and self.hold_count:
             store_history = False
         else:
             store_history = True
 
         # Note: catch_warnings is not thread safe
         with warnings.catch_warnings(record=True) as ws:
             if input_lines[0].endswith(';'):
                 is_semicolon = True
             #for i, line in enumerate(input_lines):
 
             # process the first input line
             if is_verbatim:
                 self.process_input_lines([''])
                 self.IP.execution_count += 1 # increment it anyway
             else:
                 # only submit the line in non-verbatim mode
                 self.process_input_lines(input_lines, store_history=store_history)
 
         if not is_suppress:
             for i, line in enumerate(input_lines):
                 if i == 0:
                     formatted_line = '%s %s'%(input_prompt, line)
                 else:
                     formatted_line = '%s %s'%(continuation, line)
                 ret.append(formatted_line)
 
         if not is_suppress and len(rest.strip()) and is_verbatim:
             # The "rest" is the standard output of the input. This needs to be
             # added when in verbatim mode. If there is no "rest", then we don't
             # add it, as the new line will be added by the processed output.
             ret.append(rest)
 
         # Fetch the processed output. (This is not the submitted output.)
         self.cout.seek(0)
         processed_output = self.cout.read()
         if not is_suppress and not is_semicolon:
             #
             # In IPythonDirective.run, the elements of `ret` are eventually
             # combined such that '' entries correspond to newlines. So if
             # `processed_output` is equal to '', then the adding it to `ret`
             # ensures that there is a blank line between consecutive inputs
             # that have no outputs, as in:
             #
             #    In [1]: x = 4
             #
             #    In [2]: x = 5
             #
             # When there is processed output, it has a '\n' at the tail end. So
             # adding the output to `ret` will provide the necessary spacing
             # between consecutive input/output blocks, as in:
             #
             #   In [1]: x
             #   Out[1]: 5
             #
             #   In [2]: x
             #   Out[2]: 5
             #
             # When there is stdout from the input, it also has a '\n' at the
             # tail end, and so this ensures proper spacing as well. E.g.:
             #
             #   In [1]: print(x)
             #   5
             #
             #   In [2]: x = 5
             #
             # When in verbatim mode, `processed_output` is empty (because
             # nothing was passed to IP. Sometimes the submitted code block has
             # an Out[] portion and sometimes it does not. When it does not, we
             # need to ensure proper spacing, so we have to add '' to `ret`.
             # However, if there is an Out[] in the submitted code, then we do
             # not want to add a newline as `process_output` has stuff to add.
             # The difficulty is that `process_input` doesn't know if
             # `process_output` will be called---so it doesn't know if there is
             # Out[] in the code block. The requires that we include a hack in
             # `process_block`. See the comments there.
             #
             ret.append(processed_output)
         elif is_semicolon:
             # Make sure there is a newline after the semicolon.
             ret.append('')
 
         # context information
         filename = "Unknown"
         lineno = 0
         if self.directive.state:
             filename = self.directive.state.document.current_source
             lineno = self.directive.state.document.current_line
 
         # Use sphinx logger for warnings
         logger = logging.getLogger(__name__)
 
         # output any exceptions raised during execution to stdout
         # unless :okexcept: has been specified.
         if not is_okexcept and (
             ("Traceback" in processed_output) or ("SyntaxError" in processed_output)
         ):
             s = "\n>>>" + ("-" * 73) + "\n"
             s += "Exception in %s at block ending on line %s\n" % (filename, lineno)
             s += "Specify :okexcept: as an option in the ipython:: block to suppress this message\n"
             s += processed_output + "\n"
             s += "<<<" + ("-" * 73)
             logger.warning(s)
             if self.warning_is_error:
                 raise RuntimeError(
                     "Unexpected exception in `{}` line {}".format(filename, lineno)
                 )
 
         # output any warning raised during execution to stdout
         # unless :okwarning: has been specified.
         if not is_okwarning:
             for w in ws:
                 s = "\n>>>" + ("-" * 73) + "\n"
                 s += "Warning in %s at block ending on line %s\n" % (filename, lineno)
                 s += "Specify :okwarning: as an option in the ipython:: block to suppress this message\n"
                 s += ("-" * 76) + "\n"
                 s += warnings.formatwarning(
                     w.message, w.category, w.filename, w.lineno, w.line
                 )
                 s += "<<<" + ("-" * 73)
                 logger.warning(s)
                 if self.warning_is_error:
                     raise RuntimeError(
                         "Unexpected warning in `{}` line {}".format(filename, lineno)
                     )
 
         self.clear_cout()
         return (ret, input_lines, processed_output,
                 is_doctest, decorator, image_file, image_directive)
 
 
     def process_output(self, data, output_prompt, input_lines, output,
                        is_doctest, decorator, image_file):
         """
         Process data block for OUTPUT token.
 
         """
         # Recall: `data` is the submitted output, and `output` is the processed
         # output from `input_lines`.
 
         TAB = ' ' * 4
 
         if is_doctest and output is not None:
 
             found = output # This is the processed output
             found = found.strip()
             submitted = data.strip()
 
             if self.directive is None:
                 source = 'Unavailable'
                 content = 'Unavailable'
             else:
                 source = self.directive.state.document.current_source
                 content = self.directive.content
                 # Add tabs and join into a single string.
                 content = '\n'.join([TAB + line for line in content])
 
             # Make sure the output contains the output prompt.
             ind = found.find(output_prompt)
             if ind < 0:
                 e = ('output does not contain output prompt\n\n'
                      'Document source: {0}\n\n'
                      'Raw content: \n{1}\n\n'
                      'Input line(s):\n{TAB}{2}\n\n'
                      'Output line(s):\n{TAB}{3}\n\n')
                 e = e.format(source, content, '\n'.join(input_lines),
                              repr(found), TAB=TAB)
                 raise RuntimeError(e)
             found = found[len(output_prompt):].strip()
 
             # Handle the actual doctest comparison.
             if decorator.strip() == '@doctest':
                 # Standard doctest
                 if found != submitted:
                     e = ('doctest failure\n\n'
                          'Document source: {0}\n\n'
                          'Raw content: \n{1}\n\n'
                          'On input line(s):\n{TAB}{2}\n\n'
                          'we found output:\n{TAB}{3}\n\n'
                          'instead of the expected:\n{TAB}{4}\n\n')
                     e = e.format(source, content, '\n'.join(input_lines),
                                  repr(found), repr(submitted), TAB=TAB)
                     raise RuntimeError(e)
             else:
                 self.custom_doctest(decorator, input_lines, found, submitted)
 
         # When in verbatim mode, this holds additional submitted output
         # to be written in the final Sphinx output.
         # https://github.com/ipython/ipython/issues/5776
         out_data = []
 
         is_verbatim = decorator=='@verbatim' or self.is_verbatim
         if is_verbatim and data.strip():
             # Note that `ret` in `process_block` has '' as its last element if
             # the code block was in verbatim mode. So if there is no submitted
             # output, then we will have proper spacing only if we do not add
             # an additional '' to `out_data`. This is why we condition on
             # `and data.strip()`.
 
             # The submitted output has no output prompt. If we want the
             # prompt and the code to appear, we need to join them now
             # instead of adding them separately---as this would create an
             # undesired newline. How we do this ultimately depends on the
             # format of the output regex. I'll do what works for the default
             # prompt for now, and we might have to adjust if it doesn't work
             # in other cases. Finally, the submitted output does not have
             # a trailing newline, so we must add it manually.
             out_data.append("{0} {1}\n".format(output_prompt, data))
 
         return out_data
 
     def process_comment(self, data):
         """Process data fPblock for COMMENT token."""
         if not self.is_suppress:
             return [data]
 
     def save_image(self, image_file):
         """
         Saves the image file to disk.
         """
         self.ensure_pyplot()
         command = 'plt.gcf().savefig("%s")'%image_file
         # print('SAVEFIG', command)  # dbg
         self.process_input_line('bookmark ipy_thisdir', store_history=False)
         self.process_input_line('cd -b ipy_savedir', store_history=False)
         self.process_input_line(command, store_history=False)
         self.process_input_line('cd -b ipy_thisdir', store_history=False)
         self.process_input_line('bookmark -d ipy_thisdir', store_history=False)
         self.clear_cout()
 
     def process_block(self, block):
         """
         process block from the block_parser and return a list of processed lines
         """
         ret = []
         output = None
         input_lines = None
         lineno = self.IP.execution_count
 
         input_prompt = self.promptin % lineno
         output_prompt = self.promptout % lineno
         image_file = None
         image_directive = None
 
         found_input = False
         for token, data in block:
             if token == COMMENT:
                 out_data = self.process_comment(data)
             elif token == INPUT:
                 found_input = True
                 (out_data, input_lines, output, is_doctest,
                  decorator, image_file, image_directive) = \
                           self.process_input(data, input_prompt, lineno)
             elif token == OUTPUT:
                 if not found_input:
 
                     TAB = ' ' * 4
                     linenumber = 0
                     source = 'Unavailable'
                     content = 'Unavailable'
                     if self.directive:
                         linenumber = self.directive.state.document.current_line
                         source = self.directive.state.document.current_source
                         content = self.directive.content
                         # Add tabs and join into a single string.
                         content = '\n'.join([TAB + line for line in content])
 
                     e = ('\n\nInvalid block: Block contains an output prompt '
                          'without an input prompt.\n\n'
                          'Document source: {0}\n\n'
                          'Content begins at line {1}: \n\n{2}\n\n'
                          'Problematic block within content: \n\n{TAB}{3}\n\n')
                     e = e.format(source, linenumber, content, block, TAB=TAB)
 
                     # Write, rather than include in exception, since Sphinx
                     # will truncate tracebacks.
                     sys.stdout.write(e)
                     raise RuntimeError('An invalid block was detected.')
                 out_data = \
                     self.process_output(data, output_prompt, input_lines,
                                         output, is_doctest, decorator,
                                         image_file)
                 if out_data:
                     # Then there was user submitted output in verbatim mode.
                     # We need to remove the last element of `ret` that was
                     # added in `process_input`, as it is '' and would introduce
                     # an undesirable newline.
                     assert(ret[-1] == '')
                     del ret[-1]
 
             if out_data:
                 ret.extend(out_data)
 
         # save the image files
         if image_file is not None:
             self.save_image(image_file)
 
         return ret, image_directive
 
     def ensure_pyplot(self):
         """
         Ensures that pyplot has been imported into the embedded IPython shell.
 
         Also, makes sure to set the backend appropriately if not set already.
 
         """
         # We are here if the @figure pseudo decorator was used. Thus, it's
         # possible that we could be here even if python_mplbackend were set to
         # `None`. That's also strange and perhaps worthy of raising an
         # exception, but for now, we just set the backend to 'agg'.
 
         if not self._pyplot_imported:
             if 'matplotlib.backends' not in sys.modules:
                 # Then ipython_matplotlib was set to None but there was a
                 # call to the @figure decorator (and ipython_execlines did
                 # not set a backend).
                 #raise Exception("No backend was set, but @figure was used!")
                 import matplotlib
                 matplotlib.use('agg')
 
             # Always import pyplot into embedded shell.
             self.process_input_line('import matplotlib.pyplot as plt',
                                     store_history=False)
             self._pyplot_imported = True
 
     def process_pure_python(self, content):
         """
         content is a list of strings. it is unedited directive content
 
         This runs it line by line in the InteractiveShell, prepends
         prompts as needed capturing stderr and stdout, then returns
         the content as a list as if it were ipython code
         """
         output = []
         savefig = False # keep up with this to clear figure
         multiline = False # to handle line continuation
         multiline_start = None
         fmtin = self.promptin
 
         ct = 0
 
         for lineno, line in enumerate(content):
 
             line_stripped = line.strip()
             if not len(line):
                 output.append(line)
                 continue
 
             # handle pseudo-decorators, whilst ensuring real python decorators are treated as input
             if any(
                 line_stripped.startswith("@" + pseudo_decorator)
                 for pseudo_decorator in PSEUDO_DECORATORS
             ):
                 output.extend([line])
                 if 'savefig' in line:
                     savefig = True # and need to clear figure
                 continue
 
             # handle comments
             if line_stripped.startswith('#'):
                 output.extend([line])
                 continue
 
             # deal with lines checking for multiline
             continuation  = u'   %s:'% ''.join(['.']*(len(str(ct))+2))
             if not multiline:
                 modified = u"%s %s" % (fmtin % ct, line_stripped)
                 output.append(modified)
                 ct += 1
                 try:
                     ast.parse(line_stripped)
                     output.append(u'')
                 except Exception: # on a multiline
                     multiline = True
                     multiline_start = lineno
             else: # still on a multiline
                 modified = u'%s %s' % (continuation, line)
                 output.append(modified)
 
                 # if the next line is indented, it should be part of multiline
                 if len(content) > lineno + 1:
                     nextline = content[lineno + 1]
                     if len(nextline) - len(nextline.lstrip()) > 3:
                         continue
                 try:
                     mod = ast.parse(
                             '\n'.join(content[multiline_start:lineno+1]))
                     if isinstance(mod.body[0], ast.FunctionDef):
                         # check to see if we have the whole function
                         for element in mod.body[0].body:
                             if isinstance(element, ast.Return):
                                 multiline = False
                     else:
                         output.append(u'')
                         multiline = False
                 except Exception:
                     pass
 
             if savefig: # clear figure if plotted
                 self.ensure_pyplot()
                 self.process_input_line('plt.clf()', store_history=False)
                 self.clear_cout()
                 savefig = False
 
         return output
 
     def custom_doctest(self, decorator, input_lines, found, submitted):
         """
         Perform a specialized doctest.
 
         """
         from .custom_doctests import doctests
 
         args = decorator.split()
         doctest_type = args[1]
         if doctest_type in doctests:
             doctests[doctest_type](self, args, input_lines, found, submitted)
         else:
             e = "Invalid option to @doctest: {0}".format(doctest_type)
             raise Exception(e)
 
 
 class IPythonDirective(Directive):
 
-    has_content = True
-    required_arguments = 0
-    optional_arguments = 4 # python, suppress, verbatim, doctest
-    final_argumuent_whitespace = True
-    option_spec = { 'python': directives.unchanged,
-                    'suppress' : directives.flag,
-                    'verbatim' : directives.flag,
-                    'doctest' : directives.flag,
-                    'okexcept': directives.flag,
-                    'okwarning': directives.flag
-                  }
+    has_content: bool = True
+    required_arguments: int = 0
+    optional_arguments: int = 4  # python, suppress, verbatim, doctest
+    final_argumuent_whitespace: bool = True
+    option_spec: Dict[str, Any] = {
+        "python": directives.unchanged,
+        "suppress": directives.flag,
+        "verbatim": directives.flag,
+        "doctest": directives.flag,
+        "okexcept": directives.flag,
+        "okwarning": directives.flag,
+    }
 
     shell = None
 
-    seen_docs = set()
+    seen_docs: Set = set()
 
     def get_config_options(self):
         # contains sphinx configuration variables
         config = self.state.document.settings.env.config
 
         # get config variables to set figure output directory
         savefig_dir = config.ipython_savefig_dir
         source_dir = self.state.document.settings.env.srcdir
         savefig_dir = os.path.join(source_dir, savefig_dir)
 
         # get regex and prompt stuff
         rgxin      = config.ipython_rgxin
         rgxout     = config.ipython_rgxout
         warning_is_error= config.ipython_warning_is_error
         promptin   = config.ipython_promptin
         promptout  = config.ipython_promptout
         mplbackend = config.ipython_mplbackend
         exec_lines = config.ipython_execlines
         hold_count = config.ipython_holdcount
 
         return (savefig_dir, source_dir, rgxin, rgxout,
                 promptin, promptout, mplbackend, exec_lines, hold_count, warning_is_error)
 
     def setup(self):
         # Get configuration values.
         (savefig_dir, source_dir, rgxin, rgxout, promptin, promptout,
          mplbackend, exec_lines, hold_count, warning_is_error) = self.get_config_options()
 
         try:
             os.makedirs(savefig_dir)
         except OSError as e:
             if e.errno != errno.EEXIST:
                 raise
 
         if self.shell is None:
             # We will be here many times.  However, when the
             # EmbeddedSphinxShell is created, its interactive shell member
             # is the same for each instance.
 
             if mplbackend and 'matplotlib.backends' not in sys.modules and use_matplotlib:
                 import matplotlib
                 matplotlib.use(mplbackend)
 
             # Must be called after (potentially) importing matplotlib and
             # setting its backend since exec_lines might import pylab.
             self.shell = EmbeddedSphinxShell(exec_lines)
 
             # Store IPython directive to enable better error messages
             self.shell.directive = self
 
         # reset the execution count if we haven't processed this doc
         #NOTE: this may be borked if there are multiple seen_doc tmp files
         #check time stamp?
         if not self.state.document.current_source in self.seen_docs:
             self.shell.IP.history_manager.reset()
             self.shell.IP.execution_count = 1
             self.seen_docs.add(self.state.document.current_source)
 
         # and attach to shell so we don't have to pass them around
         self.shell.rgxin = rgxin
         self.shell.rgxout = rgxout
         self.shell.promptin = promptin
         self.shell.promptout = promptout
         self.shell.savefig_dir = savefig_dir
         self.shell.source_dir = source_dir
         self.shell.hold_count = hold_count
         self.shell.warning_is_error = warning_is_error
 
         # setup bookmark for saving figures directory
         self.shell.process_input_line(
             'bookmark ipy_savedir "%s"' % savefig_dir, store_history=False
         )
         self.shell.clear_cout()
 
         return rgxin, rgxout, promptin, promptout
 
     def teardown(self):
         # delete last bookmark
         self.shell.process_input_line('bookmark -d ipy_savedir',
                                       store_history=False)
         self.shell.clear_cout()
 
     def run(self):
         debug = False
 
         #TODO, any reason block_parser can't be a method of embeddable shell
         # then we wouldn't have to carry these around
         rgxin, rgxout, promptin, promptout = self.setup()
 
         options = self.options
         self.shell.is_suppress = 'suppress' in options
         self.shell.is_doctest = 'doctest' in options
         self.shell.is_verbatim = 'verbatim' in options
         self.shell.is_okexcept = 'okexcept' in options
         self.shell.is_okwarning = 'okwarning' in options
 
         # handle pure python code
         if 'python' in self.arguments:
             content = self.content
             self.content = self.shell.process_pure_python(content)
 
         # parts consists of all text within the ipython-block.
         # Each part is an input/output block.
         parts = '\n'.join(self.content).split('\n\n')
 
         lines = ['.. code-block:: ipython', '']
         figures = []
 
         # Use sphinx logger for warnings
         logger = logging.getLogger(__name__)
 
         for part in parts:
             block = block_parser(part, rgxin, rgxout, promptin, promptout)
             if len(block):
                 rows, figure = self.shell.process_block(block)
                 for row in rows:
                     lines.extend(['   {0}'.format(line)
                                   for line in row.split('\n')])
 
                 if figure is not None:
                     figures.append(figure)
             else:
                 message = 'Code input with no code at {}, line {}'\
                             .format(
                                 self.state.document.current_source,
                                 self.state.document.current_line)
                 if self.shell.warning_is_error:
                     raise RuntimeError(message)
                 else:
                     logger.warning(message)
 
         for figure in figures:
             lines.append('')
             lines.extend(figure.split('\n'))
             lines.append('')
 
         if len(lines) > 2:
             if debug:
                 print('\n'.join(lines))
             else:
                 # This has to do with input, not output. But if we comment
                 # these lines out, then no IPython code will appear in the
                 # final output.
                 self.state_machine.insert_input(
                     lines, self.state_machine.input_lines.source(0))
 
         # cleanup
         self.teardown()
 
         return []
 
 # Enable as a proper Sphinx directive
 def setup(app):
     setup.app = app
 
     app.add_directive('ipython', IPythonDirective)
     app.add_config_value('ipython_savefig_dir', 'savefig', 'env')
     app.add_config_value('ipython_warning_is_error', True, 'env')
     app.add_config_value('ipython_rgxin',
                          re.compile(r'In \[(\d+)\]:\s?(.*)\s*'), 'env')
     app.add_config_value('ipython_rgxout',
                          re.compile(r'Out\[(\d+)\]:\s?(.*)\s*'), 'env')
     app.add_config_value('ipython_promptin', 'In [%d]:', 'env')
     app.add_config_value('ipython_promptout', 'Out[%d]:', 'env')
 
     # We could just let matplotlib pick whatever is specified as the default
     # backend in the matplotlibrc file, but this would cause issues if the
     # backend didn't work in headless environments. For this reason, 'agg'
     # is a good default backend choice.
     app.add_config_value('ipython_mplbackend', 'agg', 'env')
 
     # If the user sets this config value to `None`, then EmbeddedSphinxShell's
     # __init__ method will treat it as [].
     execlines = ['import numpy as np']
     if use_matplotlib:
         execlines.append('import matplotlib.pyplot as plt')
     app.add_config_value('ipython_execlines', execlines, 'env')
 
     app.add_config_value('ipython_holdcount', True, 'env')
 
     metadata = {'parallel_read_safe': True, 'parallel_write_safe': True}
     return metadata
 
 # Simple smoke test, needs to be converted to a proper automatic test.
 def test():
 
     examples = [
         r"""
 In [9]: pwd
 Out[9]: '/home/jdhunter/py4science/book'
 
 In [10]: cd bookdata/
 /home/jdhunter/py4science/book/bookdata
 
 In [2]: from pylab import *
 
 In [2]: ion()
 
 In [3]: im = imread('stinkbug.png')
 
 @savefig mystinkbug.png width=4in
 In [4]: imshow(im)
 Out[4]: <matplotlib.image.AxesImage object at 0x39ea850>
 
 """,
         r"""
 
 In [1]: x = 'hello world'
 
 # string methods can be
 # used to alter the string
 @doctest
 In [2]: x.upper()
 Out[2]: 'HELLO WORLD'
 
 @verbatim
 In [3]: x.st<TAB>
 x.startswith  x.strip
 """,
     r"""
 
 In [130]: url = 'http://ichart.finance.yahoo.com/table.csv?s=CROX\
    .....: &d=9&e=22&f=2009&g=d&a=1&br=8&c=2006&ignore=.csv'
 
 In [131]: print url.split('&')
 ['http://ichart.finance.yahoo.com/table.csv?s=CROX', 'd=9', 'e=22', 'f=2009', 'g=d', 'a=1', 'b=8', 'c=2006', 'ignore=.csv']
 
 In [60]: import urllib
 
 """,
     r"""\
 
 In [133]: import numpy.random
 
 @suppress
 In [134]: numpy.random.seed(2358)
 
 @doctest
 In [135]: numpy.random.rand(10,2)
 Out[135]:
 array([[ 0.64524308,  0.59943846],
        [ 0.47102322,  0.8715456 ],
        [ 0.29370834,  0.74776844],
        [ 0.99539577,  0.1313423 ],
        [ 0.16250302,  0.21103583],
        [ 0.81626524,  0.1312433 ],
        [ 0.67338089,  0.72302393],
        [ 0.7566368 ,  0.07033696],
        [ 0.22591016,  0.77731835],
        [ 0.0072729 ,  0.34273127]])
 
 """,
 
     r"""
 In [106]: print x
 jdh
 
 In [109]: for i in range(10):
    .....:     print i
    .....:
    .....:
 0
 1
 2
 3
 4
 5
 6
 7
 8
 9
 """,
 
         r"""
 
 In [144]: from pylab import *
 
 In [145]: ion()
 
 # use a semicolon to suppress the output
 @savefig test_hist.png width=4in
 In [151]: hist(np.random.randn(10000), 100);
 
 
 @savefig test_plot.png width=4in
 In [151]: plot(np.random.randn(10000), 'o');
    """,
 
         r"""
 # use a semicolon to suppress the output
 In [151]: plt.clf()
 
 @savefig plot_simple.png width=4in
 In [151]: plot([1,2,3])
 
 @savefig hist_simple.png width=4in
 In [151]: hist(np.random.randn(10000), 100);
 
 """,
      r"""
 # update the current fig
 In [151]: ylabel('number')
 
 In [152]: title('normal distribution')
 
 
 @savefig hist_with_text.png
 In [153]: grid(True)
 
 @doctest float
 In [154]: 0.1 + 0.2
 Out[154]: 0.3
 
 @doctest float
 In [155]: np.arange(16).reshape(4,4)
 Out[155]:
 array([[ 0,  1,  2,  3],
        [ 4,  5,  6,  7],
        [ 8,  9, 10, 11],
        [12, 13, 14, 15]])
 
 In [1]: x = np.arange(16, dtype=float).reshape(4,4)
 
 In [2]: x[0,0] = np.inf
 
 In [3]: x[0,1] = np.nan
 
 @doctest float
 In [4]: x
 Out[4]:
 array([[ inf,  nan,   2.,   3.],
        [  4.,   5.,   6.,   7.],
        [  8.,   9.,  10.,  11.],
        [ 12.,  13.,  14.,  15.]])
 
 
         """,
         ]
     # skip local-file depending first example:
     examples = examples[1:]
 
     #ipython_directive.DEBUG = True  # dbg
     #options = dict(suppress=True)  # dbg
     options = {}
     for example in examples:
         content = example.split('\n')
         IPythonDirective('debug', arguments=None, options=options,
                           content=content, lineno=0,
                           content_offset=None, block_text=None,
                           state=None, state_machine=None,
                           )
 
 # Run test suite as a script
 if __name__=='__main__':
     if not os.path.isdir('_static'):
         os.mkdir('_static')
     test()
     print('All OK? Check figures in _static/')

IPython/terminal/shortcuts/auto_match.py

@@ -1,104 +1,105
 """
 Utilities function for keybinding with prompt toolkit.
 
 This will be bound to specific key press and filter modes,
 like whether we are in edit mode, and whether the completer is open.
 """
+
 import re
 from prompt_toolkit.key_binding import KeyPressEvent
 
 
 def parenthesis(event: KeyPressEvent):
     """Auto-close parenthesis"""
     event.current_buffer.insert_text("()")
     event.current_buffer.cursor_left()
 
 
 def brackets(event: KeyPressEvent):
     """Auto-close brackets"""
     event.current_buffer.insert_text("[]")
     event.current_buffer.cursor_left()
 
 
 def braces(event: KeyPressEvent):
     """Auto-close braces"""
     event.current_buffer.insert_text("{}")
     event.current_buffer.cursor_left()
 
 
 def double_quote(event: KeyPressEvent):
     """Auto-close double quotes"""
     event.current_buffer.insert_text('""')
     event.current_buffer.cursor_left()
 
 
 def single_quote(event: KeyPressEvent):
     """Auto-close single quotes"""
     event.current_buffer.insert_text("''")
     event.current_buffer.cursor_left()
 
 
 def docstring_double_quotes(event: KeyPressEvent):
     """Auto-close docstring (double quotes)"""
     event.current_buffer.insert_text('""""')
     event.current_buffer.cursor_left(3)
 
 
 def docstring_single_quotes(event: KeyPressEvent):
     """Auto-close docstring (single quotes)"""
     event.current_buffer.insert_text("''''")
     event.current_buffer.cursor_left(3)
 
 
 def raw_string_parenthesis(event: KeyPressEvent):
     """Auto-close parenthesis in raw strings"""
     matches = re.match(
         r".*(r|R)[\"'](-*)",
         event.current_buffer.document.current_line_before_cursor,
     )
     dashes = matches.group(2) if matches else ""
     event.current_buffer.insert_text("()" + dashes)
     event.current_buffer.cursor_left(len(dashes) + 1)
 
 
 def raw_string_bracket(event: KeyPressEvent):
     """Auto-close bracker in raw strings"""
     matches = re.match(
         r".*(r|R)[\"'](-*)",
         event.current_buffer.document.current_line_before_cursor,
     )
     dashes = matches.group(2) if matches else ""
     event.current_buffer.insert_text("[]" + dashes)
     event.current_buffer.cursor_left(len(dashes) + 1)
 
 
 def raw_string_braces(event: KeyPressEvent):
     """Auto-close braces in raw strings"""
     matches = re.match(
         r".*(r|R)[\"'](-*)",
         event.current_buffer.document.current_line_before_cursor,
     )
     dashes = matches.group(2) if matches else ""
     event.current_buffer.insert_text("{}" + dashes)
     event.current_buffer.cursor_left(len(dashes) + 1)
 
 
 def skip_over(event: KeyPressEvent):
     """Skip over automatically added parenthesis/quote.
 
     (rather than adding another parenthesis/quote)"""
     event.current_buffer.cursor_right()
 
 
 def delete_pair(event: KeyPressEvent):
     """Delete auto-closed parenthesis"""
     event.current_buffer.delete()
     event.current_buffer.delete_before_cursor()
 
 
 auto_match_parens = {"(": parenthesis, "[": brackets, "{": braces}
 auto_match_parens_raw_string = {
     "(": raw_string_parenthesis,
     "[": raw_string_bracket,
     "{": raw_string_braces,
 }