Resurrect fast (non-highlighted) traceback code for long files. (#13947)

IPython/core/completer.py

@@ -1,3347 +1,3346 b''
 """Completion for IPython.
 
 This module started as fork of the rlcompleter module in the Python standard
 library.  The original enhancements made to rlcompleter have been sent
 upstream and were accepted as of Python 2.3,
 
 This module now support a wide variety of completion mechanism both available
 for normal classic Python code, as well as completer for IPython specific
 Syntax like magics.
 
 Latex and Unicode completion
 ============================
 
 IPython and compatible frontends not only can complete your code, but can help
 you to input a wide range of characters. In particular we allow you to insert
 a unicode character using the tab completion mechanism.
 
 Forward latex/unicode completion
 --------------------------------
 
 Forward completion allows you to easily type a unicode character using its latex
 name, or unicode long description. To do so type a backslash follow by the
 relevant name and press tab:
 
 
 Using latex completion:
 
 .. code::
 
     \\alpha<tab>
     α
 
 or using unicode completion:
 
 
 .. code::
 
     \\GREEK SMALL LETTER ALPHA<tab>
     α
 
 
 Only valid Python identifiers will complete. Combining characters (like arrow or
 dots) are also available, unlike latex they need to be put after the their
 counterpart that is to say, ``F\\\\vec<tab>`` is correct, not ``\\\\vec<tab>F``.
 
 Some browsers are known to display combining characters incorrectly.
 
 Backward latex completion
 -------------------------
 
 It is sometime challenging to know how to type a character, if you are using
 IPython, or any compatible frontend you can prepend backslash to the character
 and press :kbd:`Tab` to expand it to its latex form.
 
 .. code::
 
     \\α<tab>
     \\alpha
 
 
 Both forward and backward completions can be deactivated by setting the
 :std:configtrait:`Completer.backslash_combining_completions` option to
 ``False``.
 
 
 Experimental
 ============
 
 Starting with IPython 6.0, this module can make use of the Jedi library to
 generate completions both using static analysis of the code, and dynamically
 inspecting multiple namespaces. Jedi is an autocompletion and static analysis
 for Python. The APIs attached to this new mechanism is unstable and will
 raise unless use in an :any:`provisionalcompleter` context manager.
 
 You will find that the following are experimental:
 
     - :any:`provisionalcompleter`
     - :any:`IPCompleter.completions`
     - :any:`Completion`
     - :any:`rectify_completions`
 
 .. note::
 
     better name for :any:`rectify_completions` ?
 
 We welcome any feedback on these new API, and we also encourage you to try this
 module in debug mode (start IPython with ``--Completer.debug=True``) in order
 to have extra logging information if :any:`jedi` is crashing, or if current
 IPython completer pending deprecations are returning results not yet handled
 by :any:`jedi`
 
 Using Jedi for tab completion allow snippets like the following to work without
 having to execute any code:
 
    >>> myvar = ['hello', 42]
    ... myvar[1].bi<tab>
 
 Tab completion will be able to infer that ``myvar[1]`` is a real number without
 executing almost any code unlike the deprecated :any:`IPCompleter.greedy`
 option.
 
 Be sure to update :any:`jedi` to the latest stable version or to try the
 current development version to get better completions.
 
 Matchers
 ========
 
 All completions routines are implemented using unified *Matchers* API.
 The matchers API is provisional and subject to change without notice.
 
 The built-in matchers include:
 
 - :any:`IPCompleter.dict_key_matcher`:  dictionary key completions,
 - :any:`IPCompleter.magic_matcher`: completions for magics,
 - :any:`IPCompleter.unicode_name_matcher`,
   :any:`IPCompleter.fwd_unicode_matcher`
   and :any:`IPCompleter.latex_name_matcher`: see `Forward latex/unicode completion`_,
 - :any:`back_unicode_name_matcher` and :any:`back_latex_name_matcher`: see `Backward latex completion`_,
 - :any:`IPCompleter.file_matcher`: paths to files and directories,
 - :any:`IPCompleter.python_func_kw_matcher` - function keywords,
 - :any:`IPCompleter.python_matches` - globals and attributes (v1 API),
 - ``IPCompleter.jedi_matcher`` - static analysis with Jedi,
 - :any:`IPCompleter.custom_completer_matcher` - pluggable completer with a default
   implementation in :any:`InteractiveShell` which uses IPython hooks system
   (`complete_command`) with string dispatch (including regular expressions).
   Differently to other matchers, ``custom_completer_matcher`` will not suppress
   Jedi results to match behaviour in earlier IPython versions.
 
 Custom matchers can be added by appending to ``IPCompleter.custom_matchers`` list.
 
 Matcher API
 -----------
 
 Simplifying some details, the ``Matcher`` interface can described as
 
 .. code-block::
 
     MatcherAPIv1 = Callable[[str], list[str]]
     MatcherAPIv2 = Callable[[CompletionContext], SimpleMatcherResult]
 
     Matcher = MatcherAPIv1 | MatcherAPIv2
 
 The ``MatcherAPIv1`` reflects the matcher API as available prior to IPython 8.6.0
 and remains supported as a simplest way for generating completions. This is also
 currently the only API supported by the IPython hooks system `complete_command`.
 
 To distinguish between matcher versions ``matcher_api_version`` attribute is used.
 More precisely, the API allows to omit ``matcher_api_version`` for v1 Matchers,
 and requires a literal ``2`` for v2 Matchers.
 
 Once the API stabilises future versions may relax the requirement for specifying
 ``matcher_api_version`` by switching to :any:`functools.singledispatch`, therefore
 please do not rely on the presence of ``matcher_api_version`` for any purposes.
 
 Suppression of competing matchers
 ---------------------------------
 
 By default results from all matchers are combined, in the order determined by
 their priority. Matchers can request to suppress results from subsequent
 matchers by setting ``suppress`` to ``True`` in the ``MatcherResult``.
 
 When multiple matchers simultaneously request surpression, the results from of
 the matcher with higher priority will be returned.
 
 Sometimes it is desirable to suppress most but not all other matchers;
 this can be achieved by adding a list of identifiers of matchers which
 should not be suppressed to ``MatcherResult`` under ``do_not_suppress`` key.
 
 The suppression behaviour can is user-configurable via
 :std:configtrait:`IPCompleter.suppress_competing_matchers`.
 """
 
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 #
 # Some of this code originated from rlcompleter in the Python standard library
 # Copyright (C) 2001 Python Software Foundation, www.python.org
 
 from __future__ import annotations
 import builtins as builtin_mod
 import enum
 import glob
 import inspect
 import itertools
 import keyword
 import os
 import re
 import string
 import sys
 import tokenize
 import time
 import unicodedata
 import uuid
 import warnings
 from ast import literal_eval
 from collections import defaultdict
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import cached_property, partial
 from types import SimpleNamespace
 from typing import (
     Iterable,
     Iterator,
     List,
     Tuple,
     Union,
     Any,
     Sequence,
     Dict,
     Optional,
     TYPE_CHECKING,
     Set,
     Sized,
     TypeVar,
     Literal,
 )
 
 from IPython.core.guarded_eval import guarded_eval, EvaluationContext
 from IPython.core.error import TryNext
 from IPython.core.inputtransformer2 import ESC_MAGIC
 from IPython.core.latex_symbols import latex_symbols, reverse_latex_symbol
 from IPython.core.oinspect import InspectColors
 from IPython.testing.skipdoctest import skip_doctest
 from IPython.utils import generics
 from IPython.utils.decorators import sphinx_options
 from IPython.utils.dir2 import dir2, get_real_method
 from IPython.utils.docs import GENERATING_DOCUMENTATION
 from IPython.utils.path import ensure_dir_exists
 from IPython.utils.process import arg_split
 from traitlets import (
     Bool,
     Enum,
     Int,
     List as ListTrait,
     Unicode,
     Dict as DictTrait,
     Union as UnionTrait,
     observe,
 )
 from traitlets.config.configurable import Configurable
 
 import __main__
 
 # skip module docstests
 __skip_doctest__ = True
 
 
 try:
     import jedi
     jedi.settings.case_insensitive_completion = False
     import jedi.api.helpers
     import jedi.api.classes
     JEDI_INSTALLED = True
 except ImportError:
     JEDI_INSTALLED = False
 
 
 if TYPE_CHECKING or GENERATING_DOCUMENTATION and sys.version_info >= (3, 11):
     from typing import cast
     from typing_extensions import TypedDict, NotRequired, Protocol, TypeAlias, TypeGuard
 else:
     from typing import Generic
 
     def cast(type_, obj):
         """Workaround for `TypeError: MatcherAPIv2() takes no arguments`"""
         return obj
 
     # do not require on runtime
     NotRequired = Tuple  # requires Python >=3.11
     TypedDict = Dict  # by extension of `NotRequired` requires 3.11 too
     Protocol = object  # requires Python >=3.8
     TypeAlias = Any  # requires Python >=3.10
     TypeGuard = Generic  # requires Python >=3.10
 if GENERATING_DOCUMENTATION:
     from typing import TypedDict
 
 # -----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # ranges where we have most of the valid unicode names. We could be more finer
 # grained but is it worth it for performance  While unicode have character in the
 # range 0, 0x110000, we seem to have name for about 10% of those. (131808 as I
 # write this). With below range we cover them all, with a density of ~67%
 # biggest next gap we consider only adds up about 1% density and there are 600
 # gaps that would need hard coding.
 _UNICODE_RANGES = [(32, 0x323B0), (0xE0001, 0xE01F0)]
 
 # Public API
 __all__ = ["Completer", "IPCompleter"]
 
 if sys.platform == 'win32':
     PROTECTABLES = ' '
 else:
     PROTECTABLES = ' ()[]{}?=\\|;:\'#*"^&'
 
 # Protect against returning an enormous number of completions which the frontend
 # may have trouble processing.
 MATCHES_LIMIT = 500
 
 # Completion type reported when no type can be inferred.
 _UNKNOWN_TYPE = "<unknown>"
 
 # sentinel value to signal lack of a match
 not_found = object()
 
 class ProvisionalCompleterWarning(FutureWarning):
     """
     Exception raise by an experimental feature in this module.
 
     Wrap code in :any:`provisionalcompleter` context manager if you
     are certain you want to use an unstable feature.
     """
     pass
 
 warnings.filterwarnings('error', category=ProvisionalCompleterWarning)
 
 
 @skip_doctest
 @contextmanager
 def provisionalcompleter(action='ignore'):
     """
     This context manager has to be used in any place where unstable completer
     behavior and API may be called.
 
     >>> with provisionalcompleter():
     ...     completer.do_experimental_things() # works
 
     >>> completer.do_experimental_things() # raises.
 
     .. note::
 
         Unstable
 
         By using this context manager you agree that the API in use may change
         without warning, and that you won't complain if they do so.
 
         You also understand that, if the API is not to your liking, you should report
         a bug to explain your use case upstream.
 
         We'll be happy to get your feedback, feature requests, and improvements on
         any of the unstable APIs!
     """
     with warnings.catch_warnings():
         warnings.filterwarnings(action, category=ProvisionalCompleterWarning)
         yield
 
 
 def has_open_quotes(s):
     """Return whether a string has open quotes.
 
     This simply counts whether the number of quote characters of either type in
     the string is odd.
 
     Returns
     -------
     If there is an open quote, the quote character is returned.  Else, return
     False.
     """
     # We check " first, then ', so complex cases with nested quotes will get
     # the " to take precedence.
     if s.count('"') % 2:
         return '"'
     elif s.count("'") % 2:
         return "'"
     else:
         return False
 
 
 def protect_filename(s, protectables=PROTECTABLES):
     """Escape a string to protect certain characters."""
     if set(s) & set(protectables):
         if sys.platform == "win32":
             return '"' + s + '"'
         else:
             return "".join(("\\" + c if c in protectables else c) for c in s)
     else:
         return s
 
 
 def expand_user(path:str) -> Tuple[str, bool, str]:
     """Expand ``~``-style usernames in strings.
 
     This is similar to :func:`os.path.expanduser`, but it computes and returns
     extra information that will be useful if the input was being used in
     computing completions, and you wish to return the completions with the
     original '~' instead of its expanded value.
 
     Parameters
     ----------
     path : str
         String to be expanded.  If no ~ is present, the output is the same as the
         input.
 
     Returns
     -------
     newpath : str
         Result of ~ expansion in the input path.
     tilde_expand : bool
         Whether any expansion was performed or not.
     tilde_val : str
         The value that ~ was replaced with.
     """
     # Default values
     tilde_expand = False
     tilde_val = ''
     newpath = path
 
     if path.startswith('~'):
         tilde_expand = True
         rest = len(path)-1
         newpath = os.path.expanduser(path)
         if rest:
             tilde_val = newpath[:-rest]
         else:
             tilde_val = newpath
 
     return newpath, tilde_expand, tilde_val
 
 
 def compress_user(path:str, tilde_expand:bool, tilde_val:str) -> str:
     """Does the opposite of expand_user, with its outputs.
     """
     if tilde_expand:
         return path.replace(tilde_val, '~')
     else:
         return path
 
 
 def completions_sorting_key(word):
     """key for sorting completions
 
     This does several things:
 
     - Demote any completions starting with underscores to the end
     - Insert any %magic and %%cellmagic completions in the alphabetical order
       by their name
     """
     prio1, prio2 = 0, 0
 
     if word.startswith('__'):
         prio1 = 2
     elif word.startswith('_'):
         prio1 = 1
 
     if word.endswith('='):
         prio1 = -1
 
     if word.startswith('%%'):
         # If there's another % in there, this is something else, so leave it alone
         if not "%" in word[2:]:
             word = word[2:]
             prio2 = 2
     elif word.startswith('%'):
         if not "%" in word[1:]:
             word = word[1:]
             prio2 = 1
 
     return prio1, word, prio2
 
 
 class _FakeJediCompletion:
     """
     This is a workaround to communicate to the UI that Jedi has crashed and to
     report a bug. Will be used only id :any:`IPCompleter.debug` is set to true.
 
     Added in IPython 6.0 so should likely be removed for 7.0
 
     """
 
     def __init__(self, name):
 
         self.name = name
         self.complete = name
         self.type = 'crashed'
         self.name_with_symbols = name
         self.signature = ""
         self._origin = "fake"
         self.text = "crashed"
 
     def __repr__(self):
         return '<Fake completion object jedi has crashed>'
 
 
 _JediCompletionLike = Union["jedi.api.Completion", _FakeJediCompletion]
 
 
 class Completion:
     """
     Completion object used and returned by IPython completers.
 
     .. warning::
 
         Unstable
 
         This function is unstable, API may change without warning.
         It will also raise unless use in proper context manager.
 
     This act as a middle ground :any:`Completion` object between the
     :any:`jedi.api.classes.Completion` object and the Prompt Toolkit completion
     object. While Jedi need a lot of information about evaluator and how the
     code should be ran/inspected, PromptToolkit (and other frontend) mostly
     need user facing information.
 
     - Which range should be replaced replaced by what.
     - Some metadata (like completion type), or meta information to displayed to
       the use user.
 
     For debugging purpose we can also store the origin of the completion (``jedi``,
     ``IPython.python_matches``, ``IPython.magics_matches``...).
     """
 
     __slots__ = ['start', 'end', 'text', 'type', 'signature', '_origin']
 
     def __init__(
         self,
         start: int,
         end: int,
         text: str,
         *,
         type: Optional[str] = None,
         _origin="",
         signature="",
     ) -> None:
         warnings.warn(
             "``Completion`` is a provisional API (as of IPython 6.0). "
             "It may change without warnings. "
             "Use in corresponding context manager.",
             category=ProvisionalCompleterWarning,
             stacklevel=2,
         )
 
         self.start = start
         self.end = end
         self.text = text
         self.type = type
         self.signature = signature
         self._origin = _origin
 
     def __repr__(self):
         return '<Completion start=%s end=%s text=%r type=%r, signature=%r,>' % \
                 (self.start, self.end, self.text, self.type or '?', self.signature or '?')
 
     def __eq__(self, other) -> bool:
         """
         Equality and hash do not hash the type (as some completer may not be
         able to infer the type), but are use to (partially) de-duplicate
         completion.
 
         Completely de-duplicating completion is a bit tricker that just
         comparing as it depends on surrounding text, which Completions are not
         aware of.
         """
         return self.start == other.start and \
             self.end == other.end and \
             self.text == other.text
 
     def __hash__(self):
         return hash((self.start, self.end, self.text))
 
 
 class SimpleCompletion:
     """Completion item to be included in the dictionary returned by new-style Matcher (API v2).
 
     .. warning::
 
         Provisional
 
         This class is used to describe the currently supported attributes of
         simple completion items, and any additional implementation details
         should not be relied on. Additional attributes may be included in
         future versions, and meaning of text disambiguated from the current
         dual meaning of "text to insert" and "text to used as a label".
     """
 
     __slots__ = ["text", "type"]
 
     def __init__(self, text: str, *, type: Optional[str] = None):
         self.text = text
         self.type = type
 
     def __repr__(self):
         return f"<SimpleCompletion text={self.text!r} type={self.type!r}>"
 
 
 class _MatcherResultBase(TypedDict):
     """Definition of dictionary to be returned by new-style Matcher (API v2)."""
 
     #: Suffix of the provided ``CompletionContext.token``, if not given defaults to full token.
     matched_fragment: NotRequired[str]
 
     #: Whether to suppress results from all other matchers (True), some
     #: matchers (set of identifiers) or none (False); default is False.
     suppress: NotRequired[Union[bool, Set[str]]]
 
     #: Identifiers of matchers which should NOT be suppressed when this matcher
     #: requests to suppress all other matchers; defaults to an empty set.
     do_not_suppress: NotRequired[Set[str]]
 
     #: Are completions already ordered and should be left as-is? default is False.
     ordered: NotRequired[bool]
 
 
 @sphinx_options(show_inherited_members=True, exclude_inherited_from=["dict"])
 class SimpleMatcherResult(_MatcherResultBase, TypedDict):
     """Result of new-style completion matcher."""
 
     # note: TypedDict is added again to the inheritance chain
     # in order to get __orig_bases__ for documentation
 
     #: List of candidate completions
     completions: Sequence[SimpleCompletion] | Iterator[SimpleCompletion]
 
 
 class _JediMatcherResult(_MatcherResultBase):
     """Matching result returned by Jedi (will be processed differently)"""
 
     #: list of candidate completions
     completions: Iterator[_JediCompletionLike]
 
 
 AnyMatcherCompletion = Union[_JediCompletionLike, SimpleCompletion]
 AnyCompletion = TypeVar("AnyCompletion", AnyMatcherCompletion, Completion)
 
 
 @dataclass
 class CompletionContext:
     """Completion context provided as an argument to matchers in the Matcher API v2."""
 
     # rationale: many legacy matchers relied on completer state (`self.text_until_cursor`)
     # which was not explicitly visible as an argument of the matcher, making any refactor
     # prone to errors; by explicitly passing `cursor_position` we can decouple the matchers
     # from the completer, and make substituting them in sub-classes easier.
 
     #: Relevant fragment of code directly preceding the cursor.
     #: The extraction of token is implemented via splitter heuristic
     #: (following readline behaviour for legacy reasons), which is user configurable
     #: (by switching the greedy mode).
     token: str
 
     #: The full available content of the editor or buffer
     full_text: str
 
     #: Cursor position in the line (the same for ``full_text`` and ``text``).
     cursor_position: int
 
     #: Cursor line in ``full_text``.
     cursor_line: int
 
     #: The maximum number of completions that will be used downstream.
     #: Matchers can use this information to abort early.
     #: The built-in Jedi matcher is currently excepted from this limit.
     # If not given, return all possible completions.
     limit: Optional[int]
 
     @cached_property
     def text_until_cursor(self) -> str:
         return self.line_with_cursor[: self.cursor_position]
 
     @cached_property
     def line_with_cursor(self) -> str:
         return self.full_text.split("\n")[self.cursor_line]
 
 
 #: Matcher results for API v2.
 MatcherResult = Union[SimpleMatcherResult, _JediMatcherResult]
 
 
 class _MatcherAPIv1Base(Protocol):
     def __call__(self, text: str) -> List[str]:
         """Call signature."""
         ...
 
     #: Used to construct the default matcher identifier
     __qualname__: str
 
 
 class _MatcherAPIv1Total(_MatcherAPIv1Base, Protocol):
     #: API version
     matcher_api_version: Optional[Literal[1]]
 
     def __call__(self, text: str) -> List[str]:
         """Call signature."""
         ...
 
 
 #: Protocol describing Matcher API v1.
 MatcherAPIv1: TypeAlias = Union[_MatcherAPIv1Base, _MatcherAPIv1Total]
 
 
 class MatcherAPIv2(Protocol):
     """Protocol describing Matcher API v2."""
 
     #: API version
     matcher_api_version: Literal[2] = 2
 
     def __call__(self, context: CompletionContext) -> MatcherResult:
         """Call signature."""
         ...
 
     #: Used to construct the default matcher identifier
     __qualname__: str
 
 
 Matcher: TypeAlias = Union[MatcherAPIv1, MatcherAPIv2]
 
 
 def _is_matcher_v1(matcher: Matcher) -> TypeGuard[MatcherAPIv1]:
     api_version = _get_matcher_api_version(matcher)
     return api_version == 1
 
 
 def _is_matcher_v2(matcher: Matcher) -> TypeGuard[MatcherAPIv2]:
     api_version = _get_matcher_api_version(matcher)
     return api_version == 2
 
 
 def _is_sizable(value: Any) -> TypeGuard[Sized]:
     """Determines whether objects is sizable"""
     return hasattr(value, "__len__")
 
 
 def _is_iterator(value: Any) -> TypeGuard[Iterator]:
     """Determines whether objects is sizable"""
     return hasattr(value, "__next__")
 
 
 def has_any_completions(result: MatcherResult) -> bool:
     """Check if any result includes any completions."""
     completions = result["completions"]
     if _is_sizable(completions):
         return len(completions) != 0
     if _is_iterator(completions):
         try:
             old_iterator = completions
             first = next(old_iterator)
             result["completions"] = cast(
                 Iterator[SimpleCompletion],
                 itertools.chain([first], old_iterator),
             )
             return True
         except StopIteration:
             return False
     raise ValueError(
         "Completions returned by matcher need to be an Iterator or a Sizable"
     )
 
 
 def completion_matcher(
     *,
     priority: Optional[float] = None,
     identifier: Optional[str] = None,
     api_version: int = 1,
 ):
     """Adds attributes describing the matcher.
 
     Parameters
     ----------
     priority : Optional[float]
         The priority of the matcher, determines the order of execution of matchers.
         Higher priority means that the matcher will be executed first. Defaults to 0.
     identifier : Optional[str]
         identifier of the matcher allowing users to modify the behaviour via traitlets,
         and also used to for debugging (will be passed as ``origin`` with the completions).
 
         Defaults to matcher function's ``__qualname__`` (for example,
         ``IPCompleter.file_matcher`` for the built-in matched defined
         as a ``file_matcher`` method of the ``IPCompleter`` class).
     api_version: Optional[int]
         version of the Matcher API used by this matcher.
         Currently supported values are 1 and 2.
         Defaults to 1.
     """
 
     def wrapper(func: Matcher):
         func.matcher_priority = priority or 0  # type: ignore
         func.matcher_identifier = identifier or func.__qualname__  # type: ignore
         func.matcher_api_version = api_version  # type: ignore
         if TYPE_CHECKING:
             if api_version == 1:
                 func = cast(MatcherAPIv1, func)
             elif api_version == 2:
                 func = cast(MatcherAPIv2, func)
         return func
 
     return wrapper
 
 
 def _get_matcher_priority(matcher: Matcher):
     return getattr(matcher, "matcher_priority", 0)
 
 
 def _get_matcher_id(matcher: Matcher):
     return getattr(matcher, "matcher_identifier", matcher.__qualname__)
 
 
 def _get_matcher_api_version(matcher):
     return getattr(matcher, "matcher_api_version", 1)
 
 
 context_matcher = partial(completion_matcher, api_version=2)
 
 
 _IC = Iterable[Completion]
 
 
 def _deduplicate_completions(text: str, completions: _IC)-> _IC:
     """
     Deduplicate a set of completions.
 
     .. warning::
 
         Unstable
 
         This function is unstable, API may change without warning.
 
     Parameters
     ----------
     text : str
         text that should be completed.
     completions : Iterator[Completion]
         iterator over the completions to deduplicate
 
     Yields
     ------
     `Completions` objects
     Completions coming from multiple sources, may be different but end up having
     the same effect when applied to ``text``. If this is the case, this will
     consider completions as equal and only emit the first encountered.
     Not folded in `completions()` yet for debugging purpose, and to detect when
     the IPython completer does return things that Jedi does not, but should be
     at some point.
     """
     completions = list(completions)
     if not completions:
         return
 
     new_start = min(c.start for c in completions)
     new_end = max(c.end for c in completions)
 
     seen = set()
     for c in completions:
         new_text = text[new_start:c.start] + c.text + text[c.end:new_end]
         if new_text not in seen:
             yield c
             seen.add(new_text)
 
 
 def rectify_completions(text: str, completions: _IC, *, _debug: bool = False) -> _IC:
     """
     Rectify a set of completions to all have the same ``start`` and ``end``
 
     .. warning::
 
         Unstable
 
         This function is unstable, API may change without warning.
         It will also raise unless use in proper context manager.
 
     Parameters
     ----------
     text : str
         text that should be completed.
     completions : Iterator[Completion]
         iterator over the completions to rectify
     _debug : bool
         Log failed completion
 
     Notes
     -----
     :any:`jedi.api.classes.Completion` s returned by Jedi may not have the same start and end, though
     the Jupyter Protocol requires them to behave like so. This will readjust
     the completion to have the same ``start`` and ``end`` by padding both
     extremities with surrounding text.
 
     During stabilisation should support a ``_debug`` option to log which
     completion are return by the IPython completer and not found in Jedi in
     order to make upstream bug report.
     """
     warnings.warn("`rectify_completions` is a provisional API (as of IPython 6.0). "
                  "It may change without warnings. "
                  "Use in corresponding context manager.",
                   category=ProvisionalCompleterWarning, stacklevel=2)
 
     completions = list(completions)
     if not completions:
         return
     starts = (c.start for c in completions)
     ends = (c.end for c in completions)
 
     new_start = min(starts)
     new_end = max(ends)
 
     seen_jedi = set()
     seen_python_matches = set()
     for c in completions:
         new_text = text[new_start:c.start] + c.text + text[c.end:new_end]
         if c._origin == 'jedi':
             seen_jedi.add(new_text)
         elif c._origin == 'IPCompleter.python_matches':
             seen_python_matches.add(new_text)
         yield Completion(new_start, new_end, new_text, type=c.type, _origin=c._origin, signature=c.signature)
     diff = seen_python_matches.difference(seen_jedi)
     if diff and _debug:
         print('IPython.python matches have extras:', diff)
 
 
 if sys.platform == 'win32':
     DELIMS = ' \t\n`!@#$^&*()=+[{]}|;\'",<>?'
 else:
     DELIMS = ' \t\n`!@#$^&*()=+[{]}\\|;:\'",<>?'
 
 GREEDY_DELIMS = ' =\r\n'
 
 
 class CompletionSplitter(object):
     """An object to split an input line in a manner similar to readline.
 
     By having our own implementation, we can expose readline-like completion in
     a uniform manner to all frontends.  This object only needs to be given the
     line of text to be split and the cursor position on said line, and it
     returns the 'word' to be completed on at the cursor after splitting the
     entire line.
 
     What characters are used as splitting delimiters can be controlled by
     setting the ``delims`` attribute (this is a property that internally
     automatically builds the necessary regular expression)"""
 
     # Private interface
 
     # A string of delimiter characters.  The default value makes sense for
     # IPython's most typical usage patterns.
     _delims = DELIMS
 
     # The expression (a normal string) to be compiled into a regular expression
     # for actual splitting.  We store it as an attribute mostly for ease of
     # debugging, since this type of code can be so tricky to debug.
     _delim_expr = None
 
     # The regular expression that does the actual splitting
     _delim_re = None
 
     def __init__(self, delims=None):
         delims = CompletionSplitter._delims if delims is None else delims
         self.delims = delims
 
     @property
     def delims(self):
         """Return the string of delimiter characters."""
         return self._delims
 
     @delims.setter
     def delims(self, delims):
         """Set the delimiters for line splitting."""
         expr = '[' + ''.join('\\'+ c for c in delims) + ']'
         self._delim_re = re.compile(expr)
         self._delims = delims
         self._delim_expr = expr
 
     def split_line(self, line, cursor_pos=None):
         """Split a line of text with a cursor at the given position.
         """
         l = line if cursor_pos is None else line[:cursor_pos]
         return self._delim_re.split(l)[-1]
 
 
 
 class Completer(Configurable):
 
     greedy = Bool(
         False,
         help="""Activate greedy completion.
 
         .. deprecated:: 8.8
             Use :std:configtrait:`Completer.evaluation` and :std:configtrait:`Completer.auto_close_dict_keys` instead.
 
         When enabled in IPython 8.8 or newer, changes configuration as follows:
 
         - ``Completer.evaluation = 'unsafe'``
         - ``Completer.auto_close_dict_keys = True``
         """,
     ).tag(config=True)
 
     evaluation = Enum(
         ("forbidden", "minimal", "limited", "unsafe", "dangerous"),
         default_value="limited",
         help="""Policy for code evaluation under completion.
 
         Successive options allow to enable more eager evaluation for better
         completion suggestions, including for nested dictionaries, nested lists,
         or even results of function calls.
         Setting ``unsafe`` or higher can lead to evaluation of arbitrary user
         code on :kbd:`Tab` with potentially unwanted or dangerous side effects.
 
         Allowed values are:
 
         - ``forbidden``: no evaluation of code is permitted,
         - ``minimal``: evaluation of literals and access to built-in namespace;
           no item/attribute evaluationm no access to locals/globals,
           no evaluation of any operations or comparisons.
         - ``limited``: access to all namespaces, evaluation of hard-coded methods
           (for example: :any:`dict.keys`, :any:`object.__getattr__`,
           :any:`object.__getitem__`) on allow-listed objects (for example:
           :any:`dict`, :any:`list`, :any:`tuple`, ``pandas.Series``),
         - ``unsafe``: evaluation of all methods and function calls but not of
           syntax with side-effects like `del x`,
         - ``dangerous``: completely arbitrary evaluation.
         """,
     ).tag(config=True)
 
     use_jedi = Bool(default_value=JEDI_INSTALLED,
                     help="Experimental: Use Jedi to generate autocompletions. "
                     "Default to True if jedi is installed.").tag(config=True)
 
     jedi_compute_type_timeout = Int(default_value=400,
         help="""Experimental: restrict time (in milliseconds) during which Jedi can compute types.
         Set to 0 to stop computing types. Non-zero value lower than 100ms may hurt
         performance by preventing jedi to build its cache.
         """).tag(config=True)
 
     debug = Bool(default_value=False,
                  help='Enable debug for the Completer. Mostly print extra '
                       'information for experimental jedi integration.')\
                       .tag(config=True)
 
     backslash_combining_completions = Bool(True,
         help="Enable unicode completions, e.g. \\alpha<tab> . "
              "Includes completion of latex commands, unicode names, and expanding "
              "unicode characters back to latex commands.").tag(config=True)
 
     auto_close_dict_keys = Bool(
         False,
         help="""
         Enable auto-closing dictionary keys.
 
         When enabled string keys will be suffixed with a final quote
         (matching the opening quote), tuple keys will also receive a
         separating comma if needed, and keys which are final will
         receive a closing bracket (``]``).
         """,
     ).tag(config=True)
 
     def __init__(self, namespace=None, global_namespace=None, **kwargs):
         """Create a new completer for the command line.
 
         Completer(namespace=ns, global_namespace=ns2) -> completer instance.
 
         If unspecified, the default namespace where completions are performed
         is __main__ (technically, __main__.__dict__). Namespaces should be
         given as dictionaries.
 
         An optional second namespace can be given.  This allows the completer
         to handle cases where both the local and global scopes need to be
         distinguished.
         """
 
         # Don't bind to namespace quite yet, but flag whether the user wants a
         # specific namespace or to use __main__.__dict__. This will allow us
         # to bind to __main__.__dict__ at completion time, not now.
         if namespace is None:
             self.use_main_ns = True
         else:
             self.use_main_ns = False
             self.namespace = namespace
 
         # The global namespace, if given, can be bound directly
         if global_namespace is None:
             self.global_namespace = {}
         else:
             self.global_namespace = global_namespace
 
         self.custom_matchers = []
 
         super(Completer, self).__init__(**kwargs)
 
     def complete(self, text, state):
         """Return the next possible completion for 'text'.
 
         This is called successively with state == 0, 1, 2, ... until it
         returns None.  The completion should begin with 'text'.
 
         """
         if self.use_main_ns:
             self.namespace = __main__.__dict__
 
         if state == 0:
             if "." in text:
                 self.matches = self.attr_matches(text)
             else:
                 self.matches = self.global_matches(text)
         try:
             return self.matches[state]
         except IndexError:
             return None
 
     def global_matches(self, text):
         """Compute matches when text is a simple name.
 
         Return a list of all keywords, built-in functions and names currently
         defined in self.namespace or self.global_namespace that match.
 
         """
         matches = []
         match_append = matches.append
         n = len(text)
         for lst in [
             keyword.kwlist,
             builtin_mod.__dict__.keys(),
             list(self.namespace.keys()),
             list(self.global_namespace.keys()),
         ]:
             for word in lst:
                 if word[:n] == text and word != "__builtins__":
                     match_append(word)
 
         snake_case_re = re.compile(r"[^_]+(_[^_]+)+?\Z")
         for lst in [list(self.namespace.keys()), list(self.global_namespace.keys())]:
             shortened = {
                 "_".join([sub[0] for sub in word.split("_")]): word
                 for word in lst
                 if snake_case_re.match(word)
             }
             for word in shortened.keys():
                 if word[:n] == text and word != "__builtins__":
                     match_append(shortened[word])
         return matches
 
     def attr_matches(self, text):
         """Compute matches when text contains a dot.
 
         Assuming the text is of the form NAME.NAME....[NAME], and is
         evaluatable in self.namespace or self.global_namespace, it will be
         evaluated and its attributes (as revealed by dir()) are used as
         possible completions.  (For class instances, class members are
         also considered.)
 
         WARNING: this can still invoke arbitrary C code, if an object
         with a __getattr__ hook is evaluated.
 
         """
         m2 = re.match(r"(.+)\.(\w*)$", self.line_buffer)
         if not m2:
             return []
         expr, attr = m2.group(1, 2)
 
         obj = self._evaluate_expr(expr)
 
         if obj is not_found:
             return []
 
         if self.limit_to__all__ and hasattr(obj, '__all__'):
             words = get__all__entries(obj)
         else:
             words = dir2(obj)
 
         try:
             words = generics.complete_object(obj, words)
         except TryNext:
             pass
         except AssertionError:
             raise
         except Exception:
             # Silence errors from completion function
             pass
         # Build match list to return
         n = len(attr)
 
         # Note: ideally we would just return words here and the prefix
         # reconciliator would know that we intend to append to rather than
         # replace the input text; this requires refactoring to return range
         # which ought to be replaced (as does jedi).
         tokens = _parse_tokens(expr)
         rev_tokens = reversed(tokens)
         skip_over = {tokenize.ENDMARKER, tokenize.NEWLINE}
         name_turn = True
 
         parts = []
         for token in rev_tokens:
             if token.type in skip_over:
                 continue
             if token.type == tokenize.NAME and name_turn:
                 parts.append(token.string)
                 name_turn = False
             elif token.type == tokenize.OP and token.string == "." and not name_turn:
                 parts.append(token.string)
                 name_turn = True
             else:
                 # short-circuit if not empty nor name token
                 break
 
         prefix_after_space = "".join(reversed(parts))
 
         return ["%s.%s" % (prefix_after_space, w) for w in words if w[:n] == attr]
 
     def _evaluate_expr(self, expr):
         obj = not_found
         done = False
         while not done and expr:
             try:
                 obj = guarded_eval(
                     expr,
                     EvaluationContext(
                         globals=self.global_namespace,
                         locals=self.namespace,
                         evaluation=self.evaluation,
                     ),
                 )
                 done = True
             except Exception as e:
                 if self.debug:
                     print("Evaluation exception", e)
                 # trim the expression to remove any invalid prefix
                 # e.g. user starts `(d[`, so we get `expr = '(d'`,
                 # where parenthesis is not closed.
                 # TODO: make this faster by reusing parts of the computation?
                 expr = expr[1:]
         return obj
 
 def get__all__entries(obj):
     """returns the strings in the __all__ attribute"""
     try:
         words = getattr(obj, '__all__')
     except:
         return []
 
     return [w for w in words if isinstance(w, str)]
 
 
 class _DictKeyState(enum.Flag):
     """Represent state of the key match in context of other possible matches.
 
     - given `d1 = {'a': 1}` completion on `d1['<tab>` will yield `{'a': END_OF_ITEM}` as there is no tuple.
     - given `d2 = {('a', 'b'): 1}`: `d2['a', '<tab>` will yield `{'b': END_OF_TUPLE}` as there is no tuple members to add beyond `'b'`.
     - given `d3 = {('a', 'b'): 1}`: `d3['<tab>` will yield `{'a': IN_TUPLE}` as `'a'` can be added.
     - given `d4 = {'a': 1, ('a', 'b'): 2}`: `d4['<tab>` will yield `{'a': END_OF_ITEM & END_OF_TUPLE}`
     """
 
     BASELINE = 0
     END_OF_ITEM = enum.auto()
     END_OF_TUPLE = enum.auto()
     IN_TUPLE = enum.auto()
 
 
 def _parse_tokens(c):
     """Parse tokens even if there is an error."""
     tokens = []
     token_generator = tokenize.generate_tokens(iter(c.splitlines()).__next__)
     while True:
         try:
             tokens.append(next(token_generator))
         except tokenize.TokenError:
             return tokens
         except StopIteration:
             return tokens
 
 
 def _match_number_in_dict_key_prefix(prefix: str) -> Union[str, None]:
     """Match any valid Python numeric literal in a prefix of dictionary keys.
 
     References:
     - https://docs.python.org/3/reference/lexical_analysis.html#numeric-literals
     - https://docs.python.org/3/library/tokenize.html
     """
     if prefix[-1].isspace():
         # if user typed a space we do not have anything to complete
         # even if there was a valid number token before
         return None
     tokens = _parse_tokens(prefix)
     rev_tokens = reversed(tokens)
     skip_over = {tokenize.ENDMARKER, tokenize.NEWLINE}
     number = None
     for token in rev_tokens:
         if token.type in skip_over:
             continue
         if number is None:
             if token.type == tokenize.NUMBER:
                 number = token.string
                 continue
             else:
                 # we did not match a number
                 return None
         if token.type == tokenize.OP:
             if token.string == ",":
                 break
             if token.string in {"+", "-"}:
                 number = token.string + number
         else:
             return None
     return number
 
 
 _INT_FORMATS = {
     "0b": bin,
     "0o": oct,
     "0x": hex,
 }
 
 
 def match_dict_keys(
     keys: List[Union[str, bytes, Tuple[Union[str, bytes], ...]]],
     prefix: str,
     delims: str,
     extra_prefix: Optional[Tuple[Union[str, bytes], ...]] = None,
 ) -> Tuple[str, int, Dict[str, _DictKeyState]]:
     """Used by dict_key_matches, matching the prefix to a list of keys
 
     Parameters
     ----------
     keys
         list of keys in dictionary currently being completed.
     prefix
         Part of the text already typed by the user. E.g. `mydict[b'fo`
     delims
         String of delimiters to consider when finding the current key.
     extra_prefix : optional
         Part of the text already typed in multi-key index cases. E.g. for
         `mydict['foo', "bar", 'b`, this would be `('foo', 'bar')`.
 
     Returns
     -------
     A tuple of three elements: ``quote``, ``token_start``, ``matched``, with
     ``quote`` being the quote that need to be used to close current string.
     ``token_start`` the position where the replacement should start occurring,
     ``matches`` a dictionary of replacement/completion keys on keys and values
         indicating whether the state.
     """
     prefix_tuple = extra_prefix if extra_prefix else ()
 
     prefix_tuple_size = sum(
         [
             # for pandas, do not count slices as taking space
             not isinstance(k, slice)
             for k in prefix_tuple
         ]
     )
     text_serializable_types = (str, bytes, int, float, slice)
 
     def filter_prefix_tuple(key):
         # Reject too short keys
         if len(key) <= prefix_tuple_size:
             return False
         # Reject keys which cannot be serialised to text
         for k in key:
             if not isinstance(k, text_serializable_types):
                 return False
         # Reject keys that do not match the prefix
         for k, pt in zip(key, prefix_tuple):
             if k != pt and not isinstance(pt, slice):
                 return False
         # All checks passed!
         return True
 
     filtered_key_is_final: Dict[
         Union[str, bytes, int, float], _DictKeyState
     ] = defaultdict(lambda: _DictKeyState.BASELINE)
 
     for k in keys:
         # If at least one of the matches is not final, mark as undetermined.
         # This can happen with `d = {111: 'b', (111, 222): 'a'}` where
         # `111` appears final on first match but is not final on the second.
 
         if isinstance(k, tuple):
             if filter_prefix_tuple(k):
                 key_fragment = k[prefix_tuple_size]
                 filtered_key_is_final[key_fragment] |= (
                     _DictKeyState.END_OF_TUPLE
                     if len(k) == prefix_tuple_size + 1
                     else _DictKeyState.IN_TUPLE
                 )
         elif prefix_tuple_size > 0:
             # we are completing a tuple but this key is not a tuple,
             # so we should ignore it
             pass
         else:
             if isinstance(k, text_serializable_types):
                 filtered_key_is_final[k] |= _DictKeyState.END_OF_ITEM
 
     filtered_keys = filtered_key_is_final.keys()
 
     if not prefix:
         return "", 0, {repr(k): v for k, v in filtered_key_is_final.items()}
 
     quote_match = re.search("(?:\"|')", prefix)
     is_user_prefix_numeric = False
 
     if quote_match:
         quote = quote_match.group()
         valid_prefix = prefix + quote
         try:
             prefix_str = literal_eval(valid_prefix)
         except Exception:
             return "", 0, {}
     else:
         # If it does not look like a string, let's assume
         # we are dealing with a number or variable.
         number_match = _match_number_in_dict_key_prefix(prefix)
 
         # We do not want the key matcher to suggest variable names so we yield:
         if number_match is None:
             # The alternative would be to assume that user forgort the quote
             # and if the substring matches, suggest adding it at the start.
             return "", 0, {}
 
         prefix_str = number_match
         is_user_prefix_numeric = True
         quote = ""
 
     pattern = '[^' + ''.join('\\' + c for c in delims) + ']*$'
     token_match = re.search(pattern, prefix, re.UNICODE)
     assert token_match is not None # silence mypy
     token_start = token_match.start()
     token_prefix = token_match.group()
 
     matched: Dict[str, _DictKeyState] = {}
 
     str_key: Union[str, bytes]
 
     for key in filtered_keys:
         if isinstance(key, (int, float)):
             # User typed a number but this key is not a number.
             if not is_user_prefix_numeric:
                 continue
             str_key = str(key)
             if isinstance(key, int):
                 int_base = prefix_str[:2].lower()
                 # if user typed integer using binary/oct/hex notation:
                 if int_base in _INT_FORMATS:
                     int_format = _INT_FORMATS[int_base]
                     str_key = int_format(key)
         else:
             # User typed a string but this key is a number.
             if is_user_prefix_numeric:
                 continue
             str_key = key
         try:
             if not str_key.startswith(prefix_str):
                 continue
         except (AttributeError, TypeError, UnicodeError) as e:
             # Python 3+ TypeError on b'a'.startswith('a') or vice-versa
             continue
 
         # reformat remainder of key to begin with prefix
         rem = str_key[len(prefix_str) :]
         # force repr wrapped in '
         rem_repr = repr(rem + '"') if isinstance(rem, str) else repr(rem + b'"')
         rem_repr = rem_repr[1 + rem_repr.index("'"):-2]
         if quote == '"':
             # The entered prefix is quoted with ",
             # but the match is quoted with '.
             # A contained " hence needs escaping for comparison:
             rem_repr = rem_repr.replace('"', '\\"')
 
         # then reinsert prefix from start of token
         match = "%s%s" % (token_prefix, rem_repr)
 
         matched[match] = filtered_key_is_final[key]
     return quote, token_start, matched
 
 
 def cursor_to_position(text:str, line:int, column:int)->int:
     """
     Convert the (line,column) position of the cursor in text to an offset in a
     string.
 
     Parameters
     ----------
     text : str
         The text in which to calculate the cursor offset
     line : int
         Line of the cursor; 0-indexed
     column : int
         Column of the cursor 0-indexed
 
     Returns
     -------
     Position of the cursor in ``text``, 0-indexed.
 
     See Also
     --------
     position_to_cursor : reciprocal of this function
 
     """
     lines = text.split('\n')
     assert line <= len(lines), '{} <= {}'.format(str(line), str(len(lines)))
 
     return sum(len(l) + 1 for l in lines[:line]) + column
 
 def position_to_cursor(text:str, offset:int)->Tuple[int, int]:
     """
     Convert the position of the cursor in text (0 indexed) to a line
     number(0-indexed) and a column number (0-indexed) pair
 
     Position should be a valid position in ``text``.
 
     Parameters
     ----------
     text : str
         The text in which to calculate the cursor offset
     offset : int
         Position of the cursor in ``text``, 0-indexed.
 
     Returns
     -------
     (line, column) : (int, int)
         Line of the cursor; 0-indexed, column of the cursor 0-indexed
 
     See Also
     --------
     cursor_to_position : reciprocal of this function
 
     """
 
     assert 0 <= offset <= len(text) , "0 <= %s <= %s" % (offset , len(text))
 
     before = text[:offset]
     blines = before.split('\n')  # ! splitnes trim trailing \n
     line = before.count('\n')
     col = len(blines[-1])
     return line, col
 
 
 def _safe_isinstance(obj, module, class_name, *attrs):
     """Checks if obj is an instance of module.class_name if loaded
     """
     if module in sys.modules:
         m = sys.modules[module]
         for attr in [class_name, *attrs]:
             m = getattr(m, attr)
         return isinstance(obj, m)
 
 
 @context_matcher()
 def back_unicode_name_matcher(context: CompletionContext):
     """Match Unicode characters back to Unicode name
 
     Same as :any:`back_unicode_name_matches`, but adopted to new Matcher API.
     """
     fragment, matches = back_unicode_name_matches(context.text_until_cursor)
     return _convert_matcher_v1_result_to_v2(
         matches, type="unicode", fragment=fragment, suppress_if_matches=True
     )
 
 
 def back_unicode_name_matches(text: str) -> Tuple[str, Sequence[str]]:
     """Match Unicode characters back to Unicode name
 
     This does  ``☃`` -> ``\\snowman``
 
     Note that snowman is not a valid python3 combining character but will be expanded.
     Though it will not recombine back to the snowman character by the completion machinery.
 
     This will not either back-complete standard sequences like \\n, \\b ...
 
     .. deprecated:: 8.6
         You can use :meth:`back_unicode_name_matcher` instead.
 
     Returns
     =======
 
     Return a tuple with two elements:
 
     - The Unicode character that was matched (preceded with a backslash), or
         empty string,
     - a sequence (of 1), name for the match Unicode character, preceded by
         backslash, or empty if no match.
     """
     if len(text)<2:
         return '', ()
     maybe_slash = text[-2]
     if maybe_slash != '\\':
         return '', ()
 
     char = text[-1]
     # no expand on quote for completion in strings.
     # nor backcomplete standard ascii keys
     if char in string.ascii_letters or char in ('"',"'"):
         return '', ()
     try :
         unic = unicodedata.name(char)
         return '\\'+char,('\\'+unic,)
     except KeyError:
         pass
     return '', ()
 
 
 @context_matcher()
 def back_latex_name_matcher(context: CompletionContext):
     """Match latex characters back to unicode name
 
     Same as :any:`back_latex_name_matches`, but adopted to new Matcher API.
     """
     fragment, matches = back_latex_name_matches(context.text_until_cursor)
     return _convert_matcher_v1_result_to_v2(
         matches, type="latex", fragment=fragment, suppress_if_matches=True
     )
 
 
 def back_latex_name_matches(text: str) -> Tuple[str, Sequence[str]]:
     """Match latex characters back to unicode name
 
     This does ``\\ℵ`` -> ``\\aleph``
 
     .. deprecated:: 8.6
         You can use :meth:`back_latex_name_matcher` instead.
     """
     if len(text)<2:
         return '', ()
     maybe_slash = text[-2]
     if maybe_slash != '\\':
         return '', ()
 
 
     char = text[-1]
     # no expand on quote for completion in strings.
     # nor backcomplete standard ascii keys
     if char in string.ascii_letters or char in ('"',"'"):
         return '', ()
     try :
         latex = reverse_latex_symbol[char]
         # '\\' replace the \ as well
         return '\\'+char,[latex]
     except KeyError:
         pass
     return '', ()
 
 
 def _formatparamchildren(parameter) -> str:
     """
     Get parameter name and value from Jedi Private API
 
     Jedi does not expose a simple way to get `param=value` from its API.
 
     Parameters
     ----------
     parameter
         Jedi's function `Param`
 
     Returns
     -------
     A string like 'a', 'b=1', '*args', '**kwargs'
 
     """
     description = parameter.description
     if not description.startswith('param '):
         raise ValueError('Jedi function parameter description have change format.'
                          'Expected "param ...", found %r".' % description)
     return description[6:]
 
 def _make_signature(completion)-> str:
     """
     Make the signature from a jedi completion
 
     Parameters
     ----------
     completion : jedi.Completion
         object does not complete a function type
 
     Returns
     -------
     a string consisting of the function signature, with the parenthesis but
     without the function name. example:
     `(a, *args, b=1, **kwargs)`
 
     """
 
     # it looks like this might work on jedi 0.17
     if hasattr(completion, 'get_signatures'):
         signatures = completion.get_signatures()
         if not signatures:
             return  '(?)'
 
         c0 = completion.get_signatures()[0]
         return '('+c0.to_string().split('(', maxsplit=1)[1]
 
     return '(%s)'% ', '.join([f for f in (_formatparamchildren(p) for signature in completion.get_signatures()
                                           for p in signature.defined_names()) if f])
 
 
 _CompleteResult = Dict[str, MatcherResult]
 
 
 DICT_MATCHER_REGEX = re.compile(
     r"""(?x)
 (  # match dict-referring - or any get item object - expression
     .+
 )
 \[   # open bracket
 \s*  # and optional whitespace
 # Capture any number of serializable objects (e.g. "a", "b", 'c')
 # and slices
 ((?:(?:
     (?: # closed string
         [uUbB]?  # string prefix (r not handled)
         (?:
             '(?:[^']|(?<!\\)\\')*'
         |
             "(?:[^"]|(?<!\\)\\")*"
         )
     )
     |
         # capture integers and slices
         (?:[-+]?\d+)?(?::(?:[-+]?\d+)?){0,2}
     |
         # integer in bin/hex/oct notation
         0[bBxXoO]_?(?:\w|\d)+
     )
     \s*,\s*
 )*)
 ((?:
     (?: # unclosed string
         [uUbB]?  # string prefix (r not handled)
         (?:
             '(?:[^']|(?<!\\)\\')*
             |
             "(?:[^"]|(?<!\\)\\")*
         )
     )
     |
         # unfinished integer
         (?:[-+]?\d+)
     |
         # integer in bin/hex/oct notation
         0[bBxXoO]_?(?:\w|\d)+
     )
 )?
 $
 """
 )
 
 
 def _convert_matcher_v1_result_to_v2(
     matches: Sequence[str],
     type: str,
     fragment: Optional[str] = None,
     suppress_if_matches: bool = False,
 ) -> SimpleMatcherResult:
     """Utility to help with transition"""
     result = {
         "completions": [SimpleCompletion(text=match, type=type) for match in matches],
         "suppress": (True if matches else False) if suppress_if_matches else False,
     }
     if fragment is not None:
         result["matched_fragment"] = fragment
     return cast(SimpleMatcherResult, result)
 
 
 class IPCompleter(Completer):
     """Extension of the completer class with IPython-specific features"""
 
     @observe('greedy')
     def _greedy_changed(self, change):
         """update the splitter and readline delims when greedy is changed"""
         if change["new"]:
             self.evaluation = "unsafe"
             self.auto_close_dict_keys = True
             self.splitter.delims = GREEDY_DELIMS
         else:
             self.evaluation = "limited"
             self.auto_close_dict_keys = False
             self.splitter.delims = DELIMS
 
     dict_keys_only = Bool(
         False,
         help="""
         Whether to show dict key matches only.
 
         (disables all matchers except for `IPCompleter.dict_key_matcher`).
         """,
     )
 
     suppress_competing_matchers = UnionTrait(
         [Bool(allow_none=True), DictTrait(Bool(None, allow_none=True))],
         default_value=None,
         help="""
         Whether to suppress completions from other *Matchers*.
 
         When set to ``None`` (default) the matchers will attempt to auto-detect
         whether suppression of other matchers is desirable. For example, at
         the beginning of a line followed by `%` we expect a magic completion
         to be the only applicable option, and after ``my_dict['`` we usually
         expect a completion with an existing dictionary key.
 
         If you want to disable this heuristic and see completions from all matchers,
         set ``IPCompleter.suppress_competing_matchers = False``.
         To disable the heuristic for specific matchers provide a dictionary mapping:
         ``IPCompleter.suppress_competing_matchers = {'IPCompleter.dict_key_matcher': False}``.
 
         Set ``IPCompleter.suppress_competing_matchers = True`` to limit
         completions to the set of matchers with the highest priority;
         this is equivalent to ``IPCompleter.merge_completions`` and
         can be beneficial for performance, but will sometimes omit relevant
         candidates from matchers further down the priority list.
         """,
     ).tag(config=True)
 
     merge_completions = Bool(
         True,
         help="""Whether to merge completion results into a single list
 
         If False, only the completion results from the first non-empty
         completer will be returned.
 
         As of version 8.6.0, setting the value to ``False`` is an alias for:
         ``IPCompleter.suppress_competing_matchers = True.``.
         """,
     ).tag(config=True)
 
     disable_matchers = ListTrait(
         Unicode(),
         help="""List of matchers to disable.
 
         The list should contain matcher identifiers (see :any:`completion_matcher`).
         """,
     ).tag(config=True)
 
     omit__names = Enum(
         (0, 1, 2),
         default_value=2,
         help="""Instruct the completer to omit private method names
 
         Specifically, when completing on ``object.<tab>``.
 
         When 2 [default]: all names that start with '_' will be excluded.
 
         When 1: all 'magic' names (``__foo__``) will be excluded.
 
         When 0: nothing will be excluded.
         """
     ).tag(config=True)
     limit_to__all__ = Bool(False,
         help="""
         DEPRECATED as of version 5.0.
 
         Instruct the completer to use __all__ for the completion
 
         Specifically, when completing on ``object.<tab>``.
 
         When True: only those names in obj.__all__ will be included.
 
         When False [default]: the __all__ attribute is ignored
         """,
     ).tag(config=True)
 
     profile_completions = Bool(
         default_value=False,
         help="If True, emit profiling data for completion subsystem using cProfile."
     ).tag(config=True)
 
     profiler_output_dir = Unicode(
         default_value=".completion_profiles",
         help="Template for path at which to output profile data for completions."
     ).tag(config=True)
 
     @observe('limit_to__all__')
     def _limit_to_all_changed(self, change):
         warnings.warn('`IPython.core.IPCompleter.limit_to__all__` configuration '
             'value has been deprecated since IPython 5.0, will be made to have '
             'no effects and then removed in future version of IPython.',
             UserWarning)
 
     def __init__(
         self, shell=None, namespace=None, global_namespace=None, config=None, **kwargs
     ):
         """IPCompleter() -> completer
 
         Return a completer object.
 
         Parameters
         ----------
         shell
             a pointer to the ipython shell itself.  This is needed
             because this completer knows about magic functions, and those can
             only be accessed via the ipython instance.
         namespace : dict, optional
             an optional dict where completions are performed.
         global_namespace : dict, optional
             secondary optional dict for completions, to
             handle cases (such as IPython embedded inside functions) where
             both Python scopes are visible.
         config : Config
             traitlet's config object
         **kwargs
             passed to super class unmodified.
         """
 
         self.magic_escape = ESC_MAGIC
         self.splitter = CompletionSplitter()
 
         # _greedy_changed() depends on splitter and readline being defined:
         super().__init__(
             namespace=namespace,
             global_namespace=global_namespace,
             config=config,
             **kwargs,
         )
 
         # List where completion matches will be stored
         self.matches = []
         self.shell = shell
         # Regexp to split filenames with spaces in them
         self.space_name_re = re.compile(r'([^\\] )')
         # Hold a local ref. to glob.glob for speed
         self.glob = glob.glob
 
         # Determine if we are running on 'dumb' terminals, like (X)Emacs
         # buffers, to avoid completion problems.
         term = os.environ.get('TERM','xterm')
         self.dumb_terminal = term in ['dumb','emacs']
 
         # Special handling of backslashes needed in win32 platforms
         if sys.platform == "win32":
             self.clean_glob = self._clean_glob_win32
         else:
             self.clean_glob = self._clean_glob
 
         #regexp to parse docstring for function signature
         self.docstring_sig_re = re.compile(r'^[\w|\s.]+\(([^)]*)\).*')
         self.docstring_kwd_re = re.compile(r'[\s|\[]*(\w+)(?:\s*=\s*.*)')
         #use this if positional argument name is also needed
         #= re.compile(r'[\s|\[]*(\w+)(?:\s*=?\s*.*)')
 
         self.magic_arg_matchers = [
             self.magic_config_matcher,
             self.magic_color_matcher,
         ]
 
         # This is set externally by InteractiveShell
         self.custom_completers = None
 
         # This is a list of names of unicode characters that can be completed
         # into their corresponding unicode value. The list is large, so we
         # lazily initialize it on first use. Consuming code should access this
         # attribute through the `@unicode_names` property.
         self._unicode_names = None
 
         self._backslash_combining_matchers = [
             self.latex_name_matcher,
             self.unicode_name_matcher,
             back_latex_name_matcher,
             back_unicode_name_matcher,
             self.fwd_unicode_matcher,
         ]
 
         if not self.backslash_combining_completions:
             for matcher in self._backslash_combining_matchers:
                 self.disable_matchers.append(_get_matcher_id(matcher))
 
         if not self.merge_completions:
             self.suppress_competing_matchers = True
 
     @property
     def matchers(self) -> List[Matcher]:
         """All active matcher routines for completion"""
         if self.dict_keys_only:
             return [self.dict_key_matcher]
 
         if self.use_jedi:
             return [
                 *self.custom_matchers,
                 *self._backslash_combining_matchers,
                 *self.magic_arg_matchers,
                 self.custom_completer_matcher,
                 self.magic_matcher,
                 self._jedi_matcher,
                 self.dict_key_matcher,
                 self.file_matcher,
             ]
         else:
             return [
                 *self.custom_matchers,
                 *self._backslash_combining_matchers,
                 *self.magic_arg_matchers,
                 self.custom_completer_matcher,
                 self.dict_key_matcher,
                 # TODO: convert python_matches to v2 API
                 self.magic_matcher,
                 self.python_matches,
                 self.file_matcher,
                 self.python_func_kw_matcher,
             ]
 
     def all_completions(self, text:str) -> List[str]:
         """
         Wrapper around the completion methods for the benefit of emacs.
         """
         prefix = text.rpartition('.')[0]
         with provisionalcompleter():
             return ['.'.join([prefix, c.text]) if prefix and self.use_jedi else c.text
                     for c in self.completions(text, len(text))]
 
         return self.complete(text)[1]
 
     def _clean_glob(self, text:str):
         return self.glob("%s*" % text)
 
     def _clean_glob_win32(self, text:str):
         return [f.replace("\\","/")
                 for f in self.glob("%s*" % text)]
 
     @context_matcher()
     def file_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Same as :any:`file_matches`, but adopted to new Matcher API."""
         matches = self.file_matches(context.token)
         # TODO: add a heuristic for suppressing (e.g. if it has OS-specific delimiter,
         #  starts with `/home/`, `C:\`, etc)
         return _convert_matcher_v1_result_to_v2(matches, type="path")
 
     def file_matches(self, text: str) -> List[str]:
         """Match filenames, expanding ~USER type strings.
 
         Most of the seemingly convoluted logic in this completer is an
         attempt to handle filenames with spaces in them.  And yet it's not
         quite perfect, because Python's readline doesn't expose all of the
         GNU readline details needed for this to be done correctly.
 
         For a filename with a space in it, the printed completions will be
         only the parts after what's already been typed (instead of the
         full completions, as is normally done).  I don't think with the
         current (as of Python 2.3) Python readline it's possible to do
         better.
 
         .. deprecated:: 8.6
             You can use :meth:`file_matcher` instead.
         """
 
         # chars that require escaping with backslash - i.e. chars
         # that readline treats incorrectly as delimiters, but we
         # don't want to treat as delimiters in filename matching
         # when escaped with backslash
         if text.startswith('!'):
             text = text[1:]
             text_prefix = u'!'
         else:
             text_prefix = u''
 
         text_until_cursor = self.text_until_cursor
         # track strings with open quotes
         open_quotes = has_open_quotes(text_until_cursor)
 
         if '(' in text_until_cursor or '[' in text_until_cursor:
             lsplit = text
         else:
             try:
                 # arg_split ~ shlex.split, but with unicode bugs fixed by us
                 lsplit = arg_split(text_until_cursor)[-1]
             except ValueError:
                 # typically an unmatched ", or backslash without escaped char.
                 if open_quotes:
                     lsplit = text_until_cursor.split(open_quotes)[-1]
                 else:
                     return []
             except IndexError:
                 # tab pressed on empty line
                 lsplit = ""
 
         if not open_quotes and lsplit != protect_filename(lsplit):
             # if protectables are found, do matching on the whole escaped name
             has_protectables = True
             text0,text = text,lsplit
         else:
             has_protectables = False
             text = os.path.expanduser(text)
 
         if text == "":
             return [text_prefix + protect_filename(f) for f in self.glob("*")]
 
         # Compute the matches from the filesystem
         if sys.platform == 'win32':
             m0 = self.clean_glob(text)
         else:
             m0 = self.clean_glob(text.replace('\\', ''))
 
         if has_protectables:
             # If we had protectables, we need to revert our changes to the
             # beginning of filename so that we don't double-write the part
             # of the filename we have so far
             len_lsplit = len(lsplit)
             matches = [text_prefix + text0 +
                        protect_filename(f[len_lsplit:]) for f in m0]
         else:
             if open_quotes:
                 # if we have a string with an open quote, we don't need to
                 # protect the names beyond the quote (and we _shouldn't_, as
                 # it would cause bugs when the filesystem call is made).
                 matches = m0 if sys.platform == "win32" else\
                     [protect_filename(f, open_quotes) for f in m0]
             else:
                 matches = [text_prefix +
                            protect_filename(f) for f in m0]
 
         # Mark directories in input list by appending '/' to their names.
         return [x+'/' if os.path.isdir(x) else x for x in matches]
 
     @context_matcher()
     def magic_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Match magics."""
         text = context.token
         matches = self.magic_matches(text)
         result = _convert_matcher_v1_result_to_v2(matches, type="magic")
         is_magic_prefix = len(text) > 0 and text[0] == "%"
         result["suppress"] = is_magic_prefix and bool(result["completions"])
         return result
 
     def magic_matches(self, text: str):
         """Match magics.
 
         .. deprecated:: 8.6
             You can use :meth:`magic_matcher` instead.
         """
         # Get all shell magics now rather than statically, so magics loaded at
         # runtime show up too.
         lsm = self.shell.magics_manager.lsmagic()
         line_magics = lsm['line']
         cell_magics = lsm['cell']
         pre = self.magic_escape
         pre2 = pre+pre
 
         explicit_magic = text.startswith(pre)
 
         # Completion logic:
         # - user gives %%: only do cell magics
         # - user gives %: do both line and cell magics
         # - no prefix: do both
         # In other words, line magics are skipped if the user gives %% explicitly
         #
         # We also exclude magics that match any currently visible names:
         # https://github.com/ipython/ipython/issues/4877, unless the user has
         # typed a %:
         # https://github.com/ipython/ipython/issues/10754
         bare_text = text.lstrip(pre)
         global_matches = self.global_matches(bare_text)
         if not explicit_magic:
             def matches(magic):
                 """
                 Filter magics, in particular remove magics that match
                 a name present in global namespace.
                 """
                 return ( magic.startswith(bare_text) and
                          magic not in global_matches )
         else:
             def matches(magic):
                 return magic.startswith(bare_text)
 
         comp = [ pre2+m for m in cell_magics if matches(m)]
         if not text.startswith(pre2):
             comp += [ pre+m for m in line_magics if matches(m)]
 
         return comp
 
     @context_matcher()
     def magic_config_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Match class names and attributes for %config magic."""
         # NOTE: uses `line_buffer` equivalent for compatibility
         matches = self.magic_config_matches(context.line_with_cursor)
         return _convert_matcher_v1_result_to_v2(matches, type="param")
 
     def magic_config_matches(self, text: str) -> List[str]:
         """Match class names and attributes for %config magic.
 
         .. deprecated:: 8.6
             You can use :meth:`magic_config_matcher` instead.
         """
         texts = text.strip().split()
 
         if len(texts) > 0 and (texts[0] == 'config' or texts[0] == '%config'):
             # get all configuration classes
             classes = sorted(set([ c for c in self.shell.configurables
                                    if c.__class__.class_traits(config=True)
                                    ]), key=lambda x: x.__class__.__name__)
             classnames = [ c.__class__.__name__ for c in classes ]
 
             # return all classnames if config or %config is given
             if len(texts) == 1:
                 return classnames
 
             # match classname
             classname_texts = texts[1].split('.')
             classname = classname_texts[0]
             classname_matches = [ c for c in classnames
                                   if c.startswith(classname) ]
 
             # return matched classes or the matched class with attributes
             if texts[1].find('.') < 0:
                 return classname_matches
             elif len(classname_matches) == 1 and \
                             classname_matches[0] == classname:
                 cls = classes[classnames.index(classname)].__class__
                 help = cls.class_get_help()
                 # strip leading '--' from cl-args:
                 help = re.sub(re.compile(r'^--', re.MULTILINE), '', help)
                 return [ attr.split('=')[0]
                          for attr in help.strip().splitlines()
                          if attr.startswith(texts[1]) ]
         return []
 
     @context_matcher()
     def magic_color_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Match color schemes for %colors magic."""
         # NOTE: uses `line_buffer` equivalent for compatibility
         matches = self.magic_color_matches(context.line_with_cursor)
         return _convert_matcher_v1_result_to_v2(matches, type="param")
 
     def magic_color_matches(self, text: str) -> List[str]:
         """Match color schemes for %colors magic.
 
         .. deprecated:: 8.6
             You can use :meth:`magic_color_matcher` instead.
         """
         texts = text.split()
         if text.endswith(' '):
             # .split() strips off the trailing whitespace. Add '' back
             # so that: '%colors ' -> ['%colors', '']
             texts.append('')
 
         if len(texts) == 2 and (texts[0] == 'colors' or texts[0] == '%colors'):
             prefix = texts[1]
             return [ color for color in InspectColors.keys()
                      if color.startswith(prefix) ]
         return []
 
     @context_matcher(identifier="IPCompleter.jedi_matcher")
     def _jedi_matcher(self, context: CompletionContext) -> _JediMatcherResult:
         matches = self._jedi_matches(
             cursor_column=context.cursor_position,
             cursor_line=context.cursor_line,
             text=context.full_text,
         )
         return {
             "completions": matches,
             # static analysis should not suppress other matchers
             "suppress": False,
         }
 
     def _jedi_matches(
         self, cursor_column: int, cursor_line: int, text: str
     ) -> Iterator[_JediCompletionLike]:
         """
         Return a list of :any:`jedi.api.Completion`s object from a ``text`` and
         cursor position.
 
         Parameters
         ----------
         cursor_column : int
             column position of the cursor in ``text``, 0-indexed.
         cursor_line : int
             line position of the cursor in ``text``, 0-indexed
         text : str
             text to complete
 
         Notes
         -----
         If ``IPCompleter.debug`` is ``True`` may return a :any:`_FakeJediCompletion`
         object containing a string with the Jedi debug information attached.
 
         .. deprecated:: 8.6
             You can use :meth:`_jedi_matcher` instead.
         """
         namespaces = [self.namespace]
         if self.global_namespace is not None:
             namespaces.append(self.global_namespace)
 
         completion_filter = lambda x:x
         offset = cursor_to_position(text, cursor_line, cursor_column)
         # filter output if we are completing for object members
         if offset:
             pre = text[offset-1]
             if pre == '.':
                 if self.omit__names == 2:
                     completion_filter = lambda c:not c.name.startswith('_')
                 elif self.omit__names == 1:
                     completion_filter = lambda c:not (c.name.startswith('__') and c.name.endswith('__'))
                 elif self.omit__names == 0:
                     completion_filter = lambda x:x
                 else:
                     raise ValueError("Don't understand self.omit__names == {}".format(self.omit__names))
 
         interpreter = jedi.Interpreter(text[:offset], namespaces)
         try_jedi = True
 
         try:
             # find the first token in the current tree -- if it is a ' or " then we are in a string
             completing_string = False
             try:
                 first_child = next(c for c in interpreter._get_module().tree_node.children if hasattr(c, 'value'))
             except StopIteration:
                 pass
             else:
                 # note the value may be ', ", or it may also be ''' or """, or
                 # in some cases, """what/you/typed..., but all of these are
                 # strings.
                 completing_string = len(first_child.value) > 0 and first_child.value[0] in {"'", '"'}
 
             # if we are in a string jedi is likely not the right candidate for
             # now. Skip it.
             try_jedi = not completing_string
         except Exception as e:
             # many of things can go wrong, we are using private API just don't crash.
             if self.debug:
                 print("Error detecting if completing a non-finished string :", e, '|')
 
         if not try_jedi:
             return iter([])
         try:
             return filter(completion_filter, interpreter.complete(column=cursor_column, line=cursor_line + 1))
         except Exception as e:
             if self.debug:
                 return iter(
                     [
                         _FakeJediCompletion(
                             'Oops Jedi has crashed, please report a bug with the following:\n"""\n%s\ns"""'
                             % (e)
                         )
                     ]
                 )
             else:
                 return iter([])
 
     @completion_matcher(api_version=1)
     def python_matches(self, text: str) -> Iterable[str]:
         """Match attributes or global python names"""
         if "." in text:
             try:
                 matches = self.attr_matches(text)
                 if text.endswith('.') and self.omit__names:
                     if self.omit__names == 1:
                         # true if txt is _not_ a __ name, false otherwise:
                         no__name = (lambda txt:
                                     re.match(r'.*\.__.*?__',txt) is None)
                     else:
                         # true if txt is _not_ a _ name, false otherwise:
                         no__name = (lambda txt:
                                     re.match(r'\._.*?',txt[txt.rindex('.'):]) is None)
                     matches = filter(no__name, matches)
             except NameError:
                 # catches <undefined attributes>.<tab>
                 matches = []
         else:
             matches = self.global_matches(text)
         return matches
 
     def _default_arguments_from_docstring(self, doc):
         """Parse the first line of docstring for call signature.
 
         Docstring should be of the form 'min(iterable[, key=func])\n'.
         It can also parse cython docstring of the form
         'Minuit.migrad(self, int ncall=10000, resume=True, int nsplit=1)'.
         """
         if doc is None:
             return []
 
         #care only the firstline
         line = doc.lstrip().splitlines()[0]
 
         #p = re.compile(r'^[\w|\s.]+\(([^)]*)\).*')
         #'min(iterable[, key=func])\n' -> 'iterable[, key=func]'
         sig = self.docstring_sig_re.search(line)
         if sig is None:
             return []
         # iterable[, key=func]' -> ['iterable[' ,' key=func]']
         sig = sig.groups()[0].split(',')
         ret = []
         for s in sig:
             #re.compile(r'[\s|\[]*(\w+)(?:\s*=\s*.*)')
             ret += self.docstring_kwd_re.findall(s)
         return ret
 
     def _default_arguments(self, obj):
         """Return the list of default arguments of obj if it is callable,
         or empty list otherwise."""
         call_obj = obj
         ret = []
         if inspect.isbuiltin(obj):
             pass
         elif not (inspect.isfunction(obj) or inspect.ismethod(obj)):
             if inspect.isclass(obj):
                 #for cython embedsignature=True the constructor docstring
                 #belongs to the object itself not __init__
                 ret += self._default_arguments_from_docstring(
                             getattr(obj, '__doc__', ''))
                 # for classes, check for __init__,__new__
                 call_obj = (getattr(obj, '__init__', None) or
                        getattr(obj, '__new__', None))
             # for all others, check if they are __call__able
             elif hasattr(obj, '__call__'):
                 call_obj = obj.__call__
         ret += self._default_arguments_from_docstring(
                  getattr(call_obj, '__doc__', ''))
 
         _keeps = (inspect.Parameter.KEYWORD_ONLY,
                   inspect.Parameter.POSITIONAL_OR_KEYWORD)
 
         try:
             sig = inspect.signature(obj)
             ret.extend(k for k, v in sig.parameters.items() if
                        v.kind in _keeps)
         except ValueError:
             pass
 
         return list(set(ret))
 
     @context_matcher()
     def python_func_kw_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Match named parameters (kwargs) of the last open function."""
         matches = self.python_func_kw_matches(context.token)
         return _convert_matcher_v1_result_to_v2(matches, type="param")
 
     def python_func_kw_matches(self, text):
         """Match named parameters (kwargs) of the last open function.
 
         .. deprecated:: 8.6
             You can use :meth:`python_func_kw_matcher` instead.
         """
 
         if "." in text: # a parameter cannot be dotted
             return []
         try: regexp = self.__funcParamsRegex
         except AttributeError:
             regexp = self.__funcParamsRegex = re.compile(r'''
                 '.*?(?<!\\)' |    # single quoted strings or
                 ".*?(?<!\\)" |    # double quoted strings or
                 \w+          |    # identifier
                 \S                # other characters
                 ''', re.VERBOSE | re.DOTALL)
         # 1. find the nearest identifier that comes before an unclosed
         # parenthesis before the cursor
         # e.g. for "foo (1+bar(x), pa<cursor>,a=1)", the candidate is "foo"
         tokens = regexp.findall(self.text_until_cursor)
         iterTokens = reversed(tokens); openPar = 0
 
         for token in iterTokens:
             if token == ')':
                 openPar -= 1
             elif token == '(':
                 openPar += 1
                 if openPar > 0:
                     # found the last unclosed parenthesis
                     break
         else:
             return []
         # 2. Concatenate dotted names ("foo.bar" for "foo.bar(x, pa" )
         ids = []
         isId = re.compile(r'\w+$').match
 
         while True:
             try:
                 ids.append(next(iterTokens))
                 if not isId(ids[-1]):
                     ids.pop(); break
                 if not next(iterTokens) == '.':
                     break
             except StopIteration:
                 break
 
         # Find all named arguments already assigned to, as to avoid suggesting
         # them again
         usedNamedArgs = set()
         par_level = -1
         for token, next_token in zip(tokens, tokens[1:]):
             if token == '(':
                 par_level += 1
             elif token == ')':
                 par_level -= 1
 
             if par_level != 0:
                 continue
 
             if next_token != '=':
                 continue
 
             usedNamedArgs.add(token)
 
         argMatches = []
         try:
             callableObj = '.'.join(ids[::-1])
             namedArgs = self._default_arguments(eval(callableObj,
                                                     self.namespace))
 
             # Remove used named arguments from the list, no need to show twice
             for namedArg in set(namedArgs) - usedNamedArgs:
                 if namedArg.startswith(text):
                     argMatches.append("%s=" %namedArg)
         except:
             pass
 
         return argMatches
 
     @staticmethod
     def _get_keys(obj: Any) -> List[Any]:
         # Objects can define their own completions by defining an
         # _ipy_key_completions_() method.
         method = get_real_method(obj, '_ipython_key_completions_')
         if method is not None:
             return method()
 
         # Special case some common in-memory dict-like types
         if isinstance(obj, dict) or _safe_isinstance(obj, "pandas", "DataFrame"):
             try:
                 return list(obj.keys())
             except Exception:
                 return []
         elif _safe_isinstance(obj, "pandas", "core", "indexing", "_LocIndexer"):
             try:
                 return list(obj.obj.keys())
             except Exception:
                 return []
         elif _safe_isinstance(obj, 'numpy', 'ndarray') or\
              _safe_isinstance(obj, 'numpy', 'void'):
             return obj.dtype.names or []
         return []
 
     @context_matcher()
     def dict_key_matcher(self, context: CompletionContext) -> SimpleMatcherResult:
         """Match string keys in a dictionary, after e.g. ``foo[``."""
         matches = self.dict_key_matches(context.token)
         return _convert_matcher_v1_result_to_v2(
             matches, type="dict key", suppress_if_matches=True
         )
 
     def dict_key_matches(self, text: str) -> List[str]:
         """Match string keys in a dictionary, after e.g. ``foo[``.
 
         .. deprecated:: 8.6
             You can use :meth:`dict_key_matcher` instead.
         """
 
         # Short-circuit on closed dictionary (regular expression would
         # not match anyway, but would take quite a while).
         if self.text_until_cursor.strip().endswith("]"):
             return []
 
         match = DICT_MATCHER_REGEX.search(self.text_until_cursor)
 
         if match is None:
             return []
 
         expr, prior_tuple_keys, key_prefix = match.groups()
 
         obj = self._evaluate_expr(expr)
 
         if obj is not_found:
             return []
 
         keys = self._get_keys(obj)
         if not keys:
             return keys
 
         tuple_prefix = guarded_eval(
             prior_tuple_keys,
             EvaluationContext(
                 globals=self.global_namespace,
                 locals=self.namespace,
                 evaluation=self.evaluation,
                 in_subscript=True,
             ),
         )
 
         closing_quote, token_offset, matches = match_dict_keys(
             keys, key_prefix, self.splitter.delims, extra_prefix=tuple_prefix
         )
         if not matches:
             return []
 
         # get the cursor position of
         # - the text being completed
         # - the start of the key text
         # - the start of the completion
         text_start = len(self.text_until_cursor) - len(text)
         if key_prefix:
             key_start = match.start(3)
             completion_start = key_start + token_offset
         else:
             key_start = completion_start = match.end()
 
         # grab the leading prefix, to make sure all completions start with `text`
         if text_start > key_start:
             leading = ''
         else:
             leading = text[text_start:completion_start]
 
         # append closing quote and bracket as appropriate
         # this is *not* appropriate if the opening quote or bracket is outside
         # the text given to this method, e.g. `d["""a\nt
         can_close_quote = False
         can_close_bracket = False
 
         continuation = self.line_buffer[len(self.text_until_cursor) :].strip()
 
         if continuation.startswith(closing_quote):
             # do not close if already closed, e.g. `d['a<tab>'`
             continuation = continuation[len(closing_quote) :]
         else:
             can_close_quote = True
 
         continuation = continuation.strip()
 
         # e.g. `pandas.DataFrame` has different tuple indexer behaviour,
         # handling it is out of scope, so let's avoid appending suffixes.
         has_known_tuple_handling = isinstance(obj, dict)
 
         can_close_bracket = (
             not continuation.startswith("]") and self.auto_close_dict_keys
         )
         can_close_tuple_item = (
             not continuation.startswith(",")
             and has_known_tuple_handling
             and self.auto_close_dict_keys
         )
         can_close_quote = can_close_quote and self.auto_close_dict_keys
 
         # fast path if closing qoute should be appended but not suffix is allowed
         if not can_close_quote and not can_close_bracket and closing_quote:
             return [leading + k for k in matches]
 
         results = []
 
         end_of_tuple_or_item = _DictKeyState.END_OF_TUPLE | _DictKeyState.END_OF_ITEM
 
         for k, state_flag in matches.items():
             result = leading + k
             if can_close_quote and closing_quote:
                 result += closing_quote
 
             if state_flag == end_of_tuple_or_item:
                 # We do not know which suffix to add,
                 # e.g. both tuple item and string
                 # match this item.
                 pass
 
             if state_flag in end_of_tuple_or_item and can_close_bracket:
                 result += "]"
             if state_flag == _DictKeyState.IN_TUPLE and can_close_tuple_item:
                 result += ", "
             results.append(result)
         return results
 
     @context_matcher()
     def unicode_name_matcher(self, context: CompletionContext):
         """Same as :any:`unicode_name_matches`, but adopted to new Matcher API."""
         fragment, matches = self.unicode_name_matches(context.text_until_cursor)
         return _convert_matcher_v1_result_to_v2(
             matches, type="unicode", fragment=fragment, suppress_if_matches=True
         )
 
     @staticmethod
     def unicode_name_matches(text: str) -> Tuple[str, List[str]]:
         """Match Latex-like syntax for unicode characters base
         on the name of the character.
 
         This does  ``\\GREEK SMALL LETTER ETA`` -> ``η``
 
         Works only on valid python 3 identifier, or on combining characters that
         will combine to form a valid identifier.
         """
         slashpos = text.rfind('\\')
         if slashpos > -1:
             s = text[slashpos+1:]
             try :
                 unic = unicodedata.lookup(s)
                 # allow combining chars
                 if ('a'+unic).isidentifier():
                     return '\\'+s,[unic]
             except KeyError:
                 pass
         return '', []
 
     @context_matcher()
     def latex_name_matcher(self, context: CompletionContext):
         """Match Latex syntax for unicode characters.
 
         This does both ``\\alp`` -> ``\\alpha`` and ``\\alpha`` -> ``α``
         """
         fragment, matches = self.latex_matches(context.text_until_cursor)
         return _convert_matcher_v1_result_to_v2(
             matches, type="latex", fragment=fragment, suppress_if_matches=True
         )
 
     def latex_matches(self, text: str) -> Tuple[str, Sequence[str]]:
         """Match Latex syntax for unicode characters.
 
         This does both ``\\alp`` -> ``\\alpha`` and ``\\alpha`` -> ``α``
 
         .. deprecated:: 8.6
             You can use :meth:`latex_name_matcher` instead.
         """
         slashpos = text.rfind('\\')
         if slashpos > -1:
             s = text[slashpos:]
             if s in latex_symbols:
                 # Try to complete a full latex symbol to unicode
                 # \\alpha -> α
                 return s, [latex_symbols[s]]
             else:
                 # If a user has partially typed a latex symbol, give them
                 # a full list of options \al -> [\aleph, \alpha]
                 matches = [k for k in latex_symbols if k.startswith(s)]
                 if matches:
                     return s, matches
         return '', ()
 
     @context_matcher()
     def custom_completer_matcher(self, context):
         """Dispatch custom completer.
 
         If a match is found, suppresses all other matchers except for Jedi.
         """
         matches = self.dispatch_custom_completer(context.token) or []
         result = _convert_matcher_v1_result_to_v2(
             matches, type=_UNKNOWN_TYPE, suppress_if_matches=True
         )
         result["ordered"] = True
         result["do_not_suppress"] = {_get_matcher_id(self._jedi_matcher)}
         return result
 
     def dispatch_custom_completer(self, text):
         """
         .. deprecated:: 8.6
             You can use :meth:`custom_completer_matcher` instead.
         """
         if not self.custom_completers:
             return
 
         line = self.line_buffer
         if not line.strip():
             return None
 
         # Create a little structure to pass all the relevant information about
         # the current completion to any custom completer.
         event = SimpleNamespace()
         event.line = line
         event.symbol = text
         cmd = line.split(None,1)[0]
         event.command = cmd
         event.text_until_cursor = self.text_until_cursor
 
         # for foo etc, try also to find completer for %foo
         if not cmd.startswith(self.magic_escape):
             try_magic = self.custom_completers.s_matches(
                 self.magic_escape + cmd)
         else:
             try_magic = []
 
         for c in itertools.chain(self.custom_completers.s_matches(cmd),
                  try_magic,
                  self.custom_completers.flat_matches(self.text_until_cursor)):
             try:
                 res = c(event)
                 if res:
                     # first, try case sensitive match
                     withcase = [r for r in res if r.startswith(text)]
                     if withcase:
                         return withcase
                     # if none, then case insensitive ones are ok too
                     text_low = text.lower()
                     return [r for r in res if r.lower().startswith(text_low)]
             except TryNext:
                 pass
             except KeyboardInterrupt:
                 """
                 If custom completer take too long,
                 let keyboard interrupt abort and return nothing.
                 """
                 break
 
         return None
 
     def completions(self, text: str, offset: int)->Iterator[Completion]:
         """
         Returns an iterator over the possible completions
 
         .. warning::
 
             Unstable
 
             This function is unstable, API may change without warning.
             It will also raise unless use in proper context manager.
 
         Parameters
         ----------
         text : str
             Full text of the current input, multi line string.
         offset : int
             Integer representing the position of the cursor in ``text``. Offset
             is 0-based indexed.
 
         Yields
         ------
         Completion
 
         Notes
         -----
         The cursor on a text can either be seen as being "in between"
         characters or "On" a character depending on the interface visible to
         the user. For consistency the cursor being on "in between" characters X
         and Y is equivalent to the cursor being "on" character Y, that is to say
         the character the cursor is on is considered as being after the cursor.
 
         Combining characters may span more that one position in the
         text.
 
         .. note::
 
             If ``IPCompleter.debug`` is :any:`True` will yield a ``--jedi/ipython--``
             fake Completion token to distinguish completion returned by Jedi
             and usual IPython completion.
 
         .. note::
 
             Completions are not completely deduplicated yet. If identical
             completions are coming from different sources this function does not
             ensure that each completion object will only be present once.
         """
         warnings.warn("_complete is a provisional API (as of IPython 6.0). "
                       "It may change without warnings. "
                       "Use in corresponding context manager.",
                       category=ProvisionalCompleterWarning, stacklevel=2)
 
         seen = set()
         profiler:Optional[cProfile.Profile]
         try:
             if self.profile_completions:
                 import cProfile
                 profiler = cProfile.Profile()
                 profiler.enable()
             else:
                 profiler = None
 
             for c in self._completions(text, offset, _timeout=self.jedi_compute_type_timeout/1000):
                 if c and (c in seen):
                     continue
                 yield c
                 seen.add(c)
         except KeyboardInterrupt:
             """if completions take too long and users send keyboard interrupt,
             do not crash and return ASAP. """
             pass
         finally:
             if profiler is not None:
                 profiler.disable()
                 ensure_dir_exists(self.profiler_output_dir)
                 output_path = os.path.join(self.profiler_output_dir, str(uuid.uuid4()))
                 print("Writing profiler output to", output_path)
                 profiler.dump_stats(output_path)
 
     def _completions(self, full_text: str, offset: int, *, _timeout) -> Iterator[Completion]:
         """
         Core completion module.Same signature as :any:`completions`, with the
         extra `timeout` parameter (in seconds).
 
         Computing jedi's completion ``.type`` can be quite expensive (it is a
         lazy property) and can require some warm-up, more warm up than just
         computing the ``name`` of a completion. The warm-up can be :
 
             - Long warm-up the first time a module is encountered after
             install/update: actually build parse/inference tree.
 
             - first time the module is encountered in a session: load tree from
             disk.
 
         We don't want to block completions for tens of seconds so we give the
         completer a "budget" of ``_timeout`` seconds per invocation to compute
         completions types, the completions that have not yet been computed will
         be marked as "unknown" an will have a chance to be computed next round
         are things get cached.
 
         Keep in mind that Jedi is not the only thing treating the completion so
         keep the timeout short-ish as if we take more than 0.3 second we still
         have lots of processing to do.
 
         """
         deadline = time.monotonic() + _timeout
 
         before = full_text[:offset]
         cursor_line, cursor_column = position_to_cursor(full_text, offset)
 
         jedi_matcher_id = _get_matcher_id(self._jedi_matcher)
 
         def is_non_jedi_result(
             result: MatcherResult, identifier: str
         ) -> TypeGuard[SimpleMatcherResult]:
             return identifier != jedi_matcher_id
 
         results = self._complete(
             full_text=full_text, cursor_line=cursor_line, cursor_pos=cursor_column
         )
 
         non_jedi_results: Dict[str, SimpleMatcherResult] = {
             identifier: result
             for identifier, result in results.items()
             if is_non_jedi_result(result, identifier)
         }
 
         jedi_matches = (
             cast(_JediMatcherResult, results[jedi_matcher_id])["completions"]
             if jedi_matcher_id in results
             else ()
         )
 
         iter_jm = iter(jedi_matches)
         if _timeout:
             for jm in iter_jm:
                 try:
                     type_ = jm.type
                 except Exception:
                     if self.debug:
                         print("Error in Jedi getting type of ", jm)
                     type_ = None
                 delta = len(jm.name_with_symbols) - len(jm.complete)
                 if type_ == 'function':
                     signature = _make_signature(jm)
                 else:
                     signature = ''
                 yield Completion(start=offset - delta,
                                  end=offset,
                                  text=jm.name_with_symbols,
                                  type=type_,
                                  signature=signature,
                                  _origin='jedi')
 
                 if time.monotonic() > deadline:
                     break
 
         for jm in iter_jm:
             delta = len(jm.name_with_symbols) - len(jm.complete)
             yield Completion(
                 start=offset - delta,
                 end=offset,
                 text=jm.name_with_symbols,
                 type=_UNKNOWN_TYPE,  # don't compute type for speed
                 _origin="jedi",
                 signature="",
             )
 
         # TODO:
         # Suppress this, right now just for debug.
         if jedi_matches and non_jedi_results and self.debug:
             some_start_offset = before.rfind(
                 next(iter(non_jedi_results.values()))["matched_fragment"]
             )
             yield Completion(
                 start=some_start_offset,
                 end=offset,
                 text="--jedi/ipython--",
                 _origin="debug",
                 type="none",
                 signature="",
             )
 
         ordered: List[Completion] = []
         sortable: List[Completion] = []
 
         for origin, result in non_jedi_results.items():
             matched_text = result["matched_fragment"]
             start_offset = before.rfind(matched_text)
             is_ordered = result.get("ordered", False)
             container = ordered if is_ordered else sortable
 
             # I'm unsure if this is always true, so let's assert and see if it
             # crash
             assert before.endswith(matched_text)
 
             for simple_completion in result["completions"]:
                 completion = Completion(
                     start=start_offset,
                     end=offset,
                     text=simple_completion.text,
                     _origin=origin,
                     signature="",
                     type=simple_completion.type or _UNKNOWN_TYPE,
                 )
                 container.append(completion)
 
         yield from list(self._deduplicate(ordered + self._sort(sortable)))[
             :MATCHES_LIMIT
         ]
 
     def complete(self, text=None, line_buffer=None, cursor_pos=None) -> Tuple[str, Sequence[str]]:
         """Find completions for the given text and line context.
 
         Note that both the text and the line_buffer are optional, but at least
         one of them must be given.
 
         Parameters
         ----------
         text : string, optional
             Text to perform the completion on.  If not given, the line buffer
             is split using the instance's CompletionSplitter object.
         line_buffer : string, optional
             If not given, the completer attempts to obtain the current line
             buffer via readline.  This keyword allows clients which are
             requesting for text completions in non-readline contexts to inform
             the completer of the entire text.
         cursor_pos : int, optional
             Index of the cursor in the full line buffer.  Should be provided by
             remote frontends where kernel has no access to frontend state.
 
         Returns
         -------
         Tuple of two items:
         text : str
             Text that was actually used in the completion.
         matches : list
             A list of completion matches.
 
         Notes
         -----
             This API is likely to be deprecated and replaced by
             :any:`IPCompleter.completions` in the future.
 
         """
         warnings.warn('`Completer.complete` is pending deprecation since '
                 'IPython 6.0 and will be replaced by `Completer.completions`.',
                       PendingDeprecationWarning)
         # potential todo, FOLD the 3rd throw away argument of _complete
         # into the first 2 one.
         # TODO: Q: does the above refer to jedi completions (i.e. 0-indexed?)
         # TODO: should we deprecate now, or does it stay?
 
         results = self._complete(
             line_buffer=line_buffer, cursor_pos=cursor_pos, text=text, cursor_line=0
         )
 
         jedi_matcher_id = _get_matcher_id(self._jedi_matcher)
 
         return self._arrange_and_extract(
             results,
             # TODO: can we confirm that excluding Jedi here was a deliberate choice in previous version?
             skip_matchers={jedi_matcher_id},
             # this API does not support different start/end positions (fragments of token).
             abort_if_offset_changes=True,
         )
 
     def _arrange_and_extract(
         self,
         results: Dict[str, MatcherResult],
         skip_matchers: Set[str],
         abort_if_offset_changes: bool,
     ):
-
         sortable: List[AnyMatcherCompletion] = []
         ordered: List[AnyMatcherCompletion] = []
         most_recent_fragment = None
         for identifier, result in results.items():
             if identifier in skip_matchers:
                 continue
             if not result["completions"]:
                 continue
             if not most_recent_fragment:
                 most_recent_fragment = result["matched_fragment"]
             if (
                 abort_if_offset_changes
                 and result["matched_fragment"] != most_recent_fragment
             ):
                 break
             if result.get("ordered", False):
                 ordered.extend(result["completions"])
             else:
                 sortable.extend(result["completions"])
 
         if not most_recent_fragment:
             most_recent_fragment = ""  # to satisfy typechecker (and just in case)
 
         return most_recent_fragment, [
             m.text for m in self._deduplicate(ordered + self._sort(sortable))
         ]
 
     def _complete(self, *, cursor_line, cursor_pos, line_buffer=None, text=None,
                   full_text=None) -> _CompleteResult:
         """
         Like complete but can also returns raw jedi completions as well as the
         origin of the completion text. This could (and should) be made much
         cleaner but that will be simpler once we drop the old (and stateful)
         :any:`complete` API.
 
         With current provisional API, cursor_pos act both (depending on the
         caller) as the offset in the ``text`` or ``line_buffer``, or as the
         ``column`` when passing multiline strings this could/should be renamed
         but would add extra noise.
 
         Parameters
         ----------
         cursor_line
             Index of the line the cursor is on. 0 indexed.
         cursor_pos
             Position of the cursor in the current line/line_buffer/text. 0
             indexed.
         line_buffer : optional, str
             The current line the cursor is in, this is mostly due to legacy
             reason that readline could only give a us the single current line.
             Prefer `full_text`.
         text : str
             The current "token" the cursor is in, mostly also for historical
             reasons. as the completer would trigger only after the current line
             was parsed.
         full_text : str
             Full text of the current cell.
 
         Returns
         -------
         An ordered dictionary where keys are identifiers of completion
         matchers and values are ``MatcherResult``s.
         """
 
         # if the cursor position isn't given, the only sane assumption we can
         # make is that it's at the end of the line (the common case)
         if cursor_pos is None:
             cursor_pos = len(line_buffer) if text is None else len(text)
 
         if self.use_main_ns:
             self.namespace = __main__.__dict__
 
         # if text is either None or an empty string, rely on the line buffer
         if (not line_buffer) and full_text:
             line_buffer = full_text.split('\n')[cursor_line]
         if not text:  # issue #11508: check line_buffer before calling split_line
             text = (
                 self.splitter.split_line(line_buffer, cursor_pos) if line_buffer else ""
             )
 
         # If no line buffer is given, assume the input text is all there was
         if line_buffer is None:
             line_buffer = text
 
         # deprecated - do not use `line_buffer` in new code.
         self.line_buffer = line_buffer
         self.text_until_cursor = self.line_buffer[:cursor_pos]
 
         if not full_text:
             full_text = line_buffer
 
         context = CompletionContext(
             full_text=full_text,
             cursor_position=cursor_pos,
             cursor_line=cursor_line,
             token=text,
             limit=MATCHES_LIMIT,
         )
 
         # Start with a clean slate of completions
         results: Dict[str, MatcherResult] = {}
 
         jedi_matcher_id = _get_matcher_id(self._jedi_matcher)
 
         suppressed_matchers: Set[str] = set()
 
         matchers = {
             _get_matcher_id(matcher): matcher
             for matcher in sorted(
                 self.matchers, key=_get_matcher_priority, reverse=True
             )
         }
 
         for matcher_id, matcher in matchers.items():
             matcher_id = _get_matcher_id(matcher)
 
             if matcher_id in self.disable_matchers:
                 continue
 
             if matcher_id in results:
                 warnings.warn(f"Duplicate matcher ID: {matcher_id}.")
 
             if matcher_id in suppressed_matchers:
                 continue
 
             result: MatcherResult
             try:
                 if _is_matcher_v1(matcher):
                     result = _convert_matcher_v1_result_to_v2(
                         matcher(text), type=_UNKNOWN_TYPE
                     )
                 elif _is_matcher_v2(matcher):
                     result = matcher(context)
                 else:
                     api_version = _get_matcher_api_version(matcher)
                     raise ValueError(f"Unsupported API version {api_version}")
             except:
                 # Show the ugly traceback if the matcher causes an
                 # exception, but do NOT crash the kernel!
                 sys.excepthook(*sys.exc_info())
                 continue
 
             # set default value for matched fragment if suffix was not selected.
             result["matched_fragment"] = result.get("matched_fragment", context.token)
 
             if not suppressed_matchers:
                 suppression_recommended: Union[bool, Set[str]] = result.get(
                     "suppress", False
                 )
 
                 suppression_config = (
                     self.suppress_competing_matchers.get(matcher_id, None)
                     if isinstance(self.suppress_competing_matchers, dict)
                     else self.suppress_competing_matchers
                 )
                 should_suppress = (
                     (suppression_config is True)
                     or (suppression_recommended and (suppression_config is not False))
                 ) and has_any_completions(result)
 
                 if should_suppress:
                     suppression_exceptions: Set[str] = result.get(
                         "do_not_suppress", set()
                     )
                     if isinstance(suppression_recommended, Iterable):
                         to_suppress = set(suppression_recommended)
                     else:
                         to_suppress = set(matchers)
                     suppressed_matchers = to_suppress - suppression_exceptions
 
                     new_results = {}
                     for previous_matcher_id, previous_result in results.items():
                         if previous_matcher_id not in suppressed_matchers:
                             new_results[previous_matcher_id] = previous_result
                     results = new_results
 
             results[matcher_id] = result
 
         _, matches = self._arrange_and_extract(
             results,
             # TODO Jedi completions non included in legacy stateful API; was this deliberate or omission?
             #  if it was omission, we can remove the filtering step, otherwise remove this comment.
             skip_matchers={jedi_matcher_id},
             abort_if_offset_changes=False,
         )
 
         # populate legacy stateful API
         self.matches = matches
 
         return results
 
     @staticmethod
     def _deduplicate(
         matches: Sequence[AnyCompletion],
     ) -> Iterable[AnyCompletion]:
         filtered_matches: Dict[str, AnyCompletion] = {}
         for match in matches:
             text = match.text
             if (
                 text not in filtered_matches
                 or filtered_matches[text].type == _UNKNOWN_TYPE
             ):
                 filtered_matches[text] = match
 
         return filtered_matches.values()
 
     @staticmethod
     def _sort(matches: Sequence[AnyCompletion]):
         return sorted(matches, key=lambda x: completions_sorting_key(x.text))
 
     @context_matcher()
     def fwd_unicode_matcher(self, context: CompletionContext):
         """Same as :any:`fwd_unicode_match`, but adopted to new Matcher API."""
         # TODO: use `context.limit` to terminate early once we matched the maximum
         #  number that will be used downstream; can be added as an optional to
         #  `fwd_unicode_match(text: str, limit: int = None)` or we could re-implement here.
         fragment, matches = self.fwd_unicode_match(context.text_until_cursor)
         return _convert_matcher_v1_result_to_v2(
             matches, type="unicode", fragment=fragment, suppress_if_matches=True
         )
 
     def fwd_unicode_match(self, text: str) -> Tuple[str, Sequence[str]]:
         """
         Forward match a string starting with a backslash with a list of
         potential Unicode completions.
 
         Will compute list of Unicode character names on first call and cache it.
 
         .. deprecated:: 8.6
             You can use :meth:`fwd_unicode_matcher` instead.
 
         Returns
         -------
         At tuple with:
             - matched text (empty if no matches)
             - list of potential completions, empty tuple  otherwise)
         """
         # TODO: self.unicode_names is here a list we traverse each time with ~100k elements.
         # We could do a faster match using a Trie.
 
         # Using pygtrie the following seem to work:
 
         #     s = PrefixSet()
 
         #     for c in range(0,0x10FFFF + 1):
         #         try:
         #             s.add(unicodedata.name(chr(c)))
         #         except ValueError:
         #             pass
         #     [''.join(k) for k in s.iter(prefix)]
 
         # But need to be timed and adds an extra dependency.
 
         slashpos = text.rfind('\\')
         # if text starts with slash
         if slashpos > -1:
             # PERF: It's important that we don't access self._unicode_names
             # until we're inside this if-block. _unicode_names is lazily
             # initialized, and it takes a user-noticeable amount of time to
             # initialize it, so we don't want to initialize it unless we're
             # actually going to use it.
             s = text[slashpos + 1 :]
             sup = s.upper()
             candidates = [x for x in self.unicode_names if x.startswith(sup)]
             if candidates:
                 return s, candidates
             candidates = [x for x in self.unicode_names if sup in x]
             if candidates:
                 return s, candidates
             splitsup = sup.split(" ")
             candidates = [
                 x for x in self.unicode_names if all(u in x for u in splitsup)
             ]
             if candidates:
                 return s, candidates
 
             return "", ()
 
         # if text does not start with slash
         else:
             return '', ()
 
     @property
     def unicode_names(self) -> List[str]:
         """List of names of unicode code points that can be completed.
 
         The list is lazily initialized on first access.
         """
         if self._unicode_names is None:
             names = []
             for c in range(0,0x10FFFF + 1):
                 try:
                     names.append(unicodedata.name(chr(c)))
                 except ValueError:
                     pass
             self._unicode_names = _unicode_name_compute(_UNICODE_RANGES)
 
         return self._unicode_names
 
 def _unicode_name_compute(ranges:List[Tuple[int,int]]) -> List[str]:
     names = []
     for start,stop in ranges:
         for c in range(start, stop) :
             try:
                 names.append(unicodedata.name(chr(c)))
             except ValueError:
                 pass
     return names

IPython/core/debugger.py

@@ -1,998 +1,997 b''
 # -*- coding: utf-8 -*-
 """
 Pdb debugger class.
 
 
 This is an extension to PDB which adds a number of new features.
 Note that there is also the `IPython.terminal.debugger` class which provides UI
 improvements.
 
 We also strongly recommend to use this via the `ipdb` package, which provides
 extra configuration options.
 
 Among other things, this subclass of PDB:
  - supports many IPython magics like pdef/psource
  - hide frames in tracebacks based on `__tracebackhide__`
  - allows to skip frames based on `__debuggerskip__`
 
 The skipping and hiding frames are configurable via the `skip_predicates`
 command.
 
 By default, frames from readonly files will be hidden, frames containing
 ``__tracebackhide__=True`` will be hidden.
 
 Frames containing ``__debuggerskip__`` will be stepped over, frames who's parent
 frames value of ``__debuggerskip__`` is ``True`` will be skipped.
 
     >>> def helpers_helper():
     ...     pass
     ...
     ... def helper_1():
     ...     print("don't step in me")
     ...     helpers_helpers() # will be stepped over unless breakpoint set.
     ...
     ...
     ... def helper_2():
     ...     print("in me neither")
     ...
 
 One can define a decorator that wraps a function between the two helpers:
 
     >>> def pdb_skipped_decorator(function):
     ...
     ...
     ...     def wrapped_fn(*args, **kwargs):
     ...         __debuggerskip__ = True
     ...         helper_1()
     ...         __debuggerskip__ = False
     ...         result = function(*args, **kwargs)
     ...         __debuggerskip__ = True
     ...         helper_2()
     ...         # setting __debuggerskip__ to False again is not necessary
     ...         return result
     ...
     ...     return wrapped_fn
 
 When decorating a function, ipdb will directly step into ``bar()`` by
 default:
 
     >>> @foo_decorator
     ... def bar(x, y):
     ...     return x * y
 
 
 You can toggle the behavior with
 
     ipdb> skip_predicates debuggerskip false
 
 or configure it in your ``.pdbrc``
 
 
 
 License
 -------
 
 Modified from the standard pdb.Pdb class to avoid including readline, so that
 the command line completion of other programs which include this isn't
 damaged.
 
 In the future, this class will be expanded with improvements over the standard
 pdb.
 
 The original code in this file is mainly lifted out of cmd.py in Python 2.2,
 with minor changes. Licensing should therefore be under the standard Python
 terms.  For details on the PSF (Python Software Foundation) standard license,
 see:
 
 https://docs.python.org/2/license.html
 
 
 All the changes since then are under the same license as IPython.
 
 """
 
 #*****************************************************************************
 #
 #       This file is licensed under the PSF license.
 #
 #       Copyright (C) 2001 Python Software Foundation, www.python.org
 #       Copyright (C) 2005-2006 Fernando Perez. <[e-mail]>
 #
 #
 #*****************************************************************************
 
 import inspect
 import linecache
 import sys
 import re
 import os
 
 from IPython import get_ipython
 from IPython.utils import PyColorize
 from IPython.utils import coloransi, py3compat
 from IPython.core.excolors import exception_colors
 
 # skip module docstests
 __skip_doctest__ = True
 
 prompt = 'ipdb> '
 
 # We have to check this directly from sys.argv, config struct not yet available
 from pdb import Pdb as OldPdb
 
 # Allow the set_trace code to operate outside of an ipython instance, even if
 # it does so with some limitations.  The rest of this support is implemented in
 # the Tracer constructor.
 
 DEBUGGERSKIP = "__debuggerskip__"
 
 
 def make_arrow(pad):
     """generate the leading arrow in front of traceback or debugger"""
     if pad >= 2:
         return '-'*(pad-2) + '> '
     elif pad == 1:
         return '>'
     return ''
 
 
 def BdbQuit_excepthook(et, ev, tb, excepthook=None):
     """Exception hook which handles `BdbQuit` exceptions.
 
     All other exceptions are processed using the `excepthook`
     parameter.
     """
     raise ValueError(
         "`BdbQuit_excepthook` is deprecated since version 5.1",
     )
 
 
 def BdbQuit_IPython_excepthook(self, et, ev, tb, tb_offset=None):
     raise ValueError(
         "`BdbQuit_IPython_excepthook` is deprecated since version 5.1",
         DeprecationWarning, stacklevel=2)
 
 
 RGX_EXTRA_INDENT = re.compile(r'(?<=\n)\s+')
 
 
 def strip_indentation(multiline_string):
     return RGX_EXTRA_INDENT.sub('', multiline_string)
 
 
 def decorate_fn_with_doc(new_fn, old_fn, additional_text=""):
     """Make new_fn have old_fn's doc string. This is particularly useful
     for the ``do_...`` commands that hook into the help system.
     Adapted from from a comp.lang.python posting
     by Duncan Booth."""
     def wrapper(*args, **kw):
         return new_fn(*args, **kw)
     if old_fn.__doc__:
         wrapper.__doc__ = strip_indentation(old_fn.__doc__) + additional_text
     return wrapper
 
 
 class Pdb(OldPdb):
     """Modified Pdb class, does not load readline.
 
     for a standalone version that uses prompt_toolkit, see
     `IPython.terminal.debugger.TerminalPdb` and
     `IPython.terminal.debugger.set_trace()`
 
 
     This debugger can hide and skip frames that are tagged according to some predicates.
     See the `skip_predicates` commands.
 
     """
 
     default_predicates = {
         "tbhide": True,
         "readonly": False,
         "ipython_internal": True,
         "debuggerskip": True,
     }
 
     def __init__(self, completekey=None, stdin=None, stdout=None, context=5, **kwargs):
         """Create a new IPython debugger.
 
         Parameters
         ----------
         completekey : default None
             Passed to pdb.Pdb.
         stdin : default None
             Passed to pdb.Pdb.
         stdout : default None
             Passed to pdb.Pdb.
         context : int
             Number of lines of source code context to show when
             displaying stacktrace information.
         **kwargs
             Passed to pdb.Pdb.
 
         Notes
         -----
         The possibilities are python version dependent, see the python
         docs for more info.
         """
 
         # Parent constructor:
         try:
             self.context = int(context)
             if self.context <= 0:
                 raise ValueError("Context must be a positive integer")
         except (TypeError, ValueError) as e:
                 raise ValueError("Context must be a positive integer") from e
 
         # `kwargs` ensures full compatibility with stdlib's `pdb.Pdb`.
         OldPdb.__init__(self, completekey, stdin, stdout, **kwargs)
 
         # IPython changes...
         self.shell = get_ipython()
 
         if self.shell is None:
             save_main = sys.modules['__main__']
             # No IPython instance running, we must create one
             from IPython.terminal.interactiveshell import \
                 TerminalInteractiveShell
             self.shell = TerminalInteractiveShell.instance()
             # needed by any code which calls __import__("__main__") after
             # the debugger was entered. See also #9941.
             sys.modules["__main__"] = save_main
 
 
         color_scheme = self.shell.colors
 
         self.aliases = {}
 
         # Create color table: we copy the default one from the traceback
         # module and add a few attributes needed for debugging
         self.color_scheme_table = exception_colors()
 
         # shorthands
         C = coloransi.TermColors
         cst = self.color_scheme_table
 
         cst['NoColor'].colors.prompt = C.NoColor
         cst['NoColor'].colors.breakpoint_enabled = C.NoColor
         cst['NoColor'].colors.breakpoint_disabled = C.NoColor
 
         cst['Linux'].colors.prompt = C.Green
         cst['Linux'].colors.breakpoint_enabled = C.LightRed
         cst['Linux'].colors.breakpoint_disabled = C.Red
 
         cst['LightBG'].colors.prompt = C.Blue
         cst['LightBG'].colors.breakpoint_enabled = C.LightRed
         cst['LightBG'].colors.breakpoint_disabled = C.Red
 
         cst['Neutral'].colors.prompt = C.Blue
         cst['Neutral'].colors.breakpoint_enabled = C.LightRed
         cst['Neutral'].colors.breakpoint_disabled = C.Red
 
         # Add a python parser so we can syntax highlight source while
         # debugging.
         self.parser = PyColorize.Parser(style=color_scheme)
         self.set_colors(color_scheme)
 
         # Set the prompt - the default prompt is '(Pdb)'
         self.prompt = prompt
         self.skip_hidden = True
         self.report_skipped = True
 
         # list of predicates we use to skip frames
         self._predicates = self.default_predicates
 
     #
     def set_colors(self, scheme):
         """Shorthand access to the color table scheme selector method."""
         self.color_scheme_table.set_active_scheme(scheme)
         self.parser.style = scheme
 
     def set_trace(self, frame=None):
         if frame is None:
             frame = sys._getframe().f_back
         self.initial_frame = frame
         return super().set_trace(frame)
 
     def _hidden_predicate(self, frame):
         """
         Given a frame return whether it it should be hidden or not by IPython.
         """
 
         if self._predicates["readonly"]:
             fname = frame.f_code.co_filename
             # we need to check for file existence and interactively define
             # function would otherwise appear as RO.
             if os.path.isfile(fname) and not os.access(fname, os.W_OK):
                 return True
 
         if self._predicates["tbhide"]:
             if frame in (self.curframe, getattr(self, "initial_frame", None)):
                 return False
             frame_locals = self._get_frame_locals(frame)
             if "__tracebackhide__" not in frame_locals:
                 return False
             return frame_locals["__tracebackhide__"]
         return False
 
     def hidden_frames(self, stack):
         """
         Given an index in the stack return whether it should be skipped.
 
         This is used in up/down and where to skip frames.
         """
         # The f_locals dictionary is updated from the actual frame
         # locals whenever the .f_locals accessor is called, so we
         # avoid calling it here to preserve self.curframe_locals.
         # Furthermore, there is no good reason to hide the current frame.
         ip_hide = [self._hidden_predicate(s[0]) for s in stack]
         ip_start = [i for i, s in enumerate(ip_hide) if s == "__ipython_bottom__"]
         if ip_start and self._predicates["ipython_internal"]:
             ip_hide = [h if i > ip_start[0] else True for (i, h) in enumerate(ip_hide)]
         return ip_hide
 
     def interaction(self, frame, traceback):
         try:
             OldPdb.interaction(self, frame, traceback)
         except KeyboardInterrupt:
             self.stdout.write("\n" + self.shell.get_exception_only())
 
     def precmd(self, line):
         """Perform useful escapes on the command before it is executed."""
 
         if line.endswith("??"):
             line = "pinfo2 " + line[:-2]
         elif line.endswith("?"):
             line = "pinfo " + line[:-1]
 
         line = super().precmd(line)
 
         return line
 
     def new_do_frame(self, arg):
         OldPdb.do_frame(self, arg)
 
     def new_do_quit(self, arg):
 
         if hasattr(self, 'old_all_completions'):
             self.shell.Completer.all_completions = self.old_all_completions
 
         return OldPdb.do_quit(self, arg)
 
     do_q = do_quit = decorate_fn_with_doc(new_do_quit, OldPdb.do_quit)
 
     def new_do_restart(self, arg):
         """Restart command. In the context of ipython this is exactly the same
         thing as 'quit'."""
         self.msg("Restart doesn't make sense here. Using 'quit' instead.")
         return self.do_quit(arg)
 
     def print_stack_trace(self, context=None):
         Colors = self.color_scheme_table.active_colors
         ColorsNormal = Colors.Normal
         if context is None:
             context = self.context
         try:
             context = int(context)
             if context <= 0:
                 raise ValueError("Context must be a positive integer")
         except (TypeError, ValueError) as e:
                 raise ValueError("Context must be a positive integer") from e
         try:
             skipped = 0
             for hidden, frame_lineno in zip(self.hidden_frames(self.stack), self.stack):
                 if hidden and self.skip_hidden:
                     skipped += 1
                     continue
                 if skipped:
                     print(
                         f"{Colors.excName}    [... skipping {skipped} hidden frame(s)]{ColorsNormal}\n"
                     )
                     skipped = 0
                 self.print_stack_entry(frame_lineno, context=context)
             if skipped:
                 print(
                     f"{Colors.excName}    [... skipping {skipped} hidden frame(s)]{ColorsNormal}\n"
                 )
         except KeyboardInterrupt:
             pass
 
     def print_stack_entry(self, frame_lineno, prompt_prefix='\n-> ',
                           context=None):
         if context is None:
             context = self.context
         try:
             context = int(context)
             if context <= 0:
                 raise ValueError("Context must be a positive integer")
         except (TypeError, ValueError) as e:
                 raise ValueError("Context must be a positive integer") from e
         print(self.format_stack_entry(frame_lineno, '', context), file=self.stdout)
 
         # vds: >>
         frame, lineno = frame_lineno
         filename = frame.f_code.co_filename
         self.shell.hooks.synchronize_with_editor(filename, lineno, 0)
         # vds: <<
 
     def _get_frame_locals(self, frame):
         """ "
         Accessing f_local of current frame reset the namespace, so we want to avoid
         that or the following can happen
 
         ipdb> foo
         "old"
         ipdb> foo = "new"
         ipdb> foo
         "new"
         ipdb> where
         ipdb> foo
         "old"
 
         So if frame is self.current_frame we instead return self.curframe_locals
 
         """
         if frame is self.curframe:
             return self.curframe_locals
         else:
             return frame.f_locals
 
     def format_stack_entry(self, frame_lineno, lprefix=': ', context=None):
         if context is None:
             context = self.context
         try:
             context = int(context)
             if context <= 0:
                 print("Context must be a positive integer", file=self.stdout)
         except (TypeError, ValueError):
                 print("Context must be a positive integer", file=self.stdout)
 
         import reprlib
 
         ret = []
 
         Colors = self.color_scheme_table.active_colors
         ColorsNormal = Colors.Normal
         tpl_link = "%s%%s%s" % (Colors.filenameEm, ColorsNormal)
         tpl_call = "%s%%s%s%%s%s" % (Colors.vName, Colors.valEm, ColorsNormal)
         tpl_line = "%%s%s%%s %s%%s" % (Colors.lineno, ColorsNormal)
         tpl_line_em = "%%s%s%%s %s%%s%s" % (Colors.linenoEm, Colors.line, ColorsNormal)
 
         frame, lineno = frame_lineno
 
         return_value = ''
         loc_frame = self._get_frame_locals(frame)
         if "__return__" in loc_frame:
             rv = loc_frame["__return__"]
             # return_value += '->'
             return_value += reprlib.repr(rv) + "\n"
         ret.append(return_value)
 
         #s = filename + '(' + `lineno` + ')'
         filename = self.canonic(frame.f_code.co_filename)
         link = tpl_link % py3compat.cast_unicode(filename)
 
         if frame.f_code.co_name:
             func = frame.f_code.co_name
         else:
             func = "<lambda>"
 
         call = ""
         if func != "?":
             if "__args__" in loc_frame:
                 args = reprlib.repr(loc_frame["__args__"])
             else:
                 args = '()'
             call = tpl_call % (func, args)
 
         # The level info should be generated in the same format pdb uses, to
         # avoid breaking the pdbtrack functionality of python-mode in *emacs.
         if frame is self.curframe:
             ret.append('> ')
         else:
             ret.append("  ")
         ret.append("%s(%s)%s\n" % (link, lineno, call))
 
         start = lineno - 1 - context//2
         lines = linecache.getlines(filename)
         start = min(start, len(lines) - context)
         start = max(start, 0)
         lines = lines[start : start + context]
 
         for i, line in enumerate(lines):
             show_arrow = start + 1 + i == lineno
             linetpl = (frame is self.curframe or show_arrow) and tpl_line_em or tpl_line
             ret.append(
                 self.__format_line(
                     linetpl, filename, start + 1 + i, line, arrow=show_arrow
                 )
             )
         return "".join(ret)
 
     def __format_line(self, tpl_line, filename, lineno, line, arrow=False):
         bp_mark = ""
         bp_mark_color = ""
 
         new_line, err = self.parser.format2(line, 'str')
         if not err:
             line = new_line
 
         bp = None
         if lineno in self.get_file_breaks(filename):
             bps = self.get_breaks(filename, lineno)
             bp = bps[-1]
 
         if bp:
             Colors = self.color_scheme_table.active_colors
             bp_mark = str(bp.number)
             bp_mark_color = Colors.breakpoint_enabled
             if not bp.enabled:
                 bp_mark_color = Colors.breakpoint_disabled
 
         numbers_width = 7
         if arrow:
             # This is the line with the error
             pad = numbers_width - len(str(lineno)) - len(bp_mark)
             num = '%s%s' % (make_arrow(pad), str(lineno))
         else:
             num = '%*s' % (numbers_width - len(bp_mark), str(lineno))
 
         return tpl_line % (bp_mark_color + bp_mark, num, line)
 
     def print_list_lines(self, filename, first, last):
         """The printing (as opposed to the parsing part of a 'list'
         command."""
         try:
             Colors = self.color_scheme_table.active_colors
             ColorsNormal = Colors.Normal
             tpl_line = '%%s%s%%s %s%%s' % (Colors.lineno, ColorsNormal)
             tpl_line_em = '%%s%s%%s %s%%s%s' % (Colors.linenoEm, Colors.line, ColorsNormal)
             src = []
             if filename == "<string>" and hasattr(self, "_exec_filename"):
                 filename = self._exec_filename
 
             for lineno in range(first, last+1):
                 line = linecache.getline(filename, lineno)
                 if not line:
                     break
 
                 if lineno == self.curframe.f_lineno:
                     line = self.__format_line(
                         tpl_line_em, filename, lineno, line, arrow=True
                     )
                 else:
                     line = self.__format_line(
                         tpl_line, filename, lineno, line, arrow=False
                     )
 
                 src.append(line)
                 self.lineno = lineno
 
             print(''.join(src), file=self.stdout)
 
         except KeyboardInterrupt:
             pass
 
     def do_skip_predicates(self, args):
         """
         Turn on/off individual predicates as to whether a frame should be hidden/skip.
 
         The global option to skip (or not) hidden frames is set with skip_hidden
 
         To change the value of a predicate
 
             skip_predicates key [true|false]
 
         Call without arguments to see the current values.
 
         To permanently change the value of an option add the corresponding
         command to your ``~/.pdbrc`` file. If you are programmatically using the
         Pdb instance you can also change the ``default_predicates`` class
         attribute.
         """
         if not args.strip():
             print("current predicates:")
-            for (p, v) in self._predicates.items():
+            for p, v in self._predicates.items():
                 print("   ", p, ":", v)
             return
         type_value = args.strip().split(" ")
         if len(type_value) != 2:
             print(
                 f"Usage: skip_predicates <type> <value>, with <type> one of {set(self._predicates.keys())}"
             )
             return
 
         type_, value = type_value
         if type_ not in self._predicates:
             print(f"{type_!r} not in {set(self._predicates.keys())}")
             return
         if value.lower() not in ("true", "yes", "1", "no", "false", "0"):
             print(
                 f"{value!r} is invalid - use one of ('true', 'yes', '1', 'no', 'false', '0')"
             )
             return
 
         self._predicates[type_] = value.lower() in ("true", "yes", "1")
         if not any(self._predicates.values()):
             print(
                 "Warning, all predicates set to False, skip_hidden may not have any effects."
             )
 
     def do_skip_hidden(self, arg):
         """
         Change whether or not we should skip frames with the
         __tracebackhide__ attribute.
         """
         if not arg.strip():
             print(
                 f"skip_hidden = {self.skip_hidden}, use 'yes','no', 'true', or 'false' to change."
             )
         elif arg.strip().lower() in ("true", "yes"):
             self.skip_hidden = True
         elif arg.strip().lower() in ("false", "no"):
             self.skip_hidden = False
         if not any(self._predicates.values()):
             print(
                 "Warning, all predicates set to False, skip_hidden may not have any effects."
             )
 
     def do_list(self, arg):
         """Print lines of code from the current stack frame
         """
         self.lastcmd = 'list'
         last = None
         if arg:
             try:
                 x = eval(arg, {}, {})
                 if type(x) == type(()):
                     first, last = x
                     first = int(first)
                     last = int(last)
                     if last < first:
                         # Assume it's a count
                         last = first + last
                 else:
                     first = max(1, int(x) - 5)
             except:
                 print('*** Error in argument:', repr(arg), file=self.stdout)
                 return
         elif self.lineno is None:
             first = max(1, self.curframe.f_lineno - 5)
         else:
             first = self.lineno + 1
         if last is None:
             last = first + 10
         self.print_list_lines(self.curframe.f_code.co_filename, first, last)
 
         # vds: >>
         lineno = first
         filename = self.curframe.f_code.co_filename
         self.shell.hooks.synchronize_with_editor(filename, lineno, 0)
         # vds: <<
 
     do_l = do_list
 
     def getsourcelines(self, obj):
         lines, lineno = inspect.findsource(obj)
         if inspect.isframe(obj) and obj.f_globals is self._get_frame_locals(obj):
             # must be a module frame: do not try to cut a block out of it
             return lines, 1
         elif inspect.ismodule(obj):
             return lines, 1
         return inspect.getblock(lines[lineno:]), lineno+1
 
     def do_longlist(self, arg):
         """Print lines of code from the current stack frame.
 
         Shows more lines than 'list' does.
         """
         self.lastcmd = 'longlist'
         try:
             lines, lineno = self.getsourcelines(self.curframe)
         except OSError as err:
             self.error(err)
             return
         last = lineno + len(lines)
         self.print_list_lines(self.curframe.f_code.co_filename, lineno, last)
     do_ll = do_longlist
 
     def do_debug(self, arg):
         """debug code
         Enter a recursive debugger that steps through the code
         argument (which is an arbitrary expression or statement to be
         executed in the current environment).
         """
         trace_function = sys.gettrace()
         sys.settrace(None)
         globals = self.curframe.f_globals
         locals = self.curframe_locals
         p = self.__class__(completekey=self.completekey,
                            stdin=self.stdin, stdout=self.stdout)
         p.use_rawinput = self.use_rawinput
         p.prompt = "(%s) " % self.prompt.strip()
         self.message("ENTERING RECURSIVE DEBUGGER")
         sys.call_tracing(p.run, (arg, globals, locals))
         self.message("LEAVING RECURSIVE DEBUGGER")
         sys.settrace(trace_function)
         self.lastcmd = p.lastcmd
 
     def do_pdef(self, arg):
         """Print the call signature for any callable object.
 
         The debugger interface to %pdef"""
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("pdef")(arg, namespaces=namespaces)
 
     def do_pdoc(self, arg):
         """Print the docstring for an object.
 
         The debugger interface to %pdoc."""
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("pdoc")(arg, namespaces=namespaces)
 
     def do_pfile(self, arg):
         """Print (or run through pager) the file where an object is defined.
 
         The debugger interface to %pfile.
         """
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("pfile")(arg, namespaces=namespaces)
 
     def do_pinfo(self, arg):
         """Provide detailed information about an object.
 
         The debugger interface to %pinfo, i.e., obj?."""
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("pinfo")(arg, namespaces=namespaces)
 
     def do_pinfo2(self, arg):
         """Provide extra detailed information about an object.
 
         The debugger interface to %pinfo2, i.e., obj??."""
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("pinfo2")(arg, namespaces=namespaces)
 
     def do_psource(self, arg):
         """Print (or run through pager) the source code for an object."""
         namespaces = [
             ("Locals", self.curframe_locals),
             ("Globals", self.curframe.f_globals),
         ]
         self.shell.find_line_magic("psource")(arg, namespaces=namespaces)
 
     def do_where(self, arg):
         """w(here)
         Print a stack trace, with the most recent frame at the bottom.
         An arrow indicates the "current frame", which determines the
         context of most commands. 'bt' is an alias for this command.
 
         Take a number as argument as an (optional) number of context line to
         print"""
         if arg:
             try:
                 context = int(arg)
             except ValueError as err:
                 self.error(err)
                 return
             self.print_stack_trace(context)
         else:
             self.print_stack_trace()
 
     do_w = do_where
 
     def break_anywhere(self, frame):
         """
         _stop_in_decorator_internals is overly restrictive, as we may still want
         to trace function calls, so we need to also update break_anywhere so
         that is we don't `stop_here`, because of debugger skip, we may still
         stop at any point inside the function
 
         """
 
         sup = super().break_anywhere(frame)
         if sup:
             return sup
         if self._predicates["debuggerskip"]:
             if DEBUGGERSKIP in frame.f_code.co_varnames:
                 return True
             if frame.f_back and self._get_frame_locals(frame.f_back).get(DEBUGGERSKIP):
                 return True
         return False
 
     def _is_in_decorator_internal_and_should_skip(self, frame):
         """
         Utility to tell us whether we are in a decorator internal and should stop.
 
         """
 
         # if we are disabled don't skip
         if not self._predicates["debuggerskip"]:
             return False
 
         # if frame is tagged, skip by default.
         if DEBUGGERSKIP in frame.f_code.co_varnames:
             return True
 
         # if one of the parent frame value set to True skip as well.
 
         cframe = frame
         while getattr(cframe, "f_back", None):
             cframe = cframe.f_back
             if self._get_frame_locals(cframe).get(DEBUGGERSKIP):
                 return True
 
         return False
 
     def stop_here(self, frame):
-
         if self._is_in_decorator_internal_and_should_skip(frame) is True:
             return False
 
         hidden = False
         if self.skip_hidden:
             hidden = self._hidden_predicate(frame)
         if hidden:
             if self.report_skipped:
                 Colors = self.color_scheme_table.active_colors
                 ColorsNormal = Colors.Normal
                 print(
                     f"{Colors.excName}    [... skipped 1 hidden frame]{ColorsNormal}\n"
                 )
         return super().stop_here(frame)
 
     def do_up(self, arg):
         """u(p) [count]
         Move the current frame count (default one) levels up in the
         stack trace (to an older frame).
 
         Will skip hidden frames.
         """
         # modified version of upstream that skips
         # frames with __tracebackhide__
         if self.curindex == 0:
             self.error("Oldest frame")
             return
         try:
             count = int(arg or 1)
         except ValueError:
             self.error("Invalid frame count (%s)" % arg)
             return
         skipped = 0
         if count < 0:
             _newframe = 0
         else:
             counter = 0
             hidden_frames = self.hidden_frames(self.stack)
             for i in range(self.curindex - 1, -1, -1):
                 if hidden_frames[i] and self.skip_hidden:
                     skipped += 1
                     continue
                 counter += 1
                 if counter >= count:
                     break
             else:
                 # if no break occurred.
                 self.error(
                     "all frames above hidden, use `skip_hidden False` to get get into those."
                 )
                 return
 
             Colors = self.color_scheme_table.active_colors
             ColorsNormal = Colors.Normal
             _newframe = i
         self._select_frame(_newframe)
         if skipped:
             print(
                 f"{Colors.excName}    [... skipped {skipped} hidden frame(s)]{ColorsNormal}\n"
             )
 
     def do_down(self, arg):
         """d(own) [count]
         Move the current frame count (default one) levels down in the
         stack trace (to a newer frame).
 
         Will skip hidden frames.
         """
         if self.curindex + 1 == len(self.stack):
             self.error("Newest frame")
             return
         try:
             count = int(arg or 1)
         except ValueError:
             self.error("Invalid frame count (%s)" % arg)
             return
         if count < 0:
             _newframe = len(self.stack) - 1
         else:
             counter = 0
             skipped = 0
             hidden_frames = self.hidden_frames(self.stack)
             for i in range(self.curindex + 1, len(self.stack)):
                 if hidden_frames[i] and self.skip_hidden:
                     skipped += 1
                     continue
                 counter += 1
                 if counter >= count:
                     break
             else:
                 self.error(
                     "all frames below hidden, use `skip_hidden False` to get get into those."
                 )
                 return
 
             Colors = self.color_scheme_table.active_colors
             ColorsNormal = Colors.Normal
             if skipped:
                 print(
                     f"{Colors.excName}    [... skipped {skipped} hidden frame(s)]{ColorsNormal}\n"
                 )
             _newframe = i
 
         self._select_frame(_newframe)
 
     do_d = do_down
     do_u = do_up
 
     def do_context(self, context):
         """context number_of_lines
         Set the number of lines of source code to show when displaying
         stacktrace information.
         """
         try:
             new_context = int(context)
             if new_context <= 0:
                 raise ValueError()
             self.context = new_context
         except ValueError:
             self.error("The 'context' command requires a positive integer argument.")
 
 
 class InterruptiblePdb(Pdb):
     """Version of debugger where KeyboardInterrupt exits the debugger altogether."""
 
     def cmdloop(self, intro=None):
         """Wrap cmdloop() such that KeyboardInterrupt stops the debugger."""
         try:
             return OldPdb.cmdloop(self, intro=intro)
         except KeyboardInterrupt:
             self.stop_here = lambda frame: False
             self.do_quit("")
             sys.settrace(None)
             self.quitting = False
             raise
 
     def _cmdloop(self):
         while True:
             try:
                 # keyboard interrupts allow for an easy way to cancel
                 # the current command, so allow them during interactive input
                 self.allow_kbdint = True
                 self.cmdloop()
                 self.allow_kbdint = False
                 break
             except KeyboardInterrupt:
                 self.message('--KeyboardInterrupt--')
                 raise
 
 
 def set_trace(frame=None):
     """
     Start debugging from `frame`.
 
     If frame is not specified, debugging starts from caller's frame.
     """
     Pdb().set_trace(frame or sys._getframe().f_back)

IPython/core/tests/test_debugger.py

@@ -1,575 +1,574 b''
 """Tests for debugging machinery.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import builtins
 import os
 import sys
 import platform
 
 from tempfile import NamedTemporaryFile
 from textwrap import dedent
 from unittest.mock import patch
 
 from IPython.core import debugger
 from IPython.testing import IPYTHON_TESTING_TIMEOUT_SCALE
 from IPython.testing.decorators import skip_win32
 import pytest
 
 #-----------------------------------------------------------------------------
 # Helper classes, from CPython's Pdb test suite
 #-----------------------------------------------------------------------------
 
 class _FakeInput(object):
     """
     A fake input stream for pdb's interactive debugger.  Whenever a
     line is read, print it (to simulate the user typing it), and then
     return it.  The set of lines to return is specified in the
     constructor; they should not have trailing newlines.
     """
     def __init__(self, lines):
         self.lines = iter(lines)
 
     def readline(self):
         line = next(self.lines)
         print(line)
         return line+'\n'
 
 class PdbTestInput(object):
     """Context manager that makes testing Pdb in doctests easier."""
 
     def __init__(self, input):
         self.input = input
 
     def __enter__(self):
         self.real_stdin = sys.stdin
         sys.stdin = _FakeInput(self.input)
 
     def __exit__(self, *exc):
         sys.stdin = self.real_stdin
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 def test_ipdb_magics():
     '''Test calling some IPython magics from ipdb.
 
     First, set up some test functions and classes which we can inspect.
 
     >>> class ExampleClass(object):
     ...    """Docstring for ExampleClass."""
     ...    def __init__(self):
     ...        """Docstring for ExampleClass.__init__"""
     ...        pass
     ...    def __str__(self):
     ...        return "ExampleClass()"
 
     >>> def example_function(x, y, z="hello"):
     ...     """Docstring for example_function."""
     ...     pass
 
     >>> old_trace = sys.gettrace()
 
     Create a function which triggers ipdb.
 
     >>> def trigger_ipdb():
     ...    a = ExampleClass()
     ...    debugger.Pdb().set_trace()
 
     >>> with PdbTestInput([
     ...    'pdef example_function',
     ...    'pdoc ExampleClass',
     ...    'up',
     ...    'down',
     ...    'list',
     ...    'pinfo a',
     ...    'll',
     ...    'continue',
     ... ]):
     ...     trigger_ipdb()
     --Return--
     None
     > <doctest ...>(3)trigger_ipdb()
           1 def trigger_ipdb():
           2    a = ExampleClass()
     ----> 3    debugger.Pdb().set_trace()
     <BLANKLINE>
     ipdb> pdef example_function
      example_function(x, y, z='hello')
      ipdb> pdoc ExampleClass
     Class docstring:
         Docstring for ExampleClass.
     Init docstring:
         Docstring for ExampleClass.__init__
     ipdb> up
     > <doctest ...>(11)<module>()
           7    'pinfo a',
           8    'll',
           9    'continue',
          10 ]):
     ---> 11     trigger_ipdb()
     <BLANKLINE>
     ipdb> down
     None
     > <doctest ...>(3)trigger_ipdb()
           1 def trigger_ipdb():
           2    a = ExampleClass()
     ----> 3    debugger.Pdb().set_trace()
     <BLANKLINE>
     ipdb> list
           1 def trigger_ipdb():
           2    a = ExampleClass()
     ----> 3    debugger.Pdb().set_trace()
     <BLANKLINE>
     ipdb> pinfo a
     Type:           ExampleClass
     String form:    ExampleClass()
     Namespace:      Local...
     Docstring:      Docstring for ExampleClass.
     Init docstring: Docstring for ExampleClass.__init__
     ipdb> ll
           1 def trigger_ipdb():
           2    a = ExampleClass()
     ----> 3    debugger.Pdb().set_trace()
     <BLANKLINE>
     ipdb> continue
     
     Restore previous trace function, e.g. for coverage.py    
     
     >>> sys.settrace(old_trace)
     '''
 
 def test_ipdb_magics2():
     '''Test ipdb with a very short function.
     
     >>> old_trace = sys.gettrace()
 
     >>> def bar():
     ...     pass
 
     Run ipdb.
 
     >>> with PdbTestInput([
     ...    'continue',
     ... ]):
     ...     debugger.Pdb().runcall(bar)
     > <doctest ...>(2)bar()
           1 def bar():
     ----> 2    pass
     <BLANKLINE>
     ipdb> continue
     
     Restore previous trace function, e.g. for coverage.py    
     
     >>> sys.settrace(old_trace)
     '''
 
 def can_quit():
     '''Test that quit work in ipydb
 
     >>> old_trace = sys.gettrace()
 
     >>> def bar():
     ...     pass
 
     >>> with PdbTestInput([
     ...    'quit',
     ... ]):
     ...     debugger.Pdb().runcall(bar)
     > <doctest ...>(2)bar()
             1 def bar():
     ----> 2    pass
     <BLANKLINE>
     ipdb> quit
 
     Restore previous trace function, e.g. for coverage.py
 
     >>> sys.settrace(old_trace)
     '''
 
 
 def can_exit():
     '''Test that quit work in ipydb
 
     >>> old_trace = sys.gettrace()
 
     >>> def bar():
     ...     pass
 
     >>> with PdbTestInput([
     ...    'exit',
     ... ]):
     ...     debugger.Pdb().runcall(bar)
     > <doctest ...>(2)bar()
             1 def bar():
     ----> 2    pass
     <BLANKLINE>
     ipdb> exit
 
     Restore previous trace function, e.g. for coverage.py
 
     >>> sys.settrace(old_trace)
     '''
 
 
 def test_interruptible_core_debugger():
     """The debugger can be interrupted.
 
     The presumption is there is some mechanism that causes a KeyboardInterrupt
     (this is implemented in ipykernel).  We want to ensure the
     KeyboardInterrupt cause debugging to cease.
     """
     def raising_input(msg="", called=[0]):
         called[0] += 1
         assert called[0] == 1, "input() should only be called once!"
         raise KeyboardInterrupt()
 
     tracer_orig = sys.gettrace()
     try:
         with patch.object(builtins, "input", raising_input):
             debugger.InterruptiblePdb().set_trace()
             # The way this test will fail is by set_trace() never exiting,
             # resulting in a timeout by the test runner. The alternative
             # implementation would involve a subprocess, but that adds issues
             # with interrupting subprocesses that are rather complex, so it's
             # simpler just to do it this way.
     finally:
         # restore the original trace function
         sys.settrace(tracer_orig)
 
 
 @skip_win32
 def test_xmode_skip():
     """that xmode skip frames
 
     Not as a doctest as pytest does not run doctests.
     """
     import pexpect
     env = os.environ.copy()
     env["IPY_TEST_SIMPLE_PROMPT"] = "1"
 
     child = pexpect.spawn(
         sys.executable, ["-m", "IPython", "--colors=nocolor"], env=env
     )
     child.timeout = 15 * IPYTHON_TESTING_TIMEOUT_SCALE
 
     child.expect("IPython")
     child.expect("\n")
     child.expect_exact("In [1]")
 
     block = dedent(
         """
     def f():
         __tracebackhide__ = True
         g()
 
     def g():
         raise ValueError
 
     f()
     """
     )
 
     for line in block.splitlines():
         child.sendline(line)
         child.expect_exact(line)
     child.expect_exact("skipping")
 
     block = dedent(
         """
     def f():
         __tracebackhide__ = True
         g()
 
     def g():
         from IPython.core.debugger import set_trace
         set_trace()
 
     f()
     """
     )
 
     for line in block.splitlines():
         child.sendline(line)
         child.expect_exact(line)
 
     child.expect("ipdb>")
     child.sendline("w")
     child.expect("hidden")
     child.expect("ipdb>")
     child.sendline("skip_hidden false")
     child.sendline("w")
     child.expect("__traceba")
     child.expect("ipdb>")
 
     child.close()
 
 
 skip_decorators_blocks = (
     """
     def helpers_helper():
         pass # should not stop here except breakpoint
     """,
     """
     def helper_1():
         helpers_helper() # should not stop here
     """,
     """
     def helper_2():
         pass # should not stop here
     """,
     """
     def pdb_skipped_decorator2(function):
         def wrapped_fn(*args, **kwargs):
             __debuggerskip__ = True
             helper_2()
             __debuggerskip__ = False
             result = function(*args, **kwargs)
             __debuggerskip__ = True
             helper_2()
             return result
         return wrapped_fn
     """,
     """
     def pdb_skipped_decorator(function):
         def wrapped_fn(*args, **kwargs):
             __debuggerskip__ = True
             helper_1()
             __debuggerskip__ = False
             result = function(*args, **kwargs)
             __debuggerskip__ = True
             helper_2()
             return result
         return wrapped_fn
     """,
     """
     @pdb_skipped_decorator
     @pdb_skipped_decorator2
     def bar(x, y):
         return x * y
     """,
     """import IPython.terminal.debugger as ipdb""",
     """
     def f():
         ipdb.set_trace()
         bar(3, 4)
     """,
     """
     f()
     """,
 )
 
 
 def _decorator_skip_setup():
     import pexpect
 
     env = os.environ.copy()
     env["IPY_TEST_SIMPLE_PROMPT"] = "1"
     env["PROMPT_TOOLKIT_NO_CPR"] = "1"
 
     child = pexpect.spawn(
         sys.executable, ["-m", "IPython", "--colors=nocolor"], env=env
     )
     child.timeout = 15 * IPYTHON_TESTING_TIMEOUT_SCALE
 
     child.expect("IPython")
     child.expect("\n")
 
     child.timeout = 5 * IPYTHON_TESTING_TIMEOUT_SCALE
     child.str_last_chars = 500
 
     dedented_blocks = [dedent(b).strip() for b in skip_decorators_blocks]
     in_prompt_number = 1
     for cblock in dedented_blocks:
         child.expect_exact(f"In [{in_prompt_number}]:")
         in_prompt_number += 1
         for line in cblock.splitlines():
             child.sendline(line)
             child.expect_exact(line)
         child.sendline("")
     return child
 
 
 @pytest.mark.skip(reason="recently fail for unknown reason on CI")
 @skip_win32
 def test_decorator_skip():
     """test that decorator frames can be skipped."""
 
     child = _decorator_skip_setup()
 
     child.expect_exact("ipython-input-8")
     child.expect_exact("3     bar(3, 4)")
     child.expect("ipdb>")
 
     child.expect("ipdb>")
     child.sendline("step")
     child.expect_exact("step")
     child.expect_exact("--Call--")
     child.expect_exact("ipython-input-6")
 
     child.expect_exact("1 @pdb_skipped_decorator")
 
     child.sendline("s")
     child.expect_exact("return x * y")
 
     child.close()
 
 
 @pytest.mark.skip(reason="recently fail for unknown reason on CI")
 @pytest.mark.skipif(platform.python_implementation() == "PyPy", reason="issues on PyPy")
 @skip_win32
 def test_decorator_skip_disabled():
     """test that decorator frame skipping can be disabled"""
 
     child = _decorator_skip_setup()
 
     child.expect_exact("3     bar(3, 4)")
 
     for input_, expected in [
         ("skip_predicates debuggerskip False", ""),
         ("skip_predicates", "debuggerskip : False"),
         ("step", "---> 2     def wrapped_fn"),
         ("step", "----> 3         __debuggerskip__"),
         ("step", "----> 4         helper_1()"),
         ("step", "---> 1 def helper_1():"),
         ("next", "----> 2     helpers_helper()"),
         ("next", "--Return--"),
         ("next", "----> 5         __debuggerskip__ = False"),
     ]:
         child.expect("ipdb>")
         child.sendline(input_)
         child.expect_exact(input_)
         child.expect_exact(expected)
 
     child.close()
 
 
 @pytest.mark.skipif(platform.python_implementation() == "PyPy", reason="issues on PyPy")
 @skip_win32
 def test_decorator_skip_with_breakpoint():
     """test that decorator frame skipping can be disabled"""
 
     import pexpect
 
     env = os.environ.copy()
     env["IPY_TEST_SIMPLE_PROMPT"] = "1"
     env["PROMPT_TOOLKIT_NO_CPR"] = "1"
 
     child = pexpect.spawn(
         sys.executable, ["-m", "IPython", "--colors=nocolor"], env=env
     )
     child.timeout = 15 * IPYTHON_TESTING_TIMEOUT_SCALE
     child.str_last_chars = 500
 
     child.expect("IPython")
     child.expect("\n")
 
     child.timeout = 5 * IPYTHON_TESTING_TIMEOUT_SCALE
 
     ### we need a filename, so we need to exec the full block with a filename
     with NamedTemporaryFile(suffix=".py", dir=".", delete=True) as tf:
-
         name = tf.name[:-3].split("/")[-1]
         tf.write("\n".join([dedent(x) for x in skip_decorators_blocks[:-1]]).encode())
         tf.flush()
         codeblock = f"from {name} import f"
 
         dedented_blocks = [
             codeblock,
             "f()",
         ]
 
         in_prompt_number = 1
         for cblock in dedented_blocks:
             child.expect_exact(f"In [{in_prompt_number}]:")
             in_prompt_number += 1
             for line in cblock.splitlines():
                 child.sendline(line)
                 child.expect_exact(line)
             child.sendline("")
 
         # as the filename does not exists, we'll rely on the filename prompt
         child.expect_exact("47     bar(3, 4)")
 
         for input_, expected in [
             (f"b {name}.py:3", ""),
             ("step", "1---> 3     pass # should not stop here except"),
             ("step", "---> 38 @pdb_skipped_decorator"),
             ("continue", ""),
         ]:
             child.expect("ipdb>")
             child.sendline(input_)
             child.expect_exact(input_)
             child.expect_exact(expected)
 
     child.close()
 
 
 @skip_win32
 def test_where_erase_value():
     """Test that `where` does not access f_locals and erase values."""
     import pexpect
 
     env = os.environ.copy()
     env["IPY_TEST_SIMPLE_PROMPT"] = "1"
 
     child = pexpect.spawn(
         sys.executable, ["-m", "IPython", "--colors=nocolor"], env=env
     )
     child.timeout = 15 * IPYTHON_TESTING_TIMEOUT_SCALE
 
     child.expect("IPython")
     child.expect("\n")
     child.expect_exact("In [1]")
 
     block = dedent(
         """
     def simple_f():
          myvar = 1
          print(myvar)
          1/0
          print(myvar)
     simple_f()    """
     )
 
     for line in block.splitlines():
         child.sendline(line)
         child.expect_exact(line)
     child.expect_exact("ZeroDivisionError")
     child.expect_exact("In [2]:")
 
     child.sendline("%debug")
 
     ##
     child.expect("ipdb>")
 
     child.sendline("myvar")
     child.expect("1")
 
     ##
     child.expect("ipdb>")
 
     child.sendline("myvar = 2")
 
     ##
     child.expect_exact("ipdb>")
 
     child.sendline("myvar")
 
     child.expect_exact("2")
 
     ##
     child.expect("ipdb>")
     child.sendline("where")
 
     ##
     child.expect("ipdb>")
     child.sendline("myvar")
 
     child.expect_exact("2")
     child.expect("ipdb>")
 
     child.close()

IPython/core/tests/test_magic.py

@@ -1,1513 +1,1512 b''
 # -*- coding: utf-8 -*-
 """Tests for various magic functions."""
 
 import gc
 import io
 import os
 import re
 import shlex
 import sys
 import warnings
 from importlib import invalidate_caches
 from io import StringIO
 from pathlib import Path
 from textwrap import dedent
 from unittest import TestCase, mock
 
 import pytest
 
 from IPython import get_ipython
 from IPython.core import magic
 from IPython.core.error import UsageError
 from IPython.core.magic import (
     Magics,
     cell_magic,
     line_magic,
     magics_class,
     register_cell_magic,
     register_line_magic,
 )
 from IPython.core.magics import code, execution, logging, osm, script
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 from IPython.utils.io import capture_output
 from IPython.utils.process import find_cmd
 from IPython.utils.tempdir import TemporaryDirectory, TemporaryWorkingDirectory
 from IPython.utils.syspathcontext import prepended_to_syspath
 
 from .test_debugger import PdbTestInput
 
 from tempfile import NamedTemporaryFile
 
 @magic.magics_class
 class DummyMagics(magic.Magics): pass
 
 def test_extract_code_ranges():
     instr = "1 3 5-6 7-9 10:15 17: :10 10- -13 :"
     expected = [
         (0, 1),
         (2, 3),
         (4, 6),
         (6, 9),
         (9, 14),
         (16, None),
         (None, 9),
         (9, None),
         (None, 13),
         (None, None),
     ]
     actual = list(code.extract_code_ranges(instr))
     assert actual == expected
 
 def test_extract_symbols():
     source = """import foo\na = 10\ndef b():\n    return 42\n\n\nclass A: pass\n\n\n"""
     symbols_args = ["a", "b", "A", "A,b", "A,a", "z"]
     expected = [([], ['a']),
                 (["def b():\n    return 42\n"], []),
                 (["class A: pass\n"], []),
                 (["class A: pass\n", "def b():\n    return 42\n"], []),
                 (["class A: pass\n"], ['a']),
                 ([], ['z'])]
     for symbols, exp in zip(symbols_args, expected):
         assert code.extract_symbols(source, symbols) == exp
 
 
 def test_extract_symbols_raises_exception_with_non_python_code():
     source = ("=begin A Ruby program :)=end\n"
               "def hello\n"
               "puts 'Hello world'\n"
               "end")
     with pytest.raises(SyntaxError):
         code.extract_symbols(source, "hello")
 
 
 def test_magic_not_found():
     # magic not found raises UsageError
     with pytest.raises(UsageError):
         _ip.run_line_magic("doesntexist", "")
 
     # ensure result isn't success when a magic isn't found
     result = _ip.run_cell('%doesntexist')
     assert isinstance(result.error_in_exec, UsageError)
 
 
 def test_cell_magic_not_found():
     # magic not found raises UsageError
     with pytest.raises(UsageError):
         _ip.run_cell_magic('doesntexist', 'line', 'cell')
 
     # ensure result isn't success when a magic isn't found
     result = _ip.run_cell('%%doesntexist')
     assert isinstance(result.error_in_exec, UsageError)
 
 
 def test_magic_error_status():
     def fail(shell):
         1/0
     _ip.register_magic_function(fail)
     result = _ip.run_cell('%fail')
     assert isinstance(result.error_in_exec, ZeroDivisionError)
 
 
 def test_config():
     """ test that config magic does not raise
     can happen if Configurable init is moved too early into
     Magics.__init__ as then a Config object will be registered as a
     magic.
     """
     ## should not raise.
     _ip.run_line_magic("config", "")
 
 
 def test_config_available_configs():
     """ test that config magic prints available configs in unique and
     sorted order. """
     with capture_output() as captured:
         _ip.run_line_magic("config", "")
 
     stdout = captured.stdout
     config_classes = stdout.strip().split('\n')[1:]
     assert config_classes == sorted(set(config_classes))
 
 def test_config_print_class():
     """ test that config with a classname prints the class's options. """
     with capture_output() as captured:
         _ip.run_line_magic("config", "TerminalInteractiveShell")
 
     stdout = captured.stdout
     assert re.match(
         "TerminalInteractiveShell.* options", stdout.splitlines()[0]
     ), f"{stdout}\n\n1st line of stdout not like 'TerminalInteractiveShell.* options'"
 
 
 def test_rehashx():
     # clear up everything
     _ip.alias_manager.clear_aliases()
     del _ip.db['syscmdlist']
 
     _ip.run_line_magic("rehashx", "")
     # Practically ALL ipython development systems will have more than 10 aliases
 
     assert len(_ip.alias_manager.aliases) > 10
     for name, cmd in _ip.alias_manager.aliases:
         # we must strip dots from alias names
         assert "." not in name
 
     # rehashx must fill up syscmdlist
     scoms = _ip.db['syscmdlist']
     assert len(scoms) > 10
 
 
 def test_magic_parse_options():
     """Test that we don't mangle paths when parsing magic options."""
     ip = get_ipython()
     path = 'c:\\x'
     m = DummyMagics(ip)
     opts = m.parse_options('-f %s' % path,'f:')[0]
     # argv splitting is os-dependent
     if os.name == 'posix':
         expected = 'c:x'
     else:
         expected = path
     assert opts["f"] == expected
 
 
 def test_magic_parse_long_options():
     """Magic.parse_options can handle --foo=bar long options"""
     ip = get_ipython()
     m = DummyMagics(ip)
     opts, _ = m.parse_options("--foo --bar=bubble", "a", "foo", "bar=")
     assert "foo" in opts
     assert "bar" in opts
     assert opts["bar"] == "bubble"
 
 
 def doctest_hist_f():
     """Test %hist -f with temporary filename.
 
     In [9]: import tempfile
 
     In [10]: tfile = tempfile.mktemp('.py','tmp-ipython-')
 
     In [11]: %hist -nl -f $tfile 3
 
     In [13]: import os; os.unlink(tfile)
     """
 
 
 def doctest_hist_op():
     """Test %hist -op
 
     In [1]: class b(float):
        ...:     pass
        ...:
 
     In [2]: class s(object):
        ...:     def __str__(self):
        ...:         return 's'
        ...:
 
     In [3]:
 
     In [4]: class r(b):
        ...:     def __repr__(self):
        ...:         return 'r'
        ...:
 
     In [5]: class sr(s,r): pass
        ...:
 
     In [6]:
 
     In [7]: bb=b()
 
     In [8]: ss=s()
 
     In [9]: rr=r()
 
     In [10]: ssrr=sr()
 
     In [11]: 4.5
     Out[11]: 4.5
 
     In [12]: str(ss)
     Out[12]: 's'
 
     In [13]:
 
     In [14]: %hist -op
     >>> class b:
     ...     pass
     ...
     >>> class s(b):
     ...     def __str__(self):
     ...         return 's'
     ...
     >>>
     >>> class r(b):
     ...     def __repr__(self):
     ...         return 'r'
     ...
     >>> class sr(s,r): pass
     >>>
     >>> bb=b()
     >>> ss=s()
     >>> rr=r()
     >>> ssrr=sr()
     >>> 4.5
     4.5
     >>> str(ss)
     's'
     >>>
     """
 
 def test_hist_pof():
     ip = get_ipython()
     ip.run_cell("1+2", store_history=True)
     #raise Exception(ip.history_manager.session_number)
     #raise Exception(list(ip.history_manager._get_range_session()))
     with TemporaryDirectory() as td:
         tf = os.path.join(td, 'hist.py')
         ip.run_line_magic('history', '-pof %s' % tf)
         assert os.path.isfile(tf)
 
 
 def test_macro():
     ip = get_ipython()
     ip.history_manager.reset()   # Clear any existing history.
     cmds = ["a=1", "def b():\n  return a**2", "print(a,b())"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
     ip.run_line_magic("macro", "test 1-3")
     assert ip.user_ns["test"].value == "\n".join(cmds) + "\n"
 
     # List macros
     assert "test" in ip.run_line_magic("macro", "")
 
 
 def test_macro_run():
     """Test that we can run a multi-line macro successfully."""
     ip = get_ipython()
     ip.history_manager.reset()
     cmds = ["a=10", "a+=1", "print(a)", "%macro test 2-3"]
     for cmd in cmds:
         ip.run_cell(cmd, store_history=True)
     assert ip.user_ns["test"].value == "a+=1\nprint(a)\n"
     with tt.AssertPrints("12"):
         ip.run_cell("test")
     with tt.AssertPrints("13"):
         ip.run_cell("test")
 
 
 def test_magic_magic():
     """Test %magic"""
     ip = get_ipython()
     with capture_output() as captured:
         ip.run_line_magic("magic", "")
 
     stdout = captured.stdout
     assert "%magic" in stdout
     assert "IPython" in stdout
     assert "Available" in stdout
 
 
 @dec.skipif_not_numpy
 def test_numpy_reset_array_undec():
     "Test '%reset array' functionality"
     _ip.ex("import numpy as np")
     _ip.ex("a = np.empty(2)")
     assert "a" in _ip.user_ns
     _ip.run_line_magic("reset", "-f array")
     assert "a" not in _ip.user_ns
 
 
 def test_reset_out():
     "Test '%reset out' magic"
     _ip.run_cell("parrot = 'dead'", store_history=True)
     # test '%reset -f out', make an Out prompt
     _ip.run_cell("parrot", store_history=True)
     assert "dead" in [_ip.user_ns[x] for x in ("_", "__", "___")]
     _ip.run_line_magic("reset", "-f out")
     assert "dead" not in [_ip.user_ns[x] for x in ("_", "__", "___")]
     assert len(_ip.user_ns["Out"]) == 0
 
 
 def test_reset_in():
     "Test '%reset in' magic"
     # test '%reset -f in'
     _ip.run_cell("parrot", store_history=True)
     assert "parrot" in [_ip.user_ns[x] for x in ("_i", "_ii", "_iii")]
     _ip.run_line_magic("reset", "-f in")
     assert "parrot" not in [_ip.user_ns[x] for x in ("_i", "_ii", "_iii")]
     assert len(set(_ip.user_ns["In"])) == 1
 
 
 def test_reset_dhist():
     "Test '%reset dhist' magic"
     _ip.run_cell("tmp = [d for d in _dh]")  # copy before clearing
     _ip.run_line_magic("cd", os.path.dirname(pytest.__file__))
     _ip.run_line_magic("cd", "-")
     assert len(_ip.user_ns["_dh"]) > 0
     _ip.run_line_magic("reset", "-f dhist")
     assert len(_ip.user_ns["_dh"]) == 0
     _ip.run_cell("_dh = [d for d in tmp]")  # restore
 
 
 def test_reset_in_length():
     "Test that '%reset in' preserves In[] length"
     _ip.run_cell("print 'foo'")
     _ip.run_cell("reset -f in")
     assert len(_ip.user_ns["In"]) == _ip.displayhook.prompt_count + 1
 
 
 class TestResetErrors(TestCase):
 
     def test_reset_redefine(self):
 
         @magics_class
         class KernelMagics(Magics):
               @line_magic
               def less(self, shell): pass
 
         _ip.register_magics(KernelMagics)
 
         with self.assertLogs() as cm:
             # hack, we want to just capture logs, but assertLogs fails if not
             # logs get produce.
             # so log one things we ignore.
             import logging as log_mod
             log = log_mod.getLogger()
             log.info('Nothing')
             # end hack.
             _ip.run_cell("reset -f")
 
         assert len(cm.output) == 1
         for out in cm.output:
             assert "Invalid alias" not in out
 
 def test_tb_syntaxerror():
     """test %tb after a SyntaxError"""
     ip = get_ipython()
     ip.run_cell("for")
 
     # trap and validate stdout
     save_stdout = sys.stdout
     try:
         sys.stdout = StringIO()
         ip.run_cell("%tb")
         out = sys.stdout.getvalue()
     finally:
         sys.stdout = save_stdout
     # trim output, and only check the last line
     last_line = out.rstrip().splitlines()[-1].strip()
     assert last_line == "SyntaxError: invalid syntax"
 
 
 def test_time():
     ip = get_ipython()
 
     with tt.AssertPrints("Wall time: "):
         ip.run_cell("%time None")
 
     ip.run_cell("def f(kmjy):\n"
                 "    %time print (2*kmjy)")
 
     with tt.AssertPrints("Wall time: "):
         with tt.AssertPrints("hihi", suppress=False):
             ip.run_cell("f('hi')")
 
 
 # ';' at the end of %time prevents instruction value to be printed.
 # This tests fix for #13837.
 def test_time_no_output_with_semicolon():
     ip = get_ipython()
 
     # Test %time cases
     with tt.AssertPrints(" 123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%time 123000+456")
 
     with tt.AssertNotPrints(" 123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%time 123000+456;")
 
     with tt.AssertPrints(" 123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%time 123000+456 # Comment")
 
     with tt.AssertNotPrints(" 123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%time 123000+456; # Comment")
 
     with tt.AssertPrints(" 123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%time 123000+456 # ;Comment")
 
     # Test %%time cases
     with tt.AssertPrints("123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%%time\n123000+456\n\n\n")
 
     with tt.AssertNotPrints("123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%%time\n123000+456;\n\n\n")
 
     with tt.AssertPrints("123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%%time\n123000+456  # Comment\n\n\n")
 
     with tt.AssertNotPrints("123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%%time\n123000+456;  # Comment\n\n\n")
 
     with tt.AssertPrints("123456"):
         with tt.AssertPrints("Wall time: ", suppress=False):
             with tt.AssertPrints("CPU times: ", suppress=False):
                 ip.run_cell("%%time\n123000+456  # ;Comment\n\n\n")
 
 
 def test_time_last_not_expression():
     ip.run_cell("%%time\n"
                 "var_1 = 1\n"
                 "var_2 = 2\n")
     assert ip.user_ns['var_1'] == 1
     del ip.user_ns['var_1']
     assert ip.user_ns['var_2'] == 2
     del ip.user_ns['var_2']
 
 
 @dec.skip_win32
 def test_time2():
     ip = get_ipython()
 
     with tt.AssertPrints("CPU times: user "):
         ip.run_cell("%time None")
 
 def test_time3():
     """Erroneous magic function calls, issue gh-3334"""
     ip = get_ipython()
     ip.user_ns.pop('run', None)
 
     with tt.AssertNotPrints("not found", channel='stderr'):
         ip.run_cell("%%time\n"
                     "run = 0\n"
                     "run += 1")
 
 def test_multiline_time():
     """Make sure last statement from time return a value."""
     ip = get_ipython()
     ip.user_ns.pop('run', None)
 
     ip.run_cell(
         dedent(
             """\
         %%time
         a = "ho"
         b = "hey"
         a+b
         """
         )
     )
     assert ip.user_ns_hidden["_"] == "hohey"
 
 
 def test_time_local_ns():
     """
     Test that local_ns is actually global_ns when running a cell magic
     """
     ip = get_ipython()
     ip.run_cell("%%time\n" "myvar = 1")
     assert ip.user_ns["myvar"] == 1
     del ip.user_ns["myvar"]
 
 
 def test_doctest_mode():
     "Toggle doctest_mode twice, it should be a no-op and run without error"
     _ip.run_line_magic("doctest_mode", "")
     _ip.run_line_magic("doctest_mode", "")
 
 
 def test_parse_options():
     """Tests for basic options parsing in magics."""
     # These are only the most minimal of tests, more should be added later.  At
     # the very least we check that basic text/unicode calls work OK.
     m = DummyMagics(_ip)
     assert m.parse_options("foo", "")[1] == "foo"
     assert m.parse_options("foo", "")[1] == "foo"
 
 
 def test_parse_options_preserve_non_option_string():
     """Test to assert preservation of non-option part of magic-block, while parsing magic options."""
     m = DummyMagics(_ip)
     opts, stmt = m.parse_options(
         " -n1  -r 13 _ = 314 + foo", "n:r:", preserve_non_opts=True
     )
     assert opts == {"n": "1", "r": "13"}
     assert stmt == "_ = 314 + foo"
 
 
 def test_run_magic_preserve_code_block():
     """Test to assert preservation of non-option part of magic-block, while running magic."""
     _ip.user_ns["spaces"] = []
     _ip.run_line_magic(
         "timeit", "-n1 -r1 spaces.append([s.count(' ') for s in ['document']])"
     )
     assert _ip.user_ns["spaces"] == [[0]]
 
 
 def test_dirops():
     """Test various directory handling operations."""
     # curpath = lambda :os.path.splitdrive(os.getcwd())[1].replace('\\','/')
     curpath = os.getcwd
     startdir = os.getcwd()
     ipdir = os.path.realpath(_ip.ipython_dir)
     try:
         _ip.run_line_magic("cd", '"%s"' % ipdir)
         assert curpath() == ipdir
         _ip.run_line_magic("cd", "-")
         assert curpath() == startdir
         _ip.run_line_magic("pushd", '"%s"' % ipdir)
         assert curpath() == ipdir
         _ip.run_line_magic("popd", "")
         assert curpath() == startdir
     finally:
         os.chdir(startdir)
 
 
 def test_cd_force_quiet():
     """Test OSMagics.cd_force_quiet option"""
     _ip.config.OSMagics.cd_force_quiet = True
     osmagics = osm.OSMagics(shell=_ip)
 
     startdir = os.getcwd()
     ipdir = os.path.realpath(_ip.ipython_dir)
 
     try:
         with tt.AssertNotPrints(ipdir):
             osmagics.cd('"%s"' % ipdir)
         with tt.AssertNotPrints(startdir):
             osmagics.cd('-')
     finally:
         os.chdir(startdir)
 
 
 def test_xmode():
     # Calling xmode three times should be a no-op
     xmode = _ip.InteractiveTB.mode
     for i in range(4):
         _ip.run_line_magic("xmode", "")
     assert _ip.InteractiveTB.mode == xmode
 
 def test_reset_hard():
     monitor = []
     class A(object):
         def __del__(self):
             monitor.append(1)
         def __repr__(self):
             return "<A instance>"
 
     _ip.user_ns["a"] = A()
     _ip.run_cell("a")
 
     assert monitor == []
     _ip.run_line_magic("reset", "-f")
     assert monitor == [1]
 
 class TestXdel(tt.TempFileMixin):
     def test_xdel(self):
         """Test that references from %run are cleared by xdel."""
         src = ("class A(object):\n"
                "    monitor = []\n"
                "    def __del__(self):\n"
                "        self.monitor.append(1)\n"
                "a = A()\n")
         self.mktmp(src)
         # %run creates some hidden references...
         _ip.run_line_magic("run", "%s" % self.fname)
         # ... as does the displayhook.
         _ip.run_cell("a")
 
         monitor = _ip.user_ns["A"].monitor
         assert monitor == []
 
         _ip.run_line_magic("xdel", "a")
 
         # Check that a's __del__ method has been called.
         gc.collect(0)
         assert monitor == [1]
 
 def doctest_who():
     """doctest for %who
 
     In [1]: %reset -sf
 
     In [2]: alpha = 123
 
     In [3]: beta = 'beta'
 
     In [4]: %who int
     alpha
 
     In [5]: %who str
     beta
 
     In [6]: %whos
     Variable   Type    Data/Info
     ----------------------------
     alpha      int     123
     beta       str     beta
 
     In [7]: %who_ls
     Out[7]: ['alpha', 'beta']
     """
 
 def test_whos():
     """Check that whos is protected against objects where repr() fails."""
     class A(object):
         def __repr__(self):
             raise Exception()
     _ip.user_ns['a'] = A()
     _ip.run_line_magic("whos", "")
 
 def doctest_precision():
     """doctest for %precision
 
     In [1]: f = get_ipython().display_formatter.formatters['text/plain']
 
     In [2]: %precision 5
     Out[2]: '%.5f'
 
     In [3]: f.float_format
     Out[3]: '%.5f'
 
     In [4]: %precision %e
     Out[4]: '%e'
 
     In [5]: f(3.1415927)
     Out[5]: '3.141593e+00'
     """
 
 def test_debug_magic():
     """Test debugging a small code with %debug
 
     In [1]: with PdbTestInput(['c']):
        ...:     %debug print("a b") #doctest: +ELLIPSIS
        ...:
     ...
     ipdb> c
     a b
     In [2]:
     """
 
 def test_psearch():
     with tt.AssertPrints("dict.fromkeys"):
         _ip.run_cell("dict.fr*?")
     with tt.AssertPrints("π.is_integer"):
         _ip.run_cell("π = 3.14;\nπ.is_integ*?")
 
 def test_timeit_shlex():
     """test shlex issues with timeit (#1109)"""
     _ip.ex("def f(*a,**kw): pass")
     _ip.run_line_magic("timeit", '-n1 "this is a bug".count(" ")')
     _ip.run_line_magic("timeit", '-r1 -n1 f(" ", 1)')
     _ip.run_line_magic("timeit", '-r1 -n1 f(" ", 1, " ", 2, " ")')
     _ip.run_line_magic("timeit", '-r1 -n1 ("a " + "b")')
     _ip.run_line_magic("timeit", '-r1 -n1 f("a " + "b")')
     _ip.run_line_magic("timeit", '-r1 -n1 f("a " + "b ")')
 
 
 def test_timeit_special_syntax():
     "Test %%timeit with IPython special syntax"
     @register_line_magic
     def lmagic(line):
         ip = get_ipython()
         ip.user_ns['lmagic_out'] = line
 
     # line mode test
     _ip.run_line_magic("timeit", "-n1 -r1 %lmagic my line")
     assert _ip.user_ns["lmagic_out"] == "my line"
     # cell mode test
     _ip.run_cell_magic("timeit", "-n1 -r1", "%lmagic my line2")
     assert _ip.user_ns["lmagic_out"] == "my line2"
 
 
 def test_timeit_return():
     """
     test whether timeit -o return object
     """
 
     res = _ip.run_line_magic('timeit','-n10 -r10 -o 1')
     assert(res is not None)
 
 def test_timeit_quiet():
     """
     test quiet option of timeit magic
     """
     with tt.AssertNotPrints("loops"):
         _ip.run_cell("%timeit -n1 -r1 -q 1")
 
 def test_timeit_return_quiet():
     with tt.AssertNotPrints("loops"):
         res = _ip.run_line_magic('timeit', '-n1 -r1 -q -o 1')
     assert (res is not None)
 
 def test_timeit_invalid_return():
     with pytest.raises(SyntaxError):
         _ip.run_line_magic('timeit', 'return')
 
 @dec.skipif(execution.profile is None)
 def test_prun_special_syntax():
     "Test %%prun with IPython special syntax"
     @register_line_magic
     def lmagic(line):
         ip = get_ipython()
         ip.user_ns['lmagic_out'] = line
 
     # line mode test
     _ip.run_line_magic("prun", "-q %lmagic my line")
     assert _ip.user_ns["lmagic_out"] == "my line"
     # cell mode test
     _ip.run_cell_magic("prun", "-q", "%lmagic my line2")
     assert _ip.user_ns["lmagic_out"] == "my line2"
 
 
 @dec.skipif(execution.profile is None)
 def test_prun_quotes():
     "Test that prun does not clobber string escapes (GH #1302)"
     _ip.magic(r"prun -q x = '\t'")
     assert _ip.user_ns["x"] == "\t"
 
 
 def test_extension():
     # Debugging information for failures of this test
     print('sys.path:')
     for p in sys.path:
         print(' ', p)
     print('CWD', os.getcwd())
 
     pytest.raises(ImportError, _ip.magic, "load_ext daft_extension")
     daft_path = os.path.join(os.path.dirname(__file__), "daft_extension")
     sys.path.insert(0, daft_path)
     try:
         _ip.user_ns.pop('arq', None)
         invalidate_caches()   # Clear import caches
         _ip.run_line_magic("load_ext", "daft_extension")
         assert _ip.user_ns["arq"] == 185
         _ip.run_line_magic("unload_ext", "daft_extension")
         assert 'arq' not in _ip.user_ns
     finally:
         sys.path.remove(daft_path)
 
 
 def test_notebook_export_json():
     pytest.importorskip("nbformat")
     _ip = get_ipython()
     _ip.history_manager.reset()   # Clear any existing history.
     cmds = ["a=1", "def b():\n  return a**2", "print('noël, été', b())"]
     for i, cmd in enumerate(cmds, start=1):
         _ip.history_manager.store_inputs(i, cmd)
     with TemporaryDirectory() as td:
         outfile = os.path.join(td, "nb.ipynb")
         _ip.run_line_magic("notebook", "%s" % outfile)
 
 
 class TestEnv(TestCase):
 
     def test_env(self):
         env = _ip.run_line_magic("env", "")
         self.assertTrue(isinstance(env, dict))
 
     def test_env_secret(self):
         env = _ip.run_line_magic("env", "")
         hidden = "<hidden>"
         with mock.patch.dict(
             os.environ,
             {
                 "API_KEY": "abc123",
                 "SECRET_THING": "ssshhh",
                 "JUPYTER_TOKEN": "",
                 "VAR": "abc"
             }
         ):
             env = _ip.run_line_magic("env", "")
         assert env["API_KEY"] == hidden
         assert env["SECRET_THING"] == hidden
         assert env["JUPYTER_TOKEN"] == hidden
         assert env["VAR"] == "abc"
 
     def test_env_get_set_simple(self):
         env = _ip.run_line_magic("env", "var val1")
         self.assertEqual(env, None)
         self.assertEqual(os.environ["var"], "val1")
         self.assertEqual(_ip.run_line_magic("env", "var"), "val1")
         env = _ip.run_line_magic("env", "var=val2")
         self.assertEqual(env, None)
         self.assertEqual(os.environ['var'], 'val2')
 
     def test_env_get_set_complex(self):
         env = _ip.run_line_magic("env", "var 'val1 '' 'val2")
         self.assertEqual(env, None)
         self.assertEqual(os.environ['var'], "'val1 '' 'val2")
         self.assertEqual(_ip.run_line_magic("env", "var"), "'val1 '' 'val2")
         env = _ip.run_line_magic("env", 'var=val2 val3="val4')
         self.assertEqual(env, None)
         self.assertEqual(os.environ['var'], 'val2 val3="val4')
 
     def test_env_set_bad_input(self):
         self.assertRaises(UsageError, lambda: _ip.run_line_magic("set_env", "var"))
 
     def test_env_set_whitespace(self):
         self.assertRaises(UsageError, lambda: _ip.run_line_magic("env", "var A=B"))
 
 
 class CellMagicTestCase(TestCase):
 
     def check_ident(self, magic):
         # Manually called, we get the result
         out = _ip.run_cell_magic(magic, "a", "b")
         assert out == ("a", "b")
         # Via run_cell, it goes into the user's namespace via displayhook
         _ip.run_cell("%%" + magic + " c\nd\n")
         assert _ip.user_ns["_"] == ("c", "d\n")
 
     def test_cell_magic_func_deco(self):
         "Cell magic using simple decorator"
         @register_cell_magic
         def cellm(line, cell):
             return line, cell
 
         self.check_ident('cellm')
 
     def test_cell_magic_reg(self):
         "Cell magic manually registered"
         def cellm(line, cell):
             return line, cell
 
         _ip.register_magic_function(cellm, 'cell', 'cellm2')
         self.check_ident('cellm2')
 
     def test_cell_magic_class(self):
         "Cell magics declared via a class"
         @magics_class
         class MyMagics(Magics):
 
             @cell_magic
             def cellm3(self, line, cell):
                 return line, cell
 
         _ip.register_magics(MyMagics)
         self.check_ident('cellm3')
 
     def test_cell_magic_class2(self):
         "Cell magics declared via a class, #2"
         @magics_class
         class MyMagics2(Magics):
 
             @cell_magic('cellm4')
             def cellm33(self, line, cell):
                 return line, cell
 
         _ip.register_magics(MyMagics2)
         self.check_ident('cellm4')
         # Check that nothing is registered as 'cellm33'
         c33 = _ip.find_cell_magic('cellm33')
         assert c33 == None
 
 def test_file():
     """Basic %%writefile"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, "file1")
         ip.run_cell_magic(
             "writefile",
             fname,
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line2" in s
 
 
 @dec.skip_win32
 def test_file_single_quote():
     """Basic %%writefile with embedded single quotes"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, "'file1'")
         ip.run_cell_magic(
             "writefile",
             fname,
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line2" in s
 
 
 @dec.skip_win32
 def test_file_double_quote():
     """Basic %%writefile with embedded double quotes"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, '"file1"')
         ip.run_cell_magic(
             "writefile",
             fname,
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line2" in s
 
 
 def test_file_var_expand():
     """%%writefile $filename"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, "file1")
         ip.user_ns["filename"] = fname
         ip.run_cell_magic(
             "writefile",
             "$filename",
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line2" in s
 
 
 def test_file_unicode():
     """%%writefile with unicode cell"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, 'file1')
         ip.run_cell_magic("writefile", fname, u'\n'.join([
             u'liné1',
             u'liné2',
         ]))
         with io.open(fname, encoding='utf-8') as f:
             s = f.read()
         assert "liné1\n" in s
         assert "liné2" in s
 
 
 def test_file_amend():
     """%%writefile -a amends files"""
     ip = get_ipython()
     with TemporaryDirectory() as td:
         fname = os.path.join(td, "file2")
         ip.run_cell_magic(
             "writefile",
             fname,
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         ip.run_cell_magic(
             "writefile",
             "-a %s" % fname,
             "\n".join(
                 [
                     "line3",
                     "line4",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line3\n" in s
 
 
 def test_file_spaces():
     """%%file with spaces in filename"""
     ip = get_ipython()
     with TemporaryWorkingDirectory() as td:
         fname = "file name"
         ip.run_cell_magic(
             "file",
             '"%s"' % fname,
             "\n".join(
                 [
                     "line1",
                     "line2",
                 ]
             ),
         )
         s = Path(fname).read_text(encoding="utf-8")
         assert "line1\n" in s
         assert "line2" in s
 
 
 def test_script_config():
     ip = get_ipython()
     ip.config.ScriptMagics.script_magics = ['whoda']
     sm = script.ScriptMagics(shell=ip)
     assert "whoda" in sm.magics["cell"]
 
 
 def test_script_out():
     ip = get_ipython()
     ip.run_cell_magic("script", f"--out output {sys.executable}", "print('hi')")
     assert ip.user_ns["output"].strip() == "hi"
 
 
 def test_script_err():
     ip = get_ipython()
     ip.run_cell_magic(
         "script",
         f"--err error {sys.executable}",
         "import sys; print('hello', file=sys.stderr)",
     )
     assert ip.user_ns["error"].strip() == "hello"
 
 
 def test_script_out_err():
-
     ip = get_ipython()
     ip.run_cell_magic(
         "script",
         f"--out output --err error {sys.executable}",
         "\n".join(
             [
                 "import sys",
                 "print('hi')",
                 "print('hello', file=sys.stderr)",
             ]
         ),
     )
     assert ip.user_ns["output"].strip() == "hi"
     assert ip.user_ns["error"].strip() == "hello"
 
 
 async def test_script_bg_out():
     ip = get_ipython()
     ip.run_cell_magic("script", f"--bg --out output {sys.executable}", "print('hi')")
     assert (await ip.user_ns["output"].read()).strip() == b"hi"
     assert ip.user_ns["output"].at_eof()
 
 
 async def test_script_bg_err():
     ip = get_ipython()
     ip.run_cell_magic(
         "script",
         f"--bg --err error {sys.executable}",
         "import sys; print('hello', file=sys.stderr)",
     )
     assert (await ip.user_ns["error"].read()).strip() == b"hello"
     assert ip.user_ns["error"].at_eof()
 
 
 async def test_script_bg_out_err():
     ip = get_ipython()
     ip.run_cell_magic(
         "script",
         f"--bg --out output --err error {sys.executable}",
         "\n".join(
             [
                 "import sys",
                 "print('hi')",
                 "print('hello', file=sys.stderr)",
             ]
         ),
     )
     assert (await ip.user_ns["output"].read()).strip() == b"hi"
     assert (await ip.user_ns["error"].read()).strip() == b"hello"
     assert ip.user_ns["output"].at_eof()
     assert ip.user_ns["error"].at_eof()
 
 
 async def test_script_bg_proc():
     ip = get_ipython()
     ip.run_cell_magic(
         "script",
         f"--bg --out output --proc p {sys.executable}",
         "\n".join(
             [
                 "import sys",
                 "print('hi')",
                 "print('hello', file=sys.stderr)",
             ]
         ),
     )
     p = ip.user_ns["p"]
     await p.wait()
     assert p.returncode == 0
     assert (await p.stdout.read()).strip() == b"hi"
     # not captured, so empty
     assert (await p.stderr.read()) == b""
     assert p.stdout.at_eof()
     assert p.stderr.at_eof()
 
 
 def test_script_defaults():
     ip = get_ipython()
     for cmd in ['sh', 'bash', 'perl', 'ruby']:
         try:
             find_cmd(cmd)
         except Exception:
             pass
         else:
             assert cmd in ip.magics_manager.magics["cell"]
 
 
 @magics_class
 class FooFoo(Magics):
     """class with both %foo and %%foo magics"""
     @line_magic('foo')
     def line_foo(self, line):
         "I am line foo"
         pass
 
     @cell_magic("foo")
     def cell_foo(self, line, cell):
         "I am cell foo, not line foo"
         pass
 
 def test_line_cell_info():
     """%%foo and %foo magics are distinguishable to inspect"""
     ip = get_ipython()
     ip.magics_manager.register(FooFoo)
     oinfo = ip.object_inspect("foo")
     assert oinfo["found"] is True
     assert oinfo["ismagic"] is True
 
     oinfo = ip.object_inspect("%%foo")
     assert oinfo["found"] is True
     assert oinfo["ismagic"] is True
     assert oinfo["docstring"] == FooFoo.cell_foo.__doc__
 
     oinfo = ip.object_inspect("%foo")
     assert oinfo["found"] is True
     assert oinfo["ismagic"] is True
     assert oinfo["docstring"] == FooFoo.line_foo.__doc__
 
 
 def test_multiple_magics():
     ip = get_ipython()
     foo1 = FooFoo(ip)
     foo2 = FooFoo(ip)
     mm = ip.magics_manager
     mm.register(foo1)
     assert mm.magics["line"]["foo"].__self__ is foo1
     mm.register(foo2)
     assert mm.magics["line"]["foo"].__self__ is foo2
 
 
 def test_alias_magic():
     """Test %alias_magic."""
     ip = get_ipython()
     mm = ip.magics_manager
 
     # Basic operation: both cell and line magics are created, if possible.
     ip.run_line_magic("alias_magic", "timeit_alias timeit")
     assert "timeit_alias" in mm.magics["line"]
     assert "timeit_alias" in mm.magics["cell"]
 
     # --cell is specified, line magic not created.
     ip.run_line_magic("alias_magic", "--cell timeit_cell_alias timeit")
     assert "timeit_cell_alias" not in mm.magics["line"]
     assert "timeit_cell_alias" in mm.magics["cell"]
 
     # Test that line alias is created successfully.
     ip.run_line_magic("alias_magic", "--line env_alias env")
     assert ip.run_line_magic("env", "") == ip.run_line_magic("env_alias", "")
 
     # Test that line alias with parameters passed in is created successfully.
     ip.run_line_magic(
         "alias_magic", "--line history_alias history --params " + shlex.quote("3")
     )
     assert "history_alias" in mm.magics["line"]
 
 
 def test_save():
     """Test %save."""
     ip = get_ipython()
     ip.history_manager.reset()   # Clear any existing history.
     cmds = ["a=1", "def b():\n  return a**2", "print(a, b())"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
     with TemporaryDirectory() as tmpdir:
         file = os.path.join(tmpdir, "testsave.py")
         ip.run_line_magic("save", "%s 1-10" % file)
         content = Path(file).read_text(encoding="utf-8")
         assert content.count(cmds[0]) == 1
         assert "coding: utf-8" in content
         ip.run_line_magic("save", "-a %s 1-10" % file)
         content = Path(file).read_text(encoding="utf-8")
         assert content.count(cmds[0]) == 2
         assert "coding: utf-8" in content
 
 
 def test_save_with_no_args():
     ip = get_ipython()
     ip.history_manager.reset()  # Clear any existing history.
     cmds = ["a=1", "def b():\n    return a**2", "print(a, b())", "%save"]
     for i, cmd in enumerate(cmds, start=1):
         ip.history_manager.store_inputs(i, cmd)
 
     with TemporaryDirectory() as tmpdir:
         path = os.path.join(tmpdir, "testsave.py")
         ip.run_line_magic("save", path)
         content = Path(path).read_text(encoding="utf-8")
         expected_content = dedent(
             """\
             # coding: utf-8
             a=1
             def b():
                 return a**2
             print(a, b())
             """
         )
         assert content == expected_content
 
 
 def test_store():
     """Test %store."""
     ip = get_ipython()
     ip.run_line_magic('load_ext', 'storemagic')
 
     # make sure the storage is empty
     ip.run_line_magic("store", "-z")
     ip.user_ns["var"] = 42
     ip.run_line_magic("store", "var")
     ip.user_ns["var"] = 39
     ip.run_line_magic("store", "-r")
     assert ip.user_ns["var"] == 42
 
     ip.run_line_magic("store", "-d var")
     ip.user_ns["var"] = 39
     ip.run_line_magic("store", "-r")
     assert ip.user_ns["var"] == 39
 
 
 def _run_edit_test(arg_s, exp_filename=None,
                         exp_lineno=-1,
                         exp_contents=None,
                         exp_is_temp=None):
     ip = get_ipython()
     M = code.CodeMagics(ip)
     last_call = ['','']
     opts,args = M.parse_options(arg_s,'prxn:')
     filename, lineno, is_temp = M._find_edit_target(ip, args, opts, last_call)
 
     if exp_filename is not None:
         assert exp_filename == filename
     if exp_contents is not None:
         with io.open(filename, 'r', encoding='utf-8') as f:
             contents = f.read()
         assert exp_contents == contents
     if exp_lineno != -1:
         assert exp_lineno == lineno
     if exp_is_temp is not None:
         assert exp_is_temp == is_temp
 
 
 def test_edit_interactive():
     """%edit on interactively defined objects"""
     ip = get_ipython()
     n = ip.execution_count
     ip.run_cell("def foo(): return 1", store_history=True)
 
     with pytest.raises(code.InteractivelyDefined) as e:
         _run_edit_test("foo")
     assert e.value.index == n
 
 
 def test_edit_cell():
     """%edit [cell id]"""
     ip = get_ipython()
 
     ip.run_cell("def foo(): return 1", store_history=True)
 
     # test
     _run_edit_test("1", exp_contents=ip.user_ns['In'][1], exp_is_temp=True)
 
 def test_edit_fname():
     """%edit file"""
     # test
     _run_edit_test("test file.py", exp_filename="test file.py")
 
 def test_bookmark():
     ip = get_ipython()
     ip.run_line_magic('bookmark', 'bmname')
     with tt.AssertPrints('bmname'):
         ip.run_line_magic('bookmark', '-l')
     ip.run_line_magic('bookmark', '-d bmname')
 
 def test_ls_magic():
     ip = get_ipython()
     json_formatter = ip.display_formatter.formatters['application/json']
     json_formatter.enabled = True
     lsmagic = ip.run_line_magic("lsmagic", "")
     with warnings.catch_warnings(record=True) as w:
         j = json_formatter(lsmagic)
     assert sorted(j) == ["cell", "line"]
     assert w == []  # no warnings
 
 
 def test_strip_initial_indent():
     def sii(s):
         lines = s.splitlines()
         return '\n'.join(code.strip_initial_indent(lines))
 
     assert sii("  a = 1\nb = 2") == "a = 1\nb = 2"
     assert sii("  a\n    b\nc") == "a\n  b\nc"
     assert sii("a\n  b") == "a\n  b"
 
 def test_logging_magic_quiet_from_arg():
     _ip.config.LoggingMagics.quiet = False
     lm = logging.LoggingMagics(shell=_ip)
     with TemporaryDirectory() as td:
         try:
             with tt.AssertNotPrints(re.compile("Activating.*")):
                 lm.logstart('-q {}'.format(
                         os.path.join(td, "quiet_from_arg.log")))
         finally:
             _ip.logger.logstop()
 
 def test_logging_magic_quiet_from_config():
     _ip.config.LoggingMagics.quiet = True
     lm = logging.LoggingMagics(shell=_ip)
     with TemporaryDirectory() as td:
         try:
             with tt.AssertNotPrints(re.compile("Activating.*")):
                 lm.logstart(os.path.join(td, "quiet_from_config.log"))
         finally:
             _ip.logger.logstop()
 
 
 def test_logging_magic_not_quiet():
     _ip.config.LoggingMagics.quiet = False
     lm = logging.LoggingMagics(shell=_ip)
     with TemporaryDirectory() as td:
         try:
             with tt.AssertPrints(re.compile("Activating.*")):
                 lm.logstart(os.path.join(td, "not_quiet.log"))
         finally:
             _ip.logger.logstop()
 
 
 def test_time_no_var_expand():
     _ip.user_ns["a"] = 5
     _ip.user_ns["b"] = []
     _ip.run_line_magic("time", 'b.append("{a}")')
     assert _ip.user_ns["b"] == ["{a}"]
 
 
 # this is slow, put at the end for local testing.
 def test_timeit_arguments():
     "Test valid timeit arguments, should not cause SyntaxError (GH #1269)"
     _ip.run_line_magic("timeit", "-n1 -r1 a=('#')")
 
 
 MINIMAL_LAZY_MAGIC = """
 from IPython.core.magic import (
     Magics,
     magics_class,
     line_magic,
     cell_magic,
 )
 
 
 @magics_class
 class LazyMagics(Magics):
     @line_magic
     def lazy_line(self, line):
         print("Lazy Line")
 
     @cell_magic
     def lazy_cell(self, line, cell):
         print("Lazy Cell")
 
 
 def load_ipython_extension(ipython):
     ipython.register_magics(LazyMagics)
 """
 
 
 def test_lazy_magics():
     with pytest.raises(UsageError):
         ip.run_line_magic("lazy_line", "")
 
     startdir = os.getcwd()
 
     with TemporaryDirectory() as tmpdir:
         with prepended_to_syspath(tmpdir):
             ptempdir = Path(tmpdir)
             tf = ptempdir / "lazy_magic_module.py"
             tf.write_text(MINIMAL_LAZY_MAGIC)
             ip.magics_manager.register_lazy("lazy_line", Path(tf.name).name[:-3])
             with tt.AssertPrints("Lazy Line"):
                 ip.run_line_magic("lazy_line", "")
 
 
 TEST_MODULE = """
 print('Loaded my_tmp')
 if __name__ == "__main__":
     print('I just ran a script')
 """
 
 def test_run_module_from_import_hook():
     "Test that a module can be loaded via an import hook"
     with TemporaryDirectory() as tmpdir:
         fullpath = os.path.join(tmpdir, "my_tmp.py")
         Path(fullpath).write_text(TEST_MODULE, encoding="utf-8")
 
         import importlib.abc
         import importlib.util
 
         class MyTempImporter(importlib.abc.MetaPathFinder, importlib.abc.SourceLoader):
             def find_spec(self, fullname, path, target=None):
                 if fullname == "my_tmp":
                     return importlib.util.spec_from_loader(fullname, self)
 
             def get_filename(self, fullname):
                 assert fullname == "my_tmp"
                 return fullpath
 
             def get_data(self, path):
                 assert Path(path).samefile(fullpath)
                 return Path(fullpath).read_text(encoding="utf-8")
 
         sys.meta_path.insert(0, MyTempImporter())
 
         with capture_output() as captured:
             _ip.run_line_magic("run", "-m my_tmp")
             _ip.run_cell("import my_tmp")
 
         output = "Loaded my_tmp\nI just ran a script\nLoaded my_tmp\n"
         assert output == captured.stdout
 
         sys.meta_path.pop(0)

IPython/core/tests/test_pylabtools.py

@@ -1,271 +1,270 b''
 """Tests for pylab tools module.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 
 from binascii import a2b_base64
 from io import BytesIO
 
 import pytest
 
 matplotlib = pytest.importorskip("matplotlib")
 matplotlib.use('Agg')
 from matplotlib.figure import Figure
 
 from matplotlib import pyplot as plt
 from matplotlib_inline import backend_inline
 import numpy as np
 
 from IPython.core.getipython import get_ipython
 from IPython.core.interactiveshell import InteractiveShell
 from IPython.core.display import _PNG, _JPEG
 from .. import pylabtools as pt
 
 from IPython.testing import decorators as dec
 
 
 def test_figure_to_svg():
     # simple empty-figure test
     fig = plt.figure()
     assert pt.print_figure(fig, "svg") is None
 
     plt.close('all')
 
     # simple check for at least svg-looking output
     fig = plt.figure()
     ax = fig.add_subplot(1,1,1)
     ax.plot([1,2,3])
     plt.draw()
     svg = pt.print_figure(fig, "svg")[:100].lower()
     assert "doctype svg" in svg
 
 
 def _check_pil_jpeg_bytes():
     """Skip if PIL can't write JPEGs to BytesIO objects"""
     # PIL's JPEG plugin can't write to BytesIO objects
     # Pillow fixes this
     from PIL import Image
     buf = BytesIO()
     img = Image.new("RGB", (4,4))
     try:
         img.save(buf, 'jpeg')
     except Exception as e:
         ename = e.__class__.__name__
         raise pytest.skip("PIL can't write JPEG to BytesIO: %s: %s" % (ename, e)) from e
 
 @dec.skip_without("PIL.Image")
 def test_figure_to_jpeg():
     _check_pil_jpeg_bytes()
     # simple check for at least jpeg-looking output
     fig = plt.figure()
     ax = fig.add_subplot(1,1,1)
     ax.plot([1,2,3])
     plt.draw()
     jpeg = pt.print_figure(fig, 'jpeg', pil_kwargs={'optimize': 50})[:100].lower()
     assert jpeg.startswith(_JPEG)
 
 def test_retina_figure():
     # simple empty-figure test
     fig = plt.figure()
     assert pt.retina_figure(fig) == None
     plt.close('all')
 
     fig = plt.figure()
     ax = fig.add_subplot(1,1,1)
     ax.plot([1,2,3])
     plt.draw()
     png, md = pt.retina_figure(fig)
     assert png.startswith(_PNG)
     assert "width" in md
     assert "height" in md
 
 
 _fmt_mime_map = {
     'png': 'image/png',
     'jpeg': 'image/jpeg',
     'pdf': 'application/pdf',
     'retina': 'image/png',
     'svg': 'image/svg+xml',
 }
 
 def test_select_figure_formats_str():
     ip = get_ipython()
     for fmt, active_mime in _fmt_mime_map.items():
         pt.select_figure_formats(ip, fmt)
         for mime, f in ip.display_formatter.formatters.items():
             if mime == active_mime:
                 assert Figure in f
             else:
                 assert Figure not in f
 
 def test_select_figure_formats_kwargs():
     ip = get_ipython()
     kwargs = dict(bbox_inches="tight")
     pt.select_figure_formats(ip, "png", **kwargs)
     formatter = ip.display_formatter.formatters["image/png"]
     f = formatter.lookup_by_type(Figure)
     cell = f.keywords
     expected = kwargs
     expected["base64"] = True
     expected["fmt"] = "png"
     assert cell == expected
 
     # check that the formatter doesn't raise
     fig = plt.figure()
     ax = fig.add_subplot(1,1,1)
     ax.plot([1,2,3])
     plt.draw()
     formatter.enabled = True
     png = formatter(fig)
     assert isinstance(png, str)
     png_bytes = a2b_base64(png)
     assert png_bytes.startswith(_PNG)
 
 def test_select_figure_formats_set():
     ip = get_ipython()
     for fmts in [
         {'png', 'svg'},
         ['png'],
         ('jpeg', 'pdf', 'retina'),
         {'svg'},
     ]:
         active_mimes = {_fmt_mime_map[fmt] for fmt in fmts}
         pt.select_figure_formats(ip, fmts)
         for mime, f in ip.display_formatter.formatters.items():
             if mime in active_mimes:
                 assert Figure in f
             else:
                 assert Figure not in f
 
 def test_select_figure_formats_bad():
     ip = get_ipython()
     with pytest.raises(ValueError):
         pt.select_figure_formats(ip, 'foo')
     with pytest.raises(ValueError):
         pt.select_figure_formats(ip, {'png', 'foo'})
     with pytest.raises(ValueError):
         pt.select_figure_formats(ip, ['retina', 'pdf', 'bar', 'bad'])
 
 def test_import_pylab():
     ns = {}
     pt.import_pylab(ns, import_all=False)
     assert "plt" in ns
     assert ns["np"] == np
 
 
 class TestPylabSwitch(object):
     class Shell(InteractiveShell):
         def init_history(self):
             """Sets up the command history, and starts regular autosaves."""
             self.config.HistoryManager.hist_file = ":memory:"
             super().init_history()
 
         def enable_gui(self, gui):
             pass
 
     def setup(self):
         import matplotlib
         def act_mpl(backend):
             matplotlib.rcParams['backend'] = backend
 
         # Save rcParams since they get modified
         self._saved_rcParams = matplotlib.rcParams
         self._saved_rcParamsOrig = matplotlib.rcParamsOrig
         matplotlib.rcParams = dict(backend='Qt4Agg')
         matplotlib.rcParamsOrig = dict(backend='Qt4Agg')
 
         # Mock out functions
         self._save_am = pt.activate_matplotlib
         pt.activate_matplotlib = act_mpl
         self._save_ip = pt.import_pylab
         pt.import_pylab = lambda *a,**kw:None
         self._save_cis = backend_inline.configure_inline_support
         backend_inline.configure_inline_support = lambda *a, **kw: None
 
     def teardown(self):
         pt.activate_matplotlib = self._save_am
         pt.import_pylab = self._save_ip
         backend_inline.configure_inline_support = self._save_cis
         import matplotlib
         matplotlib.rcParams = self._saved_rcParams
         matplotlib.rcParamsOrig = self._saved_rcParamsOrig
 
     def test_qt(self):
-
         s = self.Shell()
         gui, backend = s.enable_matplotlib(None)
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
         gui, backend = s.enable_matplotlib("inline")
         assert gui == "inline"
         assert s.pylab_gui_select == "qt"
 
         gui, backend = s.enable_matplotlib("qt")
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
         gui, backend = s.enable_matplotlib("inline")
         assert gui == "inline"
         assert s.pylab_gui_select == "qt"
 
         gui, backend = s.enable_matplotlib()
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
     def test_inline(self):
         s = self.Shell()
         gui, backend = s.enable_matplotlib("inline")
         assert gui == "inline"
         assert s.pylab_gui_select == None
 
         gui, backend = s.enable_matplotlib("inline")
         assert gui == "inline"
         assert s.pylab_gui_select == None
 
         gui, backend = s.enable_matplotlib("qt")
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
     def test_inline_twice(self):
         "Using '%matplotlib inline' twice should not reset formatters"
 
         ip = self.Shell()
         gui, backend = ip.enable_matplotlib("inline")
         assert gui == "inline"
 
         fmts =  {'png'}
         active_mimes = {_fmt_mime_map[fmt] for fmt in fmts}
         pt.select_figure_formats(ip, fmts)
 
         gui, backend = ip.enable_matplotlib("inline")
         assert gui == "inline"
 
         for mime, f in ip.display_formatter.formatters.items():
             if mime in active_mimes:
                 assert Figure in f
             else:
                 assert Figure not in f
 
     def test_qt_gtk(self):
         s = self.Shell()
         gui, backend = s.enable_matplotlib("qt")
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
         gui, backend = s.enable_matplotlib("gtk")
         assert gui == "qt"
         assert s.pylab_gui_select == "qt"
 
 
 def test_no_gui_backends():
     for k in ['agg', 'svg', 'pdf', 'ps']:
         assert k not in pt.backend2gui
 
 
 def test_figure_no_canvas():
     fig = Figure()
     fig.canvas = None
     pt.print_figure(fig)

IPython/core/ultratb.py

@@ -1,1207 +1,1377 b''
 # -*- coding: utf-8 -*-
 """
 Verbose and colourful traceback formatting.
 
 **ColorTB**
 
 I've always found it a bit hard to visually parse tracebacks in Python.  The
 ColorTB class is a solution to that problem.  It colors the different parts of a
 traceback in a manner similar to what you would expect from a syntax-highlighting
 text editor.
 
 Installation instructions for ColorTB::
 
     import sys,ultratb
     sys.excepthook = ultratb.ColorTB()
 
 **VerboseTB**
 
 I've also included a port of Ka-Ping Yee's "cgitb.py" that produces all kinds
 of useful info when a traceback occurs.  Ping originally had it spit out HTML
 and intended it for CGI programmers, but why should they have all the fun?  I
 altered it to spit out colored text to the terminal.  It's a bit overwhelming,
 but kind of neat, and maybe useful for long-running programs that you believe
 are bug-free.  If a crash *does* occur in that type of program you want details.
 Give it a shot--you'll love it or you'll hate it.
 
 .. note::
 
   The Verbose mode prints the variables currently visible where the exception
   happened (shortening their strings if too long). This can potentially be
   very slow, if you happen to have a huge data structure whose string
   representation is complex to compute. Your computer may appear to freeze for
   a while with cpu usage at 100%. If this occurs, you can cancel the traceback
   with Ctrl-C (maybe hitting it more than once).
 
   If you encounter this kind of situation often, you may want to use the
   Verbose_novars mode instead of the regular Verbose, which avoids formatting
   variables (but otherwise includes the information and context given by
   Verbose).
 
 .. note::
 
   The verbose mode print all variables in the stack, which means it can
   potentially leak sensitive information like access keys, or unencrypted
   password.
 
 Installation instructions for VerboseTB::
 
     import sys,ultratb
     sys.excepthook = ultratb.VerboseTB()
 
 Note:  Much of the code in this module was lifted verbatim from the standard
 library module 'traceback.py' and Ka-Ping Yee's 'cgitb.py'.
 
 Color schemes
 -------------
 
 The colors are defined in the class TBTools through the use of the
 ColorSchemeTable class. Currently the following exist:
 
   - NoColor: allows all of this module to be used in any terminal (the color
     escapes are just dummy blank strings).
 
   - Linux: is meant to look good in a terminal like the Linux console (black
     or very dark background).
 
   - LightBG: similar to Linux but swaps dark/light colors to be more readable
     in light background terminals.
 
   - Neutral: a neutral color scheme that should be readable on both light and
     dark background
 
 You can implement other color schemes easily, the syntax is fairly
 self-explanatory. Please send back new schemes you develop to the author for
 possible inclusion in future releases.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.core.ultratb
    :parts: 3
 """
 
 #*****************************************************************************
 # Copyright (C) 2001 Nathaniel Gray <[e-mail]>
 # Copyright (C) 2001-2004 Fernando Perez <[e-mail]>
 #
 # Distributed under the terms of the BSD License.  The full license is in
 # the file COPYING, distributed as part of this software.
 #*****************************************************************************
 
 
 import inspect
 import linecache
 import pydoc
 import sys
 import time
 import traceback
 from types import TracebackType
 from typing import Tuple, List, Any, Optional
 
 import stack_data
+from stack_data import FrameInfo as SDFrameInfo
 from pygments.formatters.terminal256 import Terminal256Formatter
 from pygments.styles import get_style_by_name
 
 # IPython's own modules
 from IPython import get_ipython
 from IPython.core import debugger
 from IPython.core.display_trap import DisplayTrap
 from IPython.core.excolors import exception_colors
+from IPython.utils import PyColorize
 from IPython.utils import path as util_path
 from IPython.utils import py3compat
 from IPython.utils.terminal import get_terminal_size
 
 import IPython.utils.colorable as colorable
 
 # Globals
 # amount of space to put line numbers before verbose tracebacks
 INDENT_SIZE = 8
 
 # Default color scheme.  This is used, for example, by the traceback
 # formatter.  When running in an actual IPython instance, the user's rc.colors
 # value is used, but having a module global makes this functionality available
 # to users of ultratb who are NOT running inside ipython.
 DEFAULT_SCHEME = 'NoColor'
+FAST_THRESHOLD = 10_000
 
 # ---------------------------------------------------------------------------
 # Code begins
 
 # Helper function -- largely belongs to VerboseTB, but we need the same
 # functionality to produce a pseudo verbose TB for SyntaxErrors, so that they
 # can be recognized properly by ipython.el's py-traceback-line-re
 # (SyntaxErrors have to be treated specially because they have no traceback)
 
 
 def _format_traceback_lines(lines, Colors, has_colors: bool, lvals):
     """
     Format tracebacks lines with pointing arrow, leading numbers...
 
     Parameters
     ----------
     lines : list[Line]
     Colors
         ColorScheme used.
     lvals : str
         Values of local variables, already colored, to inject just after the error line.
     """
     numbers_width = INDENT_SIZE - 1
     res = []
 
     for stack_line in lines:
         if stack_line is stack_data.LINE_GAP:
             res.append('%s   (...)%s\n' % (Colors.linenoEm, Colors.Normal))
             continue
 
         line = stack_line.render(pygmented=has_colors).rstrip('\n') + '\n'
         lineno = stack_line.lineno
         if stack_line.is_current:
             # This is the line with the error
             pad = numbers_width - len(str(lineno))
             num = '%s%s' % (debugger.make_arrow(pad), str(lineno))
             start_color = Colors.linenoEm
         else:
             num = '%*s' % (numbers_width, lineno)
             start_color = Colors.lineno
 
         line = '%s%s%s %s' % (start_color, num, Colors.Normal, line)
 
         res.append(line)
         if lvals and stack_line.is_current:
             res.append(lvals + '\n')
     return res
 
+def _simple_format_traceback_lines(lnum, index, lines, Colors, lvals, _line_format):
+    """
+    Format tracebacks lines with pointing arrow, leading numbers...
+
+    Parameters
+    ==========
+
+    lnum: int
+        number of the target line of code.
+    index: int
+        which line in the list should be highlighted.
+    lines: list[string]
+    Colors:
+        ColorScheme used.
+    lvals: bytes
+        Values of local variables, already colored, to inject just after the error line.
+    _line_format: f (str) -> (str, bool)
+        return (colorized version of str, failure to do so)
+    """
+    numbers_width = INDENT_SIZE - 1
+    res = []
+
+    for i, line in enumerate(lines, lnum - index):
+        line = py3compat.cast_unicode(line)
+
+        new_line, err = _line_format(line, "str")
+        if not err:
+            line = new_line
+
+        if i == lnum:
+            # This is the line with the error
+            pad = numbers_width - len(str(i))
+            num = "%s%s" % (debugger.make_arrow(pad), str(lnum))
+            line = "%s%s%s %s%s" % (
+                Colors.linenoEm,
+                num,
+                Colors.line,
+                line,
+                Colors.Normal,
+            )
+        else:
+            num = "%*s" % (numbers_width, i)
+            line = "%s%s%s %s" % (Colors.lineno, num, Colors.Normal, line)
+
+        res.append(line)
+        if lvals and i == lnum:
+            res.append(lvals + "\n")
+    return res
+
 
 def _format_filename(file, ColorFilename, ColorNormal, *, lineno=None):
     """
     Format filename lines with custom formatting from caching compiler or `File *.py` by default
 
     Parameters
     ----------
     file : str
     ColorFilename
         ColorScheme's filename coloring to be used.
     ColorNormal
         ColorScheme's normal coloring to be used.
     """
     ipinst = get_ipython()
     if (
         ipinst is not None
         and (data := ipinst.compile.format_code_name(file)) is not None
     ):
         label, name = data
         if lineno is None:
             tpl_link = f"{{label}} {ColorFilename}{{name}}{ColorNormal}"
         else:
             tpl_link = (
                 f"{{label}} {ColorFilename}{{name}}, line {{lineno}}{ColorNormal}"
             )
     else:
         label = "File"
         name = util_path.compress_user(
             py3compat.cast_unicode(file, util_path.fs_encoding)
         )
         if lineno is None:
             tpl_link = f"{{label}} {ColorFilename}{{name}}{ColorNormal}"
         else:
             # can we make this the more friendly ", line {{lineno}}", or do we need to preserve the formatting with the colon?
             tpl_link = f"{{label}} {ColorFilename}{{name}}:{{lineno}}{ColorNormal}"
 
     return tpl_link.format(label=label, name=name, lineno=lineno)
 
 #---------------------------------------------------------------------------
 # Module classes
 class TBTools(colorable.Colorable):
     """Basic tools used by all traceback printer classes."""
 
     # Number of frames to skip when reporting tracebacks
     tb_offset = 0
 
     def __init__(
         self,
         color_scheme="NoColor",
         call_pdb=False,
         ostream=None,
         parent=None,
         config=None,
         *,
         debugger_cls=None,
     ):
         # Whether to call the interactive pdb debugger after printing
         # tracebacks or not
         super(TBTools, self).__init__(parent=parent, config=config)
         self.call_pdb = call_pdb
 
         # Output stream to write to.  Note that we store the original value in
         # a private attribute and then make the public ostream a property, so
         # that we can delay accessing sys.stdout until runtime.  The way
         # things are written now, the sys.stdout object is dynamically managed
         # so a reference to it should NEVER be stored statically.  This
         # property approach confines this detail to a single location, and all
         # subclasses can simply access self.ostream for writing.
         self._ostream = ostream
 
         # Create color table
         self.color_scheme_table = exception_colors()
 
         self.set_colors(color_scheme)
         self.old_scheme = color_scheme  # save initial value for toggles
         self.debugger_cls = debugger_cls or debugger.Pdb
 
         if call_pdb:
             self.pdb = self.debugger_cls()
         else:
             self.pdb = None
 
     def _get_ostream(self):
         """Output stream that exceptions are written to.
 
         Valid values are:
 
         - None: the default, which means that IPython will dynamically resolve
           to sys.stdout.  This ensures compatibility with most tools, including
           Windows (where plain stdout doesn't recognize ANSI escapes).
 
         - Any object with 'write' and 'flush' attributes.
         """
         return sys.stdout if self._ostream is None else self._ostream
 
     def _set_ostream(self, val):
         assert val is None or (hasattr(val, 'write') and hasattr(val, 'flush'))
         self._ostream = val
 
     ostream = property(_get_ostream, _set_ostream)
 
     @staticmethod
     def _get_chained_exception(exception_value):
         cause = getattr(exception_value, "__cause__", None)
         if cause:
             return cause
         if getattr(exception_value, "__suppress_context__", False):
             return None
         return getattr(exception_value, "__context__", None)
 
     def get_parts_of_chained_exception(
         self, evalue
     ) -> Optional[Tuple[type, BaseException, TracebackType]]:
-
         chained_evalue = self._get_chained_exception(evalue)
 
         if chained_evalue:
             return chained_evalue.__class__, chained_evalue, chained_evalue.__traceback__
         return None
 
     def prepare_chained_exception_message(self, cause) -> List[Any]:
         direct_cause = "\nThe above exception was the direct cause of the following exception:\n"
         exception_during_handling = "\nDuring handling of the above exception, another exception occurred:\n"
 
         if cause:
             message = [[direct_cause]]
         else:
             message = [[exception_during_handling]]
         return message
 
     @property
     def has_colors(self) -> bool:
         return self.color_scheme_table.active_scheme_name.lower() != "nocolor"
 
     def set_colors(self, *args, **kw):
         """Shorthand access to the color table scheme selector method."""
 
         # Set own color table
         self.color_scheme_table.set_active_scheme(*args, **kw)
         # for convenience, set Colors to the active scheme
         self.Colors = self.color_scheme_table.active_colors
         # Also set colors of debugger
         if hasattr(self, 'pdb') and self.pdb is not None:
             self.pdb.set_colors(*args, **kw)
 
     def color_toggle(self):
         """Toggle between the currently active color scheme and NoColor."""
 
         if self.color_scheme_table.active_scheme_name == 'NoColor':
             self.color_scheme_table.set_active_scheme(self.old_scheme)
             self.Colors = self.color_scheme_table.active_colors
         else:
             self.old_scheme = self.color_scheme_table.active_scheme_name
             self.color_scheme_table.set_active_scheme('NoColor')
             self.Colors = self.color_scheme_table.active_colors
 
     def stb2text(self, stb):
         """Convert a structured traceback (a list) to a string."""
         return '\n'.join(stb)
 
     def text(self, etype, value, tb, tb_offset: Optional[int] = None, context=5):
         """Return formatted traceback.
 
         Subclasses may override this if they add extra arguments.
         """
         tb_list = self.structured_traceback(etype, value, tb,
                                             tb_offset, context)
         return self.stb2text(tb_list)
 
     def structured_traceback(
         self, etype, evalue, tb, tb_offset: Optional[int] = None, context=5, mode=None
     ):
         """Return a list of traceback frames.
 
         Must be implemented by each class.
         """
         raise NotImplementedError()
 
 
 #---------------------------------------------------------------------------
 class ListTB(TBTools):
     """Print traceback information from a traceback list, with optional color.
 
     Calling requires 3 arguments: (etype, evalue, elist)
     as would be obtained by::
 
       etype, evalue, tb = sys.exc_info()
       if tb:
         elist = traceback.extract_tb(tb)
       else:
         elist = None
 
     It can thus be used by programs which need to process the traceback before
     printing (such as console replacements based on the code module from the
     standard library).
 
     Because they are meant to be called without a full traceback (only a
     list), instances of this class can't call the interactive pdb debugger."""
 
 
     def __call__(self, etype, value, elist):
         self.ostream.flush()
         self.ostream.write(self.text(etype, value, elist))
         self.ostream.write('\n')
 
     def _extract_tb(self, tb):
         if tb:
             return traceback.extract_tb(tb)
         else:
             return None
 
     def structured_traceback(
         self,
         etype: type,
         evalue: BaseException,
         etb: Optional[TracebackType] = None,
         tb_offset: Optional[int] = None,
         context=5,
     ):
         """Return a color formatted string with the traceback info.
 
         Parameters
         ----------
         etype : exception type
             Type of the exception raised.
         evalue : object
             Data stored in the exception
         etb : list | TracebackType | None
             If list: List of frames, see class docstring for details.
             If Traceback: Traceback of the exception.
         tb_offset : int, optional
             Number of frames in the traceback to skip.  If not given, the
             instance evalue is used (set in constructor).
         context : int, optional
             Number of lines of context information to print.
 
         Returns
         -------
         String with formatted exception.
         """
         # This is a workaround to get chained_exc_ids in recursive calls
         # etb should not be a tuple if structured_traceback is not recursive
         if isinstance(etb, tuple):
             etb, chained_exc_ids = etb
         else:
             chained_exc_ids = set()
 
         if isinstance(etb, list):
             elist = etb
         elif etb is not None:
             elist = self._extract_tb(etb)
         else:
             elist = []
         tb_offset = self.tb_offset if tb_offset is None else tb_offset
         assert isinstance(tb_offset, int)
         Colors = self.Colors
         out_list = []
         if elist:
 
             if tb_offset and len(elist) > tb_offset:
                 elist = elist[tb_offset:]
 
             out_list.append('Traceback %s(most recent call last)%s:' %
                             (Colors.normalEm, Colors.Normal) + '\n')
             out_list.extend(self._format_list(elist))
         # The exception info should be a single entry in the list.
         lines = ''.join(self._format_exception_only(etype, evalue))
         out_list.append(lines)
 
         exception = self.get_parts_of_chained_exception(evalue)
 
         if exception and not id(exception[1]) in chained_exc_ids:
             chained_exception_message = self.prepare_chained_exception_message(
                 evalue.__cause__)[0]
             etype, evalue, etb = exception
             # Trace exception to avoid infinite 'cause' loop
             chained_exc_ids.add(id(exception[1]))
             chained_exceptions_tb_offset = 0
             out_list = (
                 self.structured_traceback(
                     etype, evalue, (etb, chained_exc_ids),
                     chained_exceptions_tb_offset, context)
                 + chained_exception_message
                 + out_list)
 
         return out_list
 
     def _format_list(self, extracted_list):
         """Format a list of traceback entry tuples for printing.
 
         Given a list of tuples as returned by extract_tb() or
         extract_stack(), return a list of strings ready for printing.
         Each string in the resulting list corresponds to the item with the
         same index in the argument list.  Each string ends in a newline;
         the strings may contain internal newlines as well, for those items
         whose source text line is not None.
 
         Lifted almost verbatim from traceback.py
         """
 
         Colors = self.Colors
         list = []
         for ind, (filename, lineno, name, line) in enumerate(extracted_list):
             normalCol, nameCol, fileCol, lineCol = (
                 # Emphasize the last entry
                 (Colors.normalEm, Colors.nameEm, Colors.filenameEm, Colors.line)
                 if ind == len(extracted_list) - 1
                 else (Colors.Normal, Colors.name, Colors.filename, "")
             )
 
             fns = _format_filename(filename, fileCol, normalCol, lineno=lineno)
             item = f"{normalCol}  {fns}"
 
             if name != "<module>":
                 item += f" in {nameCol}{name}{normalCol}\n"
             else:
                 item += "\n"
             if line:
                 item += f"{lineCol}    {line.strip()}{normalCol}\n"
             list.append(item)
 
         return list
 
     def _format_exception_only(self, etype, value):
         """Format the exception part of a traceback.
 
         The arguments are the exception type and value such as given by
         sys.exc_info()[:2]. The return value is a list of strings, each ending
         in a newline.  Normally, the list contains a single string; however,
         for SyntaxError exceptions, it contains several lines that (when
         printed) display detailed information about where the syntax error
         occurred.  The message indicating which exception occurred is the
         always last string in the list.
 
         Also lifted nearly verbatim from traceback.py
         """
         have_filedata = False
         Colors = self.Colors
         list = []
         stype = py3compat.cast_unicode(Colors.excName + etype.__name__ + Colors.Normal)
         if value is None:
             # Not sure if this can still happen in Python 2.6 and above
             list.append(stype + '\n')
         else:
             if issubclass(etype, SyntaxError):
                 have_filedata = True
                 if not value.filename: value.filename = "<string>"
                 if value.lineno:
                     lineno = value.lineno
                     textline = linecache.getline(value.filename, value.lineno)
                 else:
                     lineno = "unknown"
                     textline = ""
                 list.append(
                     "%s  %s%s\n"
                     % (
                         Colors.normalEm,
                         _format_filename(
                             value.filename,
                             Colors.filenameEm,
                             Colors.normalEm,
                             lineno=(None if lineno == "unknown" else lineno),
                         ),
                         Colors.Normal,
                     )
                 )
                 if textline == "":
                     textline = py3compat.cast_unicode(value.text, "utf-8")
 
                 if textline is not None:
                     i = 0
                     while i < len(textline) and textline[i].isspace():
                         i += 1
                     list.append('%s    %s%s\n' % (Colors.line,
                                                   textline.strip(),
                                                   Colors.Normal))
                     if value.offset is not None:
                         s = '    '
                         for c in textline[i:value.offset - 1]:
                             if c.isspace():
                                 s += c
                             else:
                                 s += ' '
                         list.append('%s%s^%s\n' % (Colors.caret, s,
                                                    Colors.Normal))
 
             try:
                 s = value.msg
             except Exception:
                 s = self._some_str(value)
             if s:
                 list.append('%s%s:%s %s\n' % (stype, Colors.excName,
                                               Colors.Normal, s))
             else:
                 list.append('%s\n' % stype)
 
         # sync with user hooks
         if have_filedata:
             ipinst = get_ipython()
             if ipinst is not None:
                 ipinst.hooks.synchronize_with_editor(value.filename, value.lineno, 0)
 
         return list
 
     def get_exception_only(self, etype, value):
         """Only print the exception type and message, without a traceback.
 
         Parameters
         ----------
         etype : exception type
         value : exception value
         """
         return ListTB.structured_traceback(self, etype, value)
 
     def show_exception_only(self, etype, evalue):
         """Only print the exception type and message, without a traceback.
 
         Parameters
         ----------
         etype : exception type
         evalue : exception value
         """
         # This method needs to use __call__ from *this* class, not the one from
         # a subclass whose signature or behavior may be different
         ostream = self.ostream
         ostream.flush()
         ostream.write('\n'.join(self.get_exception_only(etype, evalue)))
         ostream.flush()
 
     def _some_str(self, value):
         # Lifted from traceback.py
         try:
             return py3compat.cast_unicode(str(value))
         except:
             return u'<unprintable %s object>' % type(value).__name__
 
 
-#----------------------------------------------------------------------------
+class FrameInfo:
+    """
+    Mirror of stack data's FrameInfo, but so that we can bypass highlighting on
+    really long frames.
+    """
+
+    description: Optional[str]
+    filename: str
+    lineno: int
+
+    @classmethod
+    def _from_stack_data_FrameInfo(cls, frame_info):
+        return cls(
+            getattr(frame_info, "description", None),
+            getattr(frame_info, "filename", None),
+            getattr(frame_info, "lineno", None),
+            getattr(frame_info, "frame", None),
+            getattr(frame_info, "code", None),
+            sd=frame_info,
+        )
+
+    def __init__(self, description, filename, lineno, frame, code, sd=None):
+        self.description = description
+        self.filename = filename
+        self.lineno = lineno
+        self.frame = frame
+        self.code = code
+        self._sd = sd
+
+        # self.lines = []
+        if sd is None:
+            ix = inspect.getsourcelines(frame)
+            self.raw_lines = ix[0]
+
+    @property
+    def variables_in_executing_piece(self):
+        if self._sd:
+            return self._sd.variables_in_executing_piece
+        else:
+            return []
+
+    @property
+    def lines(self):
+        return self._sd.lines
+
+    @property
+    def executing(self):
+        if self._sd:
+            return self._sd.executing
+        else:
+            return None
+
+
+# ----------------------------------------------------------------------------
 class VerboseTB(TBTools):
     """A port of Ka-Ping Yee's cgitb.py module that outputs color text instead
     of HTML.  Requires inspect and pydoc.  Crazy, man.
 
     Modified version which optionally strips the topmost entries from the
     traceback, to be used with alternate interpreters (because their own code
     would appear in the traceback)."""
 
     _tb_highlight = "bg:ansiyellow"
 
     def __init__(
         self,
         color_scheme: str = "Linux",
         call_pdb: bool = False,
         ostream=None,
         tb_offset: int = 0,
         long_header: bool = False,
         include_vars: bool = True,
         check_cache=None,
         debugger_cls=None,
         parent=None,
         config=None,
     ):
         """Specify traceback offset, headers and color scheme.
 
         Define how many frames to drop from the tracebacks. Calling it with
         tb_offset=1 allows use of this handler in interpreters which will have
         their own code at the top of the traceback (VerboseTB will first
         remove that frame before printing the traceback info)."""
         TBTools.__init__(
             self,
             color_scheme=color_scheme,
             call_pdb=call_pdb,
             ostream=ostream,
             parent=parent,
             config=config,
             debugger_cls=debugger_cls,
         )
         self.tb_offset = tb_offset
         self.long_header = long_header
         self.include_vars = include_vars
         # By default we use linecache.checkcache, but the user can provide a
         # different check_cache implementation.  This was formerly used by the
         # IPython kernel for interactive code, but is no longer necessary.
         if check_cache is None:
             check_cache = linecache.checkcache
         self.check_cache = check_cache
 
         self.skip_hidden = True
 
-    def format_record(self, frame_info):
+    def format_record(self, frame_info: FrameInfo):
         """Format a single stack frame"""
+        assert isinstance(frame_info, FrameInfo)
         Colors = self.Colors  # just a shorthand + quicker name lookup
         ColorsNormal = Colors.Normal  # used a lot
 
-        if isinstance(frame_info, stack_data.RepeatedFrames):
+        if isinstance(frame_info._sd, stack_data.RepeatedFrames):
             return '    %s[... skipping similar frames: %s]%s\n' % (
                 Colors.excName, frame_info.description, ColorsNormal)
 
         indent = " " * INDENT_SIZE
         em_normal = "%s\n%s%s" % (Colors.valEm, indent, ColorsNormal)
         tpl_call = f"in {Colors.vName}{{file}}{Colors.valEm}{{scope}}{ColorsNormal}"
         tpl_call_fail = "in %s%%s%s(***failed resolving arguments***)%s" % (
             Colors.vName,
             Colors.valEm,
             ColorsNormal,
         )
         tpl_name_val = "%%s %s= %%s%s" % (Colors.valEm, ColorsNormal)
 
         link = _format_filename(
             frame_info.filename,
             Colors.filenameEm,
             ColorsNormal,
             lineno=frame_info.lineno,
         )
         args, varargs, varkw, locals_ = inspect.getargvalues(frame_info.frame)
-
-        func = frame_info.executing.code_qualname()
+        if frame_info.executing is not None:
+            func = frame_info.executing.code_qualname()
+        else:
+            func = "?"
         if func == "<module>":
             call = ""
         else:
             # Decide whether to include variable details or not
             var_repr = eqrepr if self.include_vars else nullrepr
             try:
                 scope = inspect.formatargvalues(
                     args, varargs, varkw, locals_, formatvalue=var_repr
                 )
                 call = tpl_call.format(file=func, scope=scope)
             except KeyError:
                 # This happens in situations like errors inside generator
                 # expressions, where local variables are listed in the
                 # line, but can't be extracted from the frame.  I'm not
                 # 100% sure this isn't actually a bug in inspect itself,
                 # but since there's no info for us to compute with, the
                 # best we can do is report the failure and move on.  Here
                 # we must *not* call any traceback construction again,
                 # because that would mess up use of %debug later on.  So we
                 # simply report the failure and move on.  The only
                 # limitation will be that this frame won't have locals
                 # listed in the call signature.  Quite subtle problem...
                 # I can't think of a good way to validate this in a unit
                 # test, but running a script consisting of:
                 #  dict( (k,v.strip()) for (k,v) in range(10) )
                 # will illustrate the error, if this exception catch is
                 # disabled.
                 call = tpl_call_fail % func
 
         lvals = ''
         lvals_list = []
         if self.include_vars:
             try:
                 # we likely want to fix stackdata at some point, but
                 # still need a workaround.
                 fibp = frame_info.variables_in_executing_piece
                 for var in fibp:
                     lvals_list.append(tpl_name_val % (var.name, repr(var.value)))
             except Exception:
                 lvals_list.append(
                     "Exception trying to inspect frame. No more locals available."
                 )
         if lvals_list:
             lvals = '%s%s' % (indent, em_normal.join(lvals_list))
 
         result = f'{link}{", " if call else ""}{call}\n'
-
-        result += ''.join(_format_traceback_lines(frame_info.lines, Colors, self.has_colors, lvals))
+        if frame_info._sd is None:
+            assert False
+            # fast fallback if file is too long
+            tpl_link = "%s%%s%s" % (Colors.filenameEm, ColorsNormal)
+            link = tpl_link % util_path.compress_user(frame_info.filename)
+            level = "%s %s\n" % (link, call)
+            _line_format = PyColorize.Parser(
+                style=self.color_scheme_table.active_scheme_name, parent=self
+            ).format2
+            first_line = frame_info.code.co_firstlineno
+            current_line = frame_info.lineno[0]
+            return "%s%s" % (
+                level,
+                "".join(
+                    _simple_format_traceback_lines(
+                        current_line,
+                        current_line - first_line,
+                        frame_info.raw_lines,
+                        Colors,
+                        lvals,
+                        _line_format,
+                    )
+                ),
+            )
+            # result += "\n".join(frame_info.raw_lines)
+        else:
+            result += "".join(
+                _format_traceback_lines(
+                    frame_info.lines, Colors, self.has_colors, lvals
+                )
+            )
         return result
 
     def prepare_header(self, etype, long_version=False):
         colors = self.Colors  # just a shorthand + quicker name lookup
         colorsnormal = colors.Normal  # used a lot
         exc = '%s%s%s' % (colors.excName, etype, colorsnormal)
         width = min(75, get_terminal_size()[0])
         if long_version:
             # Header with the exception type, python version, and date
             pyver = 'Python ' + sys.version.split()[0] + ': ' + sys.executable
             date = time.ctime(time.time())
 
             head = '%s%s%s\n%s%s%s\n%s' % (colors.topline, '-' * width, colorsnormal,
                                            exc, ' ' * (width - len(str(etype)) - len(pyver)),
                                            pyver, date.rjust(width) )
             head += "\nA problem occurred executing Python code.  Here is the sequence of function" \
                     "\ncalls leading up to the error, with the most recent (innermost) call last."
         else:
             # Simplified header
             head = '%s%s' % (exc, 'Traceback (most recent call last)'. \
                              rjust(width - len(str(etype))) )
 
         return head
 
     def format_exception(self, etype, evalue):
         colors = self.Colors  # just a shorthand + quicker name lookup
         colorsnormal = colors.Normal  # used a lot
         # Get (safely) a string form of the exception info
         try:
             etype_str, evalue_str = map(str, (etype, evalue))
         except:
             # User exception is improperly defined.
             etype, evalue = str, sys.exc_info()[:2]
             etype_str, evalue_str = map(str, (etype, evalue))
         # ... and format it
         return ['%s%s%s: %s' % (colors.excName, etype_str,
                                 colorsnormal, py3compat.cast_unicode(evalue_str))]
 
     def format_exception_as_a_whole(
         self,
         etype: type,
         evalue: BaseException,
         etb: Optional[TracebackType],
         number_of_lines_of_context,
         tb_offset: Optional[int],
     ):
         """Formats the header, traceback and exception message for a single exception.
 
         This may be called multiple times by Python 3 exception chaining
         (PEP 3134).
         """
         # some locals
         orig_etype = etype
         try:
             etype = etype.__name__
         except AttributeError:
             pass
 
         tb_offset = self.tb_offset if tb_offset is None else tb_offset
         assert isinstance(tb_offset, int)
         head = self.prepare_header(etype, self.long_header)
         records = (
             self.get_records(etb, number_of_lines_of_context, tb_offset) if etb else []
         )
 
         frames = []
         skipped = 0
         lastrecord = len(records) - 1
-        for i, r in enumerate(records):
-            if not isinstance(r, stack_data.RepeatedFrames) and self.skip_hidden:
-                if r.frame.f_locals.get("__tracebackhide__", 0) and i != lastrecord:
+        for i, record in enumerate(records):
+            if (
+                not isinstance(record._sd, stack_data.RepeatedFrames)
+                and self.skip_hidden
+            ):
+                if (
+                    record.frame.f_locals.get("__tracebackhide__", 0)
+                    and i != lastrecord
+                ):
                     skipped += 1
                     continue
             if skipped:
                 Colors = self.Colors  # just a shorthand + quicker name lookup
                 ColorsNormal = Colors.Normal  # used a lot
                 frames.append(
                     "    %s[... skipping hidden %s frame]%s\n"
                     % (Colors.excName, skipped, ColorsNormal)
                 )
                 skipped = 0
-            frames.append(self.format_record(r))
+            frames.append(self.format_record(record))
         if skipped:
             Colors = self.Colors  # just a shorthand + quicker name lookup
             ColorsNormal = Colors.Normal  # used a lot
             frames.append(
                 "    %s[... skipping hidden %s frame]%s\n"
                 % (Colors.excName, skipped, ColorsNormal)
             )
 
         formatted_exception = self.format_exception(etype, evalue)
         if records:
             frame_info = records[-1]
             ipinst = get_ipython()
             if ipinst is not None:
                 ipinst.hooks.synchronize_with_editor(frame_info.filename, frame_info.lineno, 0)
 
         return [[head] + frames + [''.join(formatted_exception[0])]]
 
     def get_records(
         self, etb: TracebackType, number_of_lines_of_context: int, tb_offset: int
     ):
         assert etb is not None
         context = number_of_lines_of_context - 1
         after = context // 2
         before = context - after
         if self.has_colors:
             style = get_style_by_name("default")
             style = stack_data.style_with_executing_node(style, self._tb_highlight)
             formatter = Terminal256Formatter(style=style)
         else:
             formatter = None
         options = stack_data.Options(
             before=before,
             after=after,
             pygments_formatter=formatter,
         )
-        return list(stack_data.FrameInfo.stack_data(etb, options=options))[tb_offset:]
+
+        # let's estimate the amount of code we eill have to parse/highlight.
+        cf = etb
+        max_len = 0
+        tbs = []
+        while cf is not None:
+            source_file = inspect.getsourcefile(etb.tb_frame)
+            lines, first = inspect.getsourcelines(etb.tb_frame)
+            max_len = max(max_len, first + len(lines))
+            tbs.append(cf)
+            cf = cf.tb_next
+
+        if max_len > FAST_THRESHOLD:
+            FIs = []
+            for tb in tbs:
+                frame = tb.tb_frame
+                lineno = (frame.f_lineno,)
+                code = frame.f_code
+                filename = code.co_filename
+                FIs.append(FrameInfo("Raw frame", filename, lineno, frame, code))
+            return FIs
+        res = list(stack_data.FrameInfo.stack_data(etb, options=options))[tb_offset:]
+        res = [FrameInfo._from_stack_data_FrameInfo(r) for r in res]
+        return res
 
     def structured_traceback(
         self,
         etype: type,
         evalue: Optional[BaseException],
         etb: Optional[TracebackType],
         tb_offset: Optional[int] = None,
         number_of_lines_of_context: int = 5,
     ):
         """Return a nice text document describing the traceback."""
         formatted_exception = self.format_exception_as_a_whole(etype, evalue, etb, number_of_lines_of_context,
                                                                tb_offset)
 
         colors = self.Colors  # just a shorthand + quicker name lookup
         colorsnormal = colors.Normal  # used a lot
         head = '%s%s%s' % (colors.topline, '-' * min(75, get_terminal_size()[0]), colorsnormal)
         structured_traceback_parts = [head]
         chained_exceptions_tb_offset = 0
         lines_of_context = 3
         formatted_exceptions = formatted_exception
         exception = self.get_parts_of_chained_exception(evalue)
         if exception:
             assert evalue is not None
             formatted_exceptions += self.prepare_chained_exception_message(evalue.__cause__)
             etype, evalue, etb = exception
         else:
             evalue = None
         chained_exc_ids = set()
         while evalue:
             formatted_exceptions += self.format_exception_as_a_whole(etype, evalue, etb, lines_of_context,
                                                                      chained_exceptions_tb_offset)
             exception = self.get_parts_of_chained_exception(evalue)
 
             if exception and not id(exception[1]) in chained_exc_ids:
                 chained_exc_ids.add(id(exception[1])) # trace exception to avoid infinite 'cause' loop
                 formatted_exceptions += self.prepare_chained_exception_message(evalue.__cause__)
                 etype, evalue, etb = exception
             else:
                 evalue = None
 
         # we want to see exceptions in a reversed order:
         # the first exception should be on top
         for formatted_exception in reversed(formatted_exceptions):
             structured_traceback_parts += formatted_exception
 
         return structured_traceback_parts
 
     def debugger(self, force: bool = False):
         """Call up the pdb debugger if desired, always clean up the tb
         reference.
 
         Keywords:
 
           - force(False): by default, this routine checks the instance call_pdb
             flag and does not actually invoke the debugger if the flag is false.
             The 'force' option forces the debugger to activate even if the flag
             is false.
 
         If the call_pdb flag is set, the pdb interactive debugger is
         invoked. In all cases, the self.tb reference to the current traceback
         is deleted to prevent lingering references which hamper memory
         management.
 
         Note that each call to pdb() does an 'import readline', so if your app
         requires a special setup for the readline completers, you'll have to
         fix that by hand after invoking the exception handler."""
 
         if force or self.call_pdb:
             if self.pdb is None:
                 self.pdb = self.debugger_cls()
             # the system displayhook may have changed, restore the original
             # for pdb
             display_trap = DisplayTrap(hook=sys.__displayhook__)
             with display_trap:
                 self.pdb.reset()
                 # Find the right frame so we don't pop up inside ipython itself
                 if hasattr(self, 'tb') and self.tb is not None:
                     etb = self.tb
                 else:
                     etb = self.tb = sys.last_traceback
                 while self.tb is not None and self.tb.tb_next is not None:
                     assert self.tb.tb_next is not None
                     self.tb = self.tb.tb_next
                 if etb and etb.tb_next:
                     etb = etb.tb_next
                 self.pdb.botframe = etb.tb_frame
                 self.pdb.interaction(None, etb)
 
         if hasattr(self, 'tb'):
             del self.tb
 
     def handler(self, info=None):
         (etype, evalue, etb) = info or sys.exc_info()
         self.tb = etb
         ostream = self.ostream
         ostream.flush()
         ostream.write(self.text(etype, evalue, etb))
         ostream.write('\n')
         ostream.flush()
 
     # Changed so an instance can just be called as VerboseTB_inst() and print
     # out the right info on its own.
     def __call__(self, etype=None, evalue=None, etb=None):
         """This hook can replace sys.excepthook (for Python 2.1 or higher)."""
         if etb is None:
             self.handler()
         else:
             self.handler((etype, evalue, etb))
         try:
             self.debugger()
         except KeyboardInterrupt:
             print("\nKeyboardInterrupt")
 
 
 #----------------------------------------------------------------------------
 class FormattedTB(VerboseTB, ListTB):
     """Subclass ListTB but allow calling with a traceback.
 
     It can thus be used as a sys.excepthook for Python > 2.1.
 
     Also adds 'Context' and 'Verbose' modes, not available in ListTB.
 
     Allows a tb_offset to be specified. This is useful for situations where
     one needs to remove a number of topmost frames from the traceback (such as
     occurs with python programs that themselves execute other python code,
     like Python shells).  """
 
     mode: str
 
     def __init__(self, mode='Plain', color_scheme='Linux', call_pdb=False,
                  ostream=None,
                  tb_offset=0, long_header=False, include_vars=False,
                  check_cache=None, debugger_cls=None,
                  parent=None, config=None):
 
         # NEVER change the order of this list. Put new modes at the end:
         self.valid_modes = ['Plain', 'Context', 'Verbose', 'Minimal']
         self.verbose_modes = self.valid_modes[1:3]
 
         VerboseTB.__init__(self, color_scheme=color_scheme, call_pdb=call_pdb,
                            ostream=ostream, tb_offset=tb_offset,
                            long_header=long_header, include_vars=include_vars,
                            check_cache=check_cache, debugger_cls=debugger_cls,
                            parent=parent, config=config)
 
         # Different types of tracebacks are joined with different separators to
         # form a single string.  They are taken from this dict
         self._join_chars = dict(Plain='', Context='\n', Verbose='\n',
                                 Minimal='')
         # set_mode also sets the tb_join_char attribute
         self.set_mode(mode)
 
     def structured_traceback(self, etype, value, tb, tb_offset=None, number_of_lines_of_context=5):
         tb_offset = self.tb_offset if tb_offset is None else tb_offset
         mode = self.mode
         if mode in self.verbose_modes:
             # Verbose modes need a full traceback
             return VerboseTB.structured_traceback(
                 self, etype, value, tb, tb_offset, number_of_lines_of_context
             )
         elif mode == 'Minimal':
             return ListTB.get_exception_only(self, etype, value)
         else:
             # We must check the source cache because otherwise we can print
             # out-of-date source code.
             self.check_cache()
             # Now we can extract and format the exception
             return ListTB.structured_traceback(
                 self, etype, value, tb, tb_offset, number_of_lines_of_context
             )
 
     def stb2text(self, stb):
         """Convert a structured traceback (a list) to a string."""
         return self.tb_join_char.join(stb)
 
     def set_mode(self, mode: Optional[str] = None):
         """Switch to the desired mode.
 
         If mode is not specified, cycles through the available modes."""
 
         if not mode:
             new_idx = (self.valid_modes.index(self.mode) + 1 ) % \
                       len(self.valid_modes)
             self.mode = self.valid_modes[new_idx]
         elif mode not in self.valid_modes:
             raise ValueError(
                 "Unrecognized mode in FormattedTB: <" + mode + ">\n"
                 "Valid modes: " + str(self.valid_modes)
             )
         else:
             assert isinstance(mode, str)
             self.mode = mode
         # include variable details only in 'Verbose' mode
         self.include_vars = (self.mode == self.valid_modes[2])
         # Set the join character for generating text tracebacks
         self.tb_join_char = self._join_chars[self.mode]
 
     # some convenient shortcuts
     def plain(self):
         self.set_mode(self.valid_modes[0])
 
     def context(self):
         self.set_mode(self.valid_modes[1])
 
     def verbose(self):
         self.set_mode(self.valid_modes[2])
 
     def minimal(self):
         self.set_mode(self.valid_modes[3])
 
 
 #----------------------------------------------------------------------------
 class AutoFormattedTB(FormattedTB):
     """A traceback printer which can be called on the fly.
 
     It will find out about exceptions by itself.
 
     A brief example::
 
         AutoTB = AutoFormattedTB(mode = 'Verbose',color_scheme='Linux')
         try:
           ...
         except:
           AutoTB()  # or AutoTB(out=logfile) where logfile is an open file object
     """
 
     def __call__(self, etype=None, evalue=None, etb=None,
                  out=None, tb_offset=None):
         """Print out a formatted exception traceback.
 
         Optional arguments:
           - out: an open file-like object to direct output to.
 
           - tb_offset: the number of frames to skip over in the stack, on a
           per-call basis (this overrides temporarily the instance's tb_offset
           given at initialization time."""
 
         if out is None:
             out = self.ostream
         out.flush()
         out.write(self.text(etype, evalue, etb, tb_offset))
         out.write('\n')
         out.flush()
         # FIXME: we should remove the auto pdb behavior from here and leave
         # that to the clients.
         try:
             self.debugger()
         except KeyboardInterrupt:
             print("\nKeyboardInterrupt")
 
-    def structured_traceback(self, etype=None, value=None, tb=None,
-                             tb_offset=None, number_of_lines_of_context=5):
-
+    def structured_traceback(
+        self,
+        etype=None,
+        value=None,
+        tb=None,
+        tb_offset=None,
+        number_of_lines_of_context=5,
+    ):
         etype: type
         value: BaseException
         # tb: TracebackType or tupleof tb types ?
         if etype is None:
             etype, value, tb = sys.exc_info()
         if isinstance(tb, tuple):
             # tb is a tuple if this is a chained exception.
             self.tb = tb[0]
         else:
             self.tb = tb
         return FormattedTB.structured_traceback(
             self, etype, value, tb, tb_offset, number_of_lines_of_context)
 
 
 #---------------------------------------------------------------------------
 
 # A simple class to preserve Nathan's original functionality.
 class ColorTB(FormattedTB):
     """Shorthand to initialize a FormattedTB in Linux colors mode."""
 
     def __init__(self, color_scheme='Linux', call_pdb=0, **kwargs):
         FormattedTB.__init__(self, color_scheme=color_scheme,
                              call_pdb=call_pdb, **kwargs)
 
 
 class SyntaxTB(ListTB):
     """Extension which holds some state: the last exception value"""
 
     def __init__(self, color_scheme='NoColor', parent=None, config=None):
         ListTB.__init__(self, color_scheme, parent=parent, config=config)
         self.last_syntax_error = None
 
     def __call__(self, etype, value, elist):
         self.last_syntax_error = value
 
         ListTB.__call__(self, etype, value, elist)
 
     def structured_traceback(self, etype, value, elist, tb_offset=None,
                              context=5):
         # If the source file has been edited, the line in the syntax error can
         # be wrong (retrieved from an outdated cache). This replaces it with
         # the current value.
         if isinstance(value, SyntaxError) \
                 and isinstance(value.filename, str) \
                 and isinstance(value.lineno, int):
             linecache.checkcache(value.filename)
             newtext = linecache.getline(value.filename, value.lineno)
             if newtext:
                 value.text = newtext
         self.last_syntax_error = value
         return super(SyntaxTB, self).structured_traceback(etype, value, elist,
                                                           tb_offset=tb_offset, context=context)
 
     def clear_err_state(self):
         """Return the current error state and clear it"""
         e = self.last_syntax_error
         self.last_syntax_error = None
         return e
 
     def stb2text(self, stb):
         """Convert a structured traceback (a list) to a string."""
         return ''.join(stb)
 
 
 # some internal-use functions
 def text_repr(value):
     """Hopefully pretty robust repr equivalent."""
     # this is pretty horrible but should always return *something*
     try:
         return pydoc.text.repr(value)
     except KeyboardInterrupt:
         raise
     except:
         try:
             return repr(value)
         except KeyboardInterrupt:
             raise
         except:
             try:
                 # all still in an except block so we catch
                 # getattr raising
                 name = getattr(value, '__name__', None)
                 if name:
                     # ick, recursion
                     return text_repr(name)
                 klass = getattr(value, '__class__', None)
                 if klass:
                     return '%s instance' % text_repr(klass)
             except KeyboardInterrupt:
                 raise
             except:
                 return 'UNRECOVERABLE REPR FAILURE'
 
 
 def eqrepr(value, repr=text_repr):
     return '=%s' % repr(value)
 
 
 def nullrepr(value, repr=text_repr):
     return ''

IPython/terminal/debugger.py

@@ -1,178 +1,177 b''
 import asyncio
 import os
 import sys
 
 from IPython.core.debugger import Pdb
 from IPython.core.completer import IPCompleter
 from .ptutils import IPythonPTCompleter
 from .shortcuts import create_ipython_shortcuts
 from . import embed
 
 from pathlib import Path
 from pygments.token import Token
 from prompt_toolkit.shortcuts.prompt import PromptSession
 from prompt_toolkit.enums import EditingMode
 from prompt_toolkit.formatted_text import PygmentsTokens
 from prompt_toolkit.history import InMemoryHistory, FileHistory
 from concurrent.futures import ThreadPoolExecutor
 
 from prompt_toolkit import __version__ as ptk_version
 PTK3 = ptk_version.startswith('3.')
 
 
 # we want to avoid ptk as much as possible when using subprocesses
 # as it uses cursor positioning requests, deletes color ....
 _use_simple_prompt = "IPY_TEST_SIMPLE_PROMPT" in os.environ
 
 
 class TerminalPdb(Pdb):
     """Standalone IPython debugger."""
 
     def __init__(self, *args, pt_session_options=None, **kwargs):
         Pdb.__init__(self, *args, **kwargs)
         self._ptcomp = None
         self.pt_init(pt_session_options)
         self.thread_executor = ThreadPoolExecutor(1)
 
     def pt_init(self, pt_session_options=None):
         """Initialize the prompt session and the prompt loop
         and store them in self.pt_app and self.pt_loop.
 
         Additional keyword arguments for the PromptSession class
         can be specified in pt_session_options.
         """
         if pt_session_options is None:
             pt_session_options = {}
 
         def get_prompt_tokens():
             return [(Token.Prompt, self.prompt)]
 
         if self._ptcomp is None:
             compl = IPCompleter(
                 shell=self.shell, namespace={}, global_namespace={}, parent=self.shell
             )
             # add a completer for all the do_ methods
             methods_names = [m[3:] for m in dir(self) if m.startswith("do_")]
 
             def gen_comp(self, text):
                 return [m for m in methods_names if m.startswith(text)]
             import types
             newcomp = types.MethodType(gen_comp, compl)
             compl.custom_matchers.insert(0, newcomp)
             # end add completer.
 
             self._ptcomp = IPythonPTCompleter(compl)
 
         # setup history only when we start pdb
         if self.shell.debugger_history is None:
             if self.shell.debugger_history_file is not None:
-
                 p = Path(self.shell.debugger_history_file).expanduser()
                 if not p.exists():
                     p.touch()
                 self.debugger_history = FileHistory(os.path.expanduser(str(p)))
             else:
                 self.debugger_history = InMemoryHistory()
         else:
             self.debugger_history = self.shell.debugger_history
 
         options = dict(
             message=(lambda: PygmentsTokens(get_prompt_tokens())),
             editing_mode=getattr(EditingMode, self.shell.editing_mode.upper()),
             key_bindings=create_ipython_shortcuts(self.shell),
             history=self.debugger_history,
             completer=self._ptcomp,
             enable_history_search=True,
             mouse_support=self.shell.mouse_support,
             complete_style=self.shell.pt_complete_style,
             style=getattr(self.shell, "style", None),
             color_depth=self.shell.color_depth,
         )
 
         if not PTK3:
             options['inputhook'] = self.shell.inputhook
         options.update(pt_session_options)
         if not _use_simple_prompt:
             self.pt_loop = asyncio.new_event_loop()
             self.pt_app = PromptSession(**options)
 
     def cmdloop(self, intro=None):
         """Repeatedly issue a prompt, accept input, parse an initial prefix
         off the received input, and dispatch to action methods, passing them
         the remainder of the line as argument.
 
         override the same methods from cmd.Cmd to provide prompt toolkit replacement.
         """
         if not self.use_rawinput:
             raise ValueError('Sorry ipdb does not support use_rawinput=False')
 
         # In order to make sure that prompt, which uses asyncio doesn't
         # interfere with applications in which it's used, we always run the
         # prompt itself in a different thread (we can't start an event loop
         # within an event loop). This new thread won't have any event loop
         # running, and here we run our prompt-loop.
         self.preloop()
 
         try:
             if intro is not None:
                 self.intro = intro
             if self.intro:
                 print(self.intro, file=self.stdout)
             stop = None
             while not stop:
                 if self.cmdqueue:
                     line = self.cmdqueue.pop(0)
                 else:
                     self._ptcomp.ipy_completer.namespace = self.curframe_locals
                     self._ptcomp.ipy_completer.global_namespace = self.curframe.f_globals
 
                     # Run the prompt in a different thread.
                     if not _use_simple_prompt:
                         try:
                             line = self.thread_executor.submit(
                                 self.pt_app.prompt
                             ).result()
                         except EOFError:
                             line = "EOF"
                     else:
                         line = input("ipdb> ")
 
                 line = self.precmd(line)
                 stop = self.onecmd(line)
                 stop = self.postcmd(stop, line)
             self.postloop()
         except Exception:
             raise
 
     def do_interact(self, arg):
         ipshell = embed.InteractiveShellEmbed(
             config=self.shell.config,
             banner1="*interactive*",
             exit_msg="*exiting interactive console...*",
         )
         global_ns = self.curframe.f_globals
         ipshell(
             module=sys.modules.get(global_ns["__name__"], None),
             local_ns=self.curframe_locals,
         )
 
 
 def set_trace(frame=None):
     """
     Start debugging from `frame`.
 
     If frame is not specified, debugging starts from caller's frame.
     """
     TerminalPdb().set_trace(frame or sys._getframe().f_back)
 
 
 if __name__ == '__main__':
     import pdb
     # IPython.core.debugger.Pdb.trace_dispatch shall not catch
     # bdb.BdbQuit. When started through __main__ and an exception
     # happened after hitting "c", this is needed in order to
     # be able to quit the debugging session (see #9950).
     old_trace_dispatch = pdb.Pdb.trace_dispatch
     pdb.Pdb = TerminalPdb  # type: ignore
     pdb.Pdb.trace_dispatch = old_trace_dispatch  # type: ignore
     pdb.main()

IPython/testing/plugin/pytest_ipdoctest.py

@@ -1,860 +1,859 b''
 # Based on Pytest doctest.py
 # Original license:
 # The MIT License (MIT)
 #
 # Copyright (c) 2004-2021 Holger Krekel and others
 """Discover and run ipdoctests in modules and test files."""
 import builtins
 import bdb
 import inspect
 import os
 import platform
 import sys
 import traceback
 import types
 import warnings
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import Generator
 from typing import Iterable
 from typing import List
 from typing import Optional
 from typing import Pattern
 from typing import Sequence
 from typing import Tuple
 from typing import Type
 from typing import TYPE_CHECKING
 from typing import Union
 
 import pytest
 from _pytest import outcomes
 from _pytest._code.code import ExceptionInfo
 from _pytest._code.code import ReprFileLocation
 from _pytest._code.code import TerminalRepr
 from _pytest._io import TerminalWriter
 from _pytest.compat import safe_getattr
 from _pytest.config import Config
 from _pytest.config.argparsing import Parser
 from _pytest.fixtures import FixtureRequest
 from _pytest.nodes import Collector
 from _pytest.outcomes import OutcomeException
 from _pytest.pathlib import fnmatch_ex
 from _pytest.pathlib import import_path
 from _pytest.python_api import approx
 from _pytest.warning_types import PytestWarning
 
 if TYPE_CHECKING:
     import doctest
 
 DOCTEST_REPORT_CHOICE_NONE = "none"
 DOCTEST_REPORT_CHOICE_CDIFF = "cdiff"
 DOCTEST_REPORT_CHOICE_NDIFF = "ndiff"
 DOCTEST_REPORT_CHOICE_UDIFF = "udiff"
 DOCTEST_REPORT_CHOICE_ONLY_FIRST_FAILURE = "only_first_failure"
 
 DOCTEST_REPORT_CHOICES = (
     DOCTEST_REPORT_CHOICE_NONE,
     DOCTEST_REPORT_CHOICE_CDIFF,
     DOCTEST_REPORT_CHOICE_NDIFF,
     DOCTEST_REPORT_CHOICE_UDIFF,
     DOCTEST_REPORT_CHOICE_ONLY_FIRST_FAILURE,
 )
 
 # Lazy definition of runner class
 RUNNER_CLASS = None
 # Lazy definition of output checker class
 CHECKER_CLASS: Optional[Type["IPDoctestOutputChecker"]] = None
 
 
 def pytest_addoption(parser: Parser) -> None:
     parser.addini(
         "ipdoctest_optionflags",
         "option flags for ipdoctests",
         type="args",
         default=["ELLIPSIS"],
     )
     parser.addini(
         "ipdoctest_encoding", "encoding used for ipdoctest files", default="utf-8"
     )
     group = parser.getgroup("collect")
     group.addoption(
         "--ipdoctest-modules",
         action="store_true",
         default=False,
         help="run ipdoctests in all .py modules",
         dest="ipdoctestmodules",
     )
     group.addoption(
         "--ipdoctest-report",
         type=str.lower,
         default="udiff",
         help="choose another output format for diffs on ipdoctest failure",
         choices=DOCTEST_REPORT_CHOICES,
         dest="ipdoctestreport",
     )
     group.addoption(
         "--ipdoctest-glob",
         action="append",
         default=[],
         metavar="pat",
         help="ipdoctests file matching pattern, default: test*.txt",
         dest="ipdoctestglob",
     )
     group.addoption(
         "--ipdoctest-ignore-import-errors",
         action="store_true",
         default=False,
         help="ignore ipdoctest ImportErrors",
         dest="ipdoctest_ignore_import_errors",
     )
     group.addoption(
         "--ipdoctest-continue-on-failure",
         action="store_true",
         default=False,
         help="for a given ipdoctest, continue to run after the first failure",
         dest="ipdoctest_continue_on_failure",
     )
 
 
 def pytest_unconfigure() -> None:
     global RUNNER_CLASS
 
     RUNNER_CLASS = None
 
 
 def pytest_collect_file(
     file_path: Path,
     parent: Collector,
 ) -> Optional[Union["IPDoctestModule", "IPDoctestTextfile"]]:
     config = parent.config
     if file_path.suffix == ".py":
         if config.option.ipdoctestmodules and not any(
             (_is_setup_py(file_path), _is_main_py(file_path))
         ):
             mod: IPDoctestModule = IPDoctestModule.from_parent(parent, path=file_path)
             return mod
     elif _is_ipdoctest(config, file_path, parent):
         txt: IPDoctestTextfile = IPDoctestTextfile.from_parent(parent, path=file_path)
         return txt
     return None
 
 
 if int(pytest.__version__.split(".")[0]) < 7:
     _collect_file = pytest_collect_file
 
     def pytest_collect_file(
         path,
         parent: Collector,
     ) -> Optional[Union["IPDoctestModule", "IPDoctestTextfile"]]:
         return _collect_file(Path(path), parent)
 
     _import_path = import_path
 
     def import_path(path, root):
         import py.path
 
         return _import_path(py.path.local(path))
 
 
 def _is_setup_py(path: Path) -> bool:
     if path.name != "setup.py":
         return False
     contents = path.read_bytes()
     return b"setuptools" in contents or b"distutils" in contents
 
 
 def _is_ipdoctest(config: Config, path: Path, parent: Collector) -> bool:
     if path.suffix in (".txt", ".rst") and parent.session.isinitpath(path):
         return True
     globs = config.getoption("ipdoctestglob") or ["test*.txt"]
     return any(fnmatch_ex(glob, path) for glob in globs)
 
 
 def _is_main_py(path: Path) -> bool:
     return path.name == "__main__.py"
 
 
 class ReprFailDoctest(TerminalRepr):
     def __init__(
         self, reprlocation_lines: Sequence[Tuple[ReprFileLocation, Sequence[str]]]
     ) -> None:
         self.reprlocation_lines = reprlocation_lines
 
     def toterminal(self, tw: TerminalWriter) -> None:
         for reprlocation, lines in self.reprlocation_lines:
             for line in lines:
                 tw.line(line)
             reprlocation.toterminal(tw)
 
 
 class MultipleDoctestFailures(Exception):
     def __init__(self, failures: Sequence["doctest.DocTestFailure"]) -> None:
         super().__init__()
         self.failures = failures
 
 
 def _init_runner_class() -> Type["IPDocTestRunner"]:
     import doctest
     from .ipdoctest import IPDocTestRunner
 
     class PytestDoctestRunner(IPDocTestRunner):
         """Runner to collect failures.
 
         Note that the out variable in this case is a list instead of a
         stdout-like object.
         """
 
         def __init__(
             self,
             checker: Optional["IPDoctestOutputChecker"] = None,
             verbose: Optional[bool] = None,
             optionflags: int = 0,
             continue_on_failure: bool = True,
         ) -> None:
             super().__init__(checker=checker, verbose=verbose, optionflags=optionflags)
             self.continue_on_failure = continue_on_failure
 
         def report_failure(
             self,
             out,
             test: "doctest.DocTest",
             example: "doctest.Example",
             got: str,
         ) -> None:
             failure = doctest.DocTestFailure(test, example, got)
             if self.continue_on_failure:
                 out.append(failure)
             else:
                 raise failure
 
         def report_unexpected_exception(
             self,
             out,
             test: "doctest.DocTest",
             example: "doctest.Example",
             exc_info: Tuple[Type[BaseException], BaseException, types.TracebackType],
         ) -> None:
             if isinstance(exc_info[1], OutcomeException):
                 raise exc_info[1]
             if isinstance(exc_info[1], bdb.BdbQuit):
                 outcomes.exit("Quitting debugger")
             failure = doctest.UnexpectedException(test, example, exc_info)
             if self.continue_on_failure:
                 out.append(failure)
             else:
                 raise failure
 
     return PytestDoctestRunner
 
 
 def _get_runner(
     checker: Optional["IPDoctestOutputChecker"] = None,
     verbose: Optional[bool] = None,
     optionflags: int = 0,
     continue_on_failure: bool = True,
 ) -> "IPDocTestRunner":
     # We need this in order to do a lazy import on doctest
     global RUNNER_CLASS
     if RUNNER_CLASS is None:
         RUNNER_CLASS = _init_runner_class()
     # Type ignored because the continue_on_failure argument is only defined on
     # PytestDoctestRunner, which is lazily defined so can't be used as a type.
     return RUNNER_CLASS(  # type: ignore
         checker=checker,
         verbose=verbose,
         optionflags=optionflags,
         continue_on_failure=continue_on_failure,
     )
 
 
 class IPDoctestItem(pytest.Item):
     def __init__(
         self,
         name: str,
         parent: "Union[IPDoctestTextfile, IPDoctestModule]",
         runner: Optional["IPDocTestRunner"] = None,
         dtest: Optional["doctest.DocTest"] = None,
     ) -> None:
         super().__init__(name, parent)
         self.runner = runner
         self.dtest = dtest
         self.obj = None
         self.fixture_request: Optional[FixtureRequest] = None
 
     @classmethod
     def from_parent(  # type: ignore
         cls,
         parent: "Union[IPDoctestTextfile, IPDoctestModule]",
         *,
         name: str,
         runner: "IPDocTestRunner",
         dtest: "doctest.DocTest",
     ):
         # incompatible signature due to imposed limits on subclass
         """The public named constructor."""
         return super().from_parent(name=name, parent=parent, runner=runner, dtest=dtest)
 
     def setup(self) -> None:
         if self.dtest is not None:
             self.fixture_request = _setup_fixtures(self)
             globs = dict(getfixture=self.fixture_request.getfixturevalue)
             for name, value in self.fixture_request.getfixturevalue(
                 "ipdoctest_namespace"
             ).items():
                 globs[name] = value
             self.dtest.globs.update(globs)
 
             from .ipdoctest import IPExample
 
             if isinstance(self.dtest.examples[0], IPExample):
                 # for IPython examples *only*, we swap the globals with the ipython
                 # namespace, after updating it with the globals (which doctest
                 # fills with the necessary info from the module being tested).
                 self._user_ns_orig = {}
                 self._user_ns_orig.update(_ip.user_ns)
                 _ip.user_ns.update(self.dtest.globs)
                 # We must remove the _ key in the namespace, so that Python's
                 # doctest code sets it naturally
                 _ip.user_ns.pop("_", None)
                 _ip.user_ns["__builtins__"] = builtins
                 self.dtest.globs = _ip.user_ns
 
     def teardown(self) -> None:
         from .ipdoctest import IPExample
 
         # Undo the test.globs reassignment we made
         if isinstance(self.dtest.examples[0], IPExample):
             self.dtest.globs = {}
             _ip.user_ns.clear()
             _ip.user_ns.update(self._user_ns_orig)
             del self._user_ns_orig
 
         self.dtest.globs.clear()
 
     def runtest(self) -> None:
         assert self.dtest is not None
         assert self.runner is not None
         _check_all_skipped(self.dtest)
         self._disable_output_capturing_for_darwin()
         failures: List["doctest.DocTestFailure"] = []
 
         # exec(compile(..., "single", ...), ...) puts result in builtins._
         had_underscore_value = hasattr(builtins, "_")
         underscore_original_value = getattr(builtins, "_", None)
 
         # Save our current directory and switch out to the one where the
         # test was originally created, in case another doctest did a
         # directory change.  We'll restore this in the finally clause.
         curdir = os.getcwd()
         os.chdir(self.fspath.dirname)
         try:
             # Type ignored because we change the type of `out` from what
             # ipdoctest expects.
             self.runner.run(self.dtest, out=failures, clear_globs=False)  # type: ignore[arg-type]
         finally:
             os.chdir(curdir)
             if had_underscore_value:
                 setattr(builtins, "_", underscore_original_value)
             elif hasattr(builtins, "_"):
                 delattr(builtins, "_")
 
         if failures:
             raise MultipleDoctestFailures(failures)
 
     def _disable_output_capturing_for_darwin(self) -> None:
         """Disable output capturing. Otherwise, stdout is lost to ipdoctest (pytest#985)."""
         if platform.system() != "Darwin":
             return
         capman = self.config.pluginmanager.getplugin("capturemanager")
         if capman:
             capman.suspend_global_capture(in_=True)
             out, err = capman.read_global_capture()
             sys.stdout.write(out)
             sys.stderr.write(err)
 
     # TODO: Type ignored -- breaks Liskov Substitution.
     def repr_failure(  # type: ignore[override]
         self,
         excinfo: ExceptionInfo[BaseException],
     ) -> Union[str, TerminalRepr]:
         import doctest
 
         failures: Optional[
             Sequence[Union[doctest.DocTestFailure, doctest.UnexpectedException]]
         ] = None
         if isinstance(
             excinfo.value, (doctest.DocTestFailure, doctest.UnexpectedException)
         ):
             failures = [excinfo.value]
         elif isinstance(excinfo.value, MultipleDoctestFailures):
             failures = excinfo.value.failures
 
         if failures is None:
             return super().repr_failure(excinfo)
 
         reprlocation_lines = []
         for failure in failures:
             example = failure.example
             test = failure.test
             filename = test.filename
             if test.lineno is None:
                 lineno = None
             else:
                 lineno = test.lineno + example.lineno + 1
             message = type(failure).__name__
             # TODO: ReprFileLocation doesn't expect a None lineno.
             reprlocation = ReprFileLocation(filename, lineno, message)  # type: ignore[arg-type]
             checker = _get_checker()
             report_choice = _get_report_choice(self.config.getoption("ipdoctestreport"))
             if lineno is not None:
                 assert failure.test.docstring is not None
                 lines = failure.test.docstring.splitlines(False)
                 # add line numbers to the left of the error message
                 assert test.lineno is not None
                 lines = [
                     "%03d %s" % (i + test.lineno + 1, x) for (i, x) in enumerate(lines)
                 ]
                 # trim docstring error lines to 10
                 lines = lines[max(example.lineno - 9, 0) : example.lineno + 1]
             else:
                 lines = [
                     "EXAMPLE LOCATION UNKNOWN, not showing all tests of that example"
                 ]
                 indent = ">>>"
                 for line in example.source.splitlines():
                     lines.append(f"??? {indent} {line}")
                     indent = "..."
             if isinstance(failure, doctest.DocTestFailure):
                 lines += checker.output_difference(
                     example, failure.got, report_choice
                 ).split("\n")
             else:
                 inner_excinfo = ExceptionInfo.from_exc_info(failure.exc_info)
                 lines += ["UNEXPECTED EXCEPTION: %s" % repr(inner_excinfo.value)]
                 lines += [
                     x.strip("\n") for x in traceback.format_exception(*failure.exc_info)
                 ]
             reprlocation_lines.append((reprlocation, lines))
         return ReprFailDoctest(reprlocation_lines)
 
     def reportinfo(self) -> Tuple[Union["os.PathLike[str]", str], Optional[int], str]:
         assert self.dtest is not None
         return self.path, self.dtest.lineno, "[ipdoctest] %s" % self.name
 
     if int(pytest.__version__.split(".")[0]) < 7:
 
         @property
         def path(self) -> Path:
             return Path(self.fspath)
 
 
 def _get_flag_lookup() -> Dict[str, int]:
     import doctest
 
     return dict(
         DONT_ACCEPT_TRUE_FOR_1=doctest.DONT_ACCEPT_TRUE_FOR_1,
         DONT_ACCEPT_BLANKLINE=doctest.DONT_ACCEPT_BLANKLINE,
         NORMALIZE_WHITESPACE=doctest.NORMALIZE_WHITESPACE,
         ELLIPSIS=doctest.ELLIPSIS,
         IGNORE_EXCEPTION_DETAIL=doctest.IGNORE_EXCEPTION_DETAIL,
         COMPARISON_FLAGS=doctest.COMPARISON_FLAGS,
         ALLOW_UNICODE=_get_allow_unicode_flag(),
         ALLOW_BYTES=_get_allow_bytes_flag(),
         NUMBER=_get_number_flag(),
     )
 
 
 def get_optionflags(parent):
     optionflags_str = parent.config.getini("ipdoctest_optionflags")
     flag_lookup_table = _get_flag_lookup()
     flag_acc = 0
     for flag in optionflags_str:
         flag_acc |= flag_lookup_table[flag]
     return flag_acc
 
 
 def _get_continue_on_failure(config):
     continue_on_failure = config.getvalue("ipdoctest_continue_on_failure")
     if continue_on_failure:
         # We need to turn off this if we use pdb since we should stop at
         # the first failure.
         if config.getvalue("usepdb"):
             continue_on_failure = False
     return continue_on_failure
 
 
 class IPDoctestTextfile(pytest.Module):
     obj = None
 
     def collect(self) -> Iterable[IPDoctestItem]:
         import doctest
         from .ipdoctest import IPDocTestParser
 
         # Inspired by doctest.testfile; ideally we would use it directly,
         # but it doesn't support passing a custom checker.
         encoding = self.config.getini("ipdoctest_encoding")
         text = self.path.read_text(encoding)
         filename = str(self.path)
         name = self.path.name
         globs = {"__name__": "__main__"}
 
         optionflags = get_optionflags(self)
 
         runner = _get_runner(
             verbose=False,
             optionflags=optionflags,
             checker=_get_checker(),
             continue_on_failure=_get_continue_on_failure(self.config),
         )
 
         parser = IPDocTestParser()
         test = parser.get_doctest(text, globs, name, filename, 0)
         if test.examples:
             yield IPDoctestItem.from_parent(
                 self, name=test.name, runner=runner, dtest=test
             )
 
     if int(pytest.__version__.split(".")[0]) < 7:
 
         @property
         def path(self) -> Path:
             return Path(self.fspath)
 
         @classmethod
         def from_parent(
             cls,
             parent,
             *,
             fspath=None,
             path: Optional[Path] = None,
             **kw,
         ):
             if path is not None:
                 import py.path
 
                 fspath = py.path.local(path)
             return super().from_parent(parent=parent, fspath=fspath, **kw)
 
 
 def _check_all_skipped(test: "doctest.DocTest") -> None:
     """Raise pytest.skip() if all examples in the given DocTest have the SKIP
     option set."""
     import doctest
 
     all_skipped = all(x.options.get(doctest.SKIP, False) for x in test.examples)
     if all_skipped:
         pytest.skip("all docstests skipped by +SKIP option")
 
 
 def _is_mocked(obj: object) -> bool:
     """Return if an object is possibly a mock object by checking the
     existence of a highly improbable attribute."""
     return (
         safe_getattr(obj, "pytest_mock_example_attribute_that_shouldnt_exist", None)
         is not None
     )
 
 
 @contextmanager
 def _patch_unwrap_mock_aware() -> Generator[None, None, None]:
     """Context manager which replaces ``inspect.unwrap`` with a version
     that's aware of mock objects and doesn't recurse into them."""
     real_unwrap = inspect.unwrap
 
     def _mock_aware_unwrap(
         func: Callable[..., Any], *, stop: Optional[Callable[[Any], Any]] = None
     ) -> Any:
         try:
             if stop is None or stop is _is_mocked:
                 return real_unwrap(func, stop=_is_mocked)
             _stop = stop
             return real_unwrap(func, stop=lambda obj: _is_mocked(obj) or _stop(func))
         except Exception as e:
             warnings.warn(
                 "Got %r when unwrapping %r.  This is usually caused "
                 "by a violation of Python's object protocol; see e.g. "
                 "https://github.com/pytest-dev/pytest/issues/5080" % (e, func),
                 PytestWarning,
             )
             raise
 
     inspect.unwrap = _mock_aware_unwrap
     try:
         yield
     finally:
         inspect.unwrap = real_unwrap
 
 
 class IPDoctestModule(pytest.Module):
     def collect(self) -> Iterable[IPDoctestItem]:
         import doctest
         from .ipdoctest import DocTestFinder, IPDocTestParser
 
         class MockAwareDocTestFinder(DocTestFinder):
             """A hackish ipdoctest finder that overrides stdlib internals to fix a stdlib bug.
 
             https://github.com/pytest-dev/pytest/issues/3456
             https://bugs.python.org/issue25532
             """
 
             def _find_lineno(self, obj, source_lines):
                 """Doctest code does not take into account `@property`, this
                 is a hackish way to fix it. https://bugs.python.org/issue17446
 
                 Wrapped Doctests will need to be unwrapped so the correct
                 line number is returned. This will be reported upstream. #8796
                 """
                 if isinstance(obj, property):
                     obj = getattr(obj, "fget", obj)
 
                 if hasattr(obj, "__wrapped__"):
                     # Get the main obj in case of it being wrapped
                     obj = inspect.unwrap(obj)
 
                 # Type ignored because this is a private function.
                 return super()._find_lineno(  # type:ignore[misc]
                     obj,
                     source_lines,
                 )
 
             def _find(
                 self, tests, obj, name, module, source_lines, globs, seen
             ) -> None:
                 if _is_mocked(obj):
                     return
                 with _patch_unwrap_mock_aware():
-
                     # Type ignored because this is a private function.
                     super()._find(  # type:ignore[misc]
                         tests, obj, name, module, source_lines, globs, seen
                     )
 
         if self.path.name == "conftest.py":
             if int(pytest.__version__.split(".")[0]) < 7:
                 module = self.config.pluginmanager._importconftest(
                     self.path,
                     self.config.getoption("importmode"),
                 )
             else:
                 module = self.config.pluginmanager._importconftest(
                     self.path,
                     self.config.getoption("importmode"),
                     rootpath=self.config.rootpath,
                 )
         else:
             try:
                 module = import_path(self.path, root=self.config.rootpath)
             except ImportError:
                 if self.config.getvalue("ipdoctest_ignore_import_errors"):
                     pytest.skip("unable to import module %r" % self.path)
                 else:
                     raise
         # Uses internal doctest module parsing mechanism.
         finder = MockAwareDocTestFinder(parser=IPDocTestParser())
         optionflags = get_optionflags(self)
         runner = _get_runner(
             verbose=False,
             optionflags=optionflags,
             checker=_get_checker(),
             continue_on_failure=_get_continue_on_failure(self.config),
         )
 
         for test in finder.find(module, module.__name__):
             if test.examples:  # skip empty ipdoctests
                 yield IPDoctestItem.from_parent(
                     self, name=test.name, runner=runner, dtest=test
                 )
 
     if int(pytest.__version__.split(".")[0]) < 7:
 
         @property
         def path(self) -> Path:
             return Path(self.fspath)
 
         @classmethod
         def from_parent(
             cls,
             parent,
             *,
             fspath=None,
             path: Optional[Path] = None,
             **kw,
         ):
             if path is not None:
                 import py.path
 
                 fspath = py.path.local(path)
             return super().from_parent(parent=parent, fspath=fspath, **kw)
 
 
 def _setup_fixtures(doctest_item: IPDoctestItem) -> FixtureRequest:
     """Used by IPDoctestTextfile and IPDoctestItem to setup fixture information."""
 
     def func() -> None:
         pass
 
     doctest_item.funcargs = {}  # type: ignore[attr-defined]
     fm = doctest_item.session._fixturemanager
     doctest_item._fixtureinfo = fm.getfixtureinfo(  # type: ignore[attr-defined]
         node=doctest_item, func=func, cls=None, funcargs=False
     )
     fixture_request = FixtureRequest(doctest_item, _ispytest=True)
     fixture_request._fillfixtures()
     return fixture_request
 
 
 def _init_checker_class() -> Type["IPDoctestOutputChecker"]:
     import doctest
     import re
     from .ipdoctest import IPDoctestOutputChecker
 
     class LiteralsOutputChecker(IPDoctestOutputChecker):
         # Based on doctest_nose_plugin.py from the nltk project
         # (https://github.com/nltk/nltk) and on the "numtest" doctest extension
         # by Sebastien Boisgerault (https://github.com/boisgera/numtest).
 
         _unicode_literal_re = re.compile(r"(\W|^)[uU]([rR]?[\'\"])", re.UNICODE)
         _bytes_literal_re = re.compile(r"(\W|^)[bB]([rR]?[\'\"])", re.UNICODE)
         _number_re = re.compile(
             r"""
             (?P<number>
               (?P<mantissa>
                 (?P<integer1> [+-]?\d*)\.(?P<fraction>\d+)
                 |
                 (?P<integer2> [+-]?\d+)\.
               )
               (?:
                 [Ee]
                 (?P<exponent1> [+-]?\d+)
               )?
               |
               (?P<integer3> [+-]?\d+)
               (?:
                 [Ee]
                 (?P<exponent2> [+-]?\d+)
               )
             )
             """,
             re.VERBOSE,
         )
 
         def check_output(self, want: str, got: str, optionflags: int) -> bool:
             if super().check_output(want, got, optionflags):
                 return True
 
             allow_unicode = optionflags & _get_allow_unicode_flag()
             allow_bytes = optionflags & _get_allow_bytes_flag()
             allow_number = optionflags & _get_number_flag()
 
             if not allow_unicode and not allow_bytes and not allow_number:
                 return False
 
             def remove_prefixes(regex: Pattern[str], txt: str) -> str:
                 return re.sub(regex, r"\1\2", txt)
 
             if allow_unicode:
                 want = remove_prefixes(self._unicode_literal_re, want)
                 got = remove_prefixes(self._unicode_literal_re, got)
 
             if allow_bytes:
                 want = remove_prefixes(self._bytes_literal_re, want)
                 got = remove_prefixes(self._bytes_literal_re, got)
 
             if allow_number:
                 got = self._remove_unwanted_precision(want, got)
 
             return super().check_output(want, got, optionflags)
 
         def _remove_unwanted_precision(self, want: str, got: str) -> str:
             wants = list(self._number_re.finditer(want))
             gots = list(self._number_re.finditer(got))
             if len(wants) != len(gots):
                 return got
             offset = 0
             for w, g in zip(wants, gots):
                 fraction: Optional[str] = w.group("fraction")
                 exponent: Optional[str] = w.group("exponent1")
                 if exponent is None:
                     exponent = w.group("exponent2")
                 precision = 0 if fraction is None else len(fraction)
                 if exponent is not None:
                     precision -= int(exponent)
                 if float(w.group()) == approx(float(g.group()), abs=10**-precision):
                     # They're close enough. Replace the text we actually
                     # got with the text we want, so that it will match when we
                     # check the string literally.
                     got = (
                         got[: g.start() + offset] + w.group() + got[g.end() + offset :]
                     )
                     offset += w.end() - w.start() - (g.end() - g.start())
             return got
 
     return LiteralsOutputChecker
 
 
 def _get_checker() -> "IPDoctestOutputChecker":
     """Return a IPDoctestOutputChecker subclass that supports some
     additional options:
 
     * ALLOW_UNICODE and ALLOW_BYTES options to ignore u'' and b''
       prefixes (respectively) in string literals. Useful when the same
       ipdoctest should run in Python 2 and Python 3.
 
     * NUMBER to ignore floating-point differences smaller than the
       precision of the literal number in the ipdoctest.
 
     An inner class is used to avoid importing "ipdoctest" at the module
     level.
     """
     global CHECKER_CLASS
     if CHECKER_CLASS is None:
         CHECKER_CLASS = _init_checker_class()
     return CHECKER_CLASS()
 
 
 def _get_allow_unicode_flag() -> int:
     """Register and return the ALLOW_UNICODE flag."""
     import doctest
 
     return doctest.register_optionflag("ALLOW_UNICODE")
 
 
 def _get_allow_bytes_flag() -> int:
     """Register and return the ALLOW_BYTES flag."""
     import doctest
 
     return doctest.register_optionflag("ALLOW_BYTES")
 
 
 def _get_number_flag() -> int:
     """Register and return the NUMBER flag."""
     import doctest
 
     return doctest.register_optionflag("NUMBER")
 
 
 def _get_report_choice(key: str) -> int:
     """Return the actual `ipdoctest` module flag value.
 
     We want to do it as late as possible to avoid importing `ipdoctest` and all
     its dependencies when parsing options, as it adds overhead and breaks tests.
     """
     import doctest
 
     return {
         DOCTEST_REPORT_CHOICE_UDIFF: doctest.REPORT_UDIFF,
         DOCTEST_REPORT_CHOICE_CDIFF: doctest.REPORT_CDIFF,
         DOCTEST_REPORT_CHOICE_NDIFF: doctest.REPORT_NDIFF,
         DOCTEST_REPORT_CHOICE_ONLY_FIRST_FAILURE: doctest.REPORT_ONLY_FIRST_FAILURE,
         DOCTEST_REPORT_CHOICE_NONE: 0,
     }[key]
 
 
 @pytest.fixture(scope="session")
 def ipdoctest_namespace() -> Dict[str, Any]:
     """Fixture that returns a :py:class:`dict` that will be injected into the
     namespace of ipdoctests."""
     return dict()