intermediate changes

ref:614ed510ddd3cdf86a8c5dbf19afd113397e0172

4 files changed, 355 insertions, 277 deletions

diff --git a/contrib/python/pytest/py3/_pytest/assertion/__init__.py b/contrib/python/pytest/py3/_pytest/assertion/__init__.py
index ee7fa6a3af0..a18cf198df0 100644
--- a/contrib/python/pytest/py3/_pytest/assertion/__init__.py
+++ b/contrib/python/pytest/py3/_pytest/assertion/__init__.py
@@ -1,24 +1,25 @@
-"""
-support for presenting detailed information in failing assertions.
-"""
+"""Support for presenting detailed information in failing assertions."""
 import sys
 from typing import Any
+from typing import Generator
 from typing import List
 from typing import Optional
+from typing import TYPE_CHECKING
 
 from _pytest.assertion import rewrite
 from _pytest.assertion import truncate
 from _pytest.assertion import util
 from _pytest.assertion.rewrite import assertstate_key
-from _pytest.compat import TYPE_CHECKING
 from _pytest.config import Config
 from _pytest.config import hookimpl
+from _pytest.config.argparsing import Parser
+from _pytest.nodes import Item
 
 if TYPE_CHECKING:
     from _pytest.main import Session
 
 
-def pytest_addoption(parser):
+def pytest_addoption(parser: Parser) -> None:
     group = parser.getgroup("debugconfig")
     group.addoption(
         "--assert",
@@ -27,11 +28,12 @@ def pytest_addoption(parser):
         choices=("rewrite", "plain"),
         default="rewrite",
         metavar="MODE",
-        help="""Control assertion debugging tools.  'plain'
-                            performs no assertion debugging.  'rewrite'
-                            (the default) rewrites assert statements in
-                            test modules on import to provide assert
-                            expression information.""",
+        help=(
+            "Control assertion debugging tools.\n"
+            "'plain' performs no assertion debugging.\n"
+            "'rewrite' (the default) rewrites assert statements in test modules"
+            " on import to provide assert expression information."
+        ),
     )
     parser.addini(
         "enable_assertion_pass_hook",
@@ -42,7 +44,7 @@ def pytest_addoption(parser):
     )
 
 
-def register_assert_rewrite(*names) -> None:
+def register_assert_rewrite(*names: str) -> None:
     """Register one or more module names to be rewritten on import.
 
     This function will make sure that this module or all modules inside
@@ -51,11 +53,11 @@ def register_assert_rewrite(*names) -> None:
     actually imported, usually in your __init__.py if you are a plugin
     using a package.
 
-    :raise TypeError: if the given module names are not strings.
+    :raises TypeError: If the given module names are not strings.
     """
     for name in names:
         if not isinstance(name, str):
-            msg = "expected module names as *args, got {0} instead"
+            msg = "expected module names as *args, got {0} instead"  # type: ignore[unreachable]
             raise TypeError(msg.format(repr(names)))
     for hook in sys.meta_path:
         if isinstance(hook, rewrite.AssertionRewritingHook):
@@ -71,27 +73,27 @@ def register_assert_rewrite(*names) -> None:
 class DummyRewriteHook:
     """A no-op import hook for when rewriting is disabled."""
 
-    def mark_rewrite(self, *names):
+    def mark_rewrite(self, *names: str) -> None:
         pass
 
 
 class AssertionState:
     """State for the assertion plugin."""
 
-    def __init__(self, config, mode):
+    def __init__(self, config: Config, mode) -> None:
         self.mode = mode
         self.trace = config.trace.root.get("assertion")
-        self.hook = None  # type: Optional[rewrite.AssertionRewritingHook]
+        self.hook: Optional[rewrite.AssertionRewritingHook] = None
 
 
-def install_importhook(config):
+def install_importhook(config: Config) -> rewrite.AssertionRewritingHook:
     """Try to install the rewrite hook, raise SystemError if it fails."""
     config._store[assertstate_key] = AssertionState(config, "rewrite")
     config._store[assertstate_key].hook = hook = rewrite.AssertionRewritingHook(config)
     sys.meta_path.insert(0, hook)
     config._store[assertstate_key].trace("installed rewrite import hook")
 
-    def undo():
+    def undo() -> None:
         hook = config._store[assertstate_key].hook
         if hook is not None and hook in sys.meta_path:
             sys.meta_path.remove(hook)
@@ -101,9 +103,9 @@ def install_importhook(config):
 
 
 def pytest_collection(session: "Session") -> None:
-    # this hook is only called when test modules are collected
+    # This hook is only called when test modules are collected
     # so for example not in the master process of pytest-xdist
-    # (which does not collect test modules)
+    # (which does not collect test modules).
     assertstate = session.config._store.get(assertstate_key, None)
     if assertstate:
         if assertstate.hook is not None:
@@ -111,18 +113,18 @@ def pytest_collection(session: "Session") -> None:
 
 
 @hookimpl(tryfirst=True, hookwrapper=True)
-def pytest_runtest_protocol(item):
-    """Setup the pytest_assertrepr_compare and pytest_assertion_pass hooks
+def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
+    """Setup the pytest_assertrepr_compare and pytest_assertion_pass hooks.
 
-    The newinterpret and rewrite modules will use util._reprcompare if
-    it exists to use custom reporting via the
-    pytest_assertrepr_compare hook.  This sets up this custom
+    The rewrite module will use util._reprcompare if it exists to use custom
+    reporting via the pytest_assertrepr_compare hook.  This sets up this custom
     comparison for the test.
     """
 
-    def callbinrepr(op, left, right):
-        # type: (str, object, object) -> Optional[str]
-        """Call the pytest_assertrepr_compare hook and prepare the result
+    ihook = item.ihook
+
+    def callbinrepr(op, left: object, right: object) -> Optional[str]:
+        """Call the pytest_assertrepr_compare hook and prepare the result.
 
         This uses the first result from the hook and then ensures the
         following:
@@ -136,7 +138,7 @@ def pytest_runtest_protocol(item):
         The result can be formatted by util.format_explanation() for
         pretty printing.
         """
-        hook_result = item.ihook.pytest_assertrepr_compare(
+        hook_result = ihook.pytest_assertrepr_compare(
             config=item.config, op=op, left=left, right=right
         )
         for new_expl in hook_result:
@@ -152,12 +154,10 @@ def pytest_runtest_protocol(item):
     saved_assert_hooks = util._reprcompare, util._assertion_pass
     util._reprcompare = callbinrepr
 
-    if item.ihook.pytest_assertion_pass.get_hookimpls():
+    if ihook.pytest_assertion_pass.get_hookimpls():
 
-        def call_assertion_pass_hook(lineno, orig, expl):
-            item.ihook.pytest_assertion_pass(
-                item=item, lineno=lineno, orig=orig, expl=expl
-            )
+        def call_assertion_pass_hook(lineno: int, orig: str, expl: str) -> None:
+            ihook.pytest_assertion_pass(item=item, lineno=lineno, orig=orig, expl=expl)
 
         util._assertion_pass = call_assertion_pass_hook
 
@@ -166,7 +166,7 @@ def pytest_runtest_protocol(item):
     util._reprcompare, util._assertion_pass = saved_assert_hooks
 
 
-def pytest_sessionfinish(session):
+def pytest_sessionfinish(session: "Session") -> None:
     assertstate = session.config._store.get(assertstate_key, None)
     if assertstate:
         if assertstate.hook is not None:
diff --git a/contrib/python/pytest/py3/_pytest/assertion/rewrite.py b/contrib/python/pytest/py3/_pytest/assertion/rewrite.py
index f84127dcaf1..37ff076aab5 100644
--- a/contrib/python/pytest/py3/_pytest/assertion/rewrite.py
+++ b/contrib/python/pytest/py3/_pytest/assertion/rewrite.py
@@ -1,4 +1,4 @@
-"""Rewrite assertion AST to produce nice error messages"""
+"""Rewrite assertion AST to produce nice error messages."""
 import ast
 import errno
 import functools
@@ -13,11 +13,21 @@ import struct
 import sys
 import tokenize
 import types
+from pathlib import Path
+from pathlib import PurePath
+from typing import Callable
 from typing import Dict
+from typing import IO
+from typing import Iterable
 from typing import List
 from typing import Optional
+from typing import Sequence
 from typing import Set
 from typing import Tuple
+from typing import TYPE_CHECKING
+from typing import Union
+
+import py
 
 from _pytest._io.saferepr import saferepr
 from _pytest._version import version
@@ -25,22 +35,20 @@ from _pytest.assertion import util
 from _pytest.assertion.util import (  # noqa: F401
     format_explanation as _format_explanation,
 )
-from _pytest.compat import fspath
-from _pytest.compat import TYPE_CHECKING
+from _pytest.config import Config
+from _pytest.main import Session
 from _pytest.pathlib import fnmatch_ex
-from _pytest.pathlib import Path
-from _pytest.pathlib import PurePath
 from _pytest.store import StoreKey
 
 if TYPE_CHECKING:
-    from _pytest.assertion import AssertionState  # noqa: F401
+    from _pytest.assertion import AssertionState
 
 
 assertstate_key = StoreKey["AssertionState"]()
 
 
 # pytest caches rewritten pycs in pycache dirs
-PYTEST_TAG = "{}-pytest-{}".format(sys.implementation.cache_tag, version)
+PYTEST_TAG = f"{sys.implementation.cache_tag}-pytest-{version}"
 PYC_EXT = ".py" + (__debug__ and "c" or "o")
 PYC_TAIL = "." + PYTEST_TAG + PYC_EXT
 
@@ -48,30 +56,35 @@ PYC_TAIL = "." + PYTEST_TAG + PYC_EXT
 class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader):
     """PEP302/PEP451 import hook which rewrites asserts."""
 
-    def __init__(self, config):
+    def __init__(self, config: Config) -> None:
         self.config = config
         try:
             self.fnpats = config.getini("python_files")
         except ValueError:
             self.fnpats = ["test_*.py", "*_test.py"]
-        self.session = None
-        self._rewritten_names = set()  # type: Set[str]
-        self._must_rewrite = set()  # type: Set[str]
+        self.session: Optional[Session] = None
+        self._rewritten_names: Set[str] = set()
+        self._must_rewrite: Set[str] = set()
         # flag to guard against trying to rewrite a pyc file while we are already writing another pyc file,
         # which might result in infinite recursion (#3506)
         self._writing_pyc = False
         self._basenames_to_check_rewrite = {"conftest"}
-        self._marked_for_rewrite_cache = {}  # type: Dict[str, bool]
+        self._marked_for_rewrite_cache: Dict[str, bool] = {}
         self._session_paths_checked = False
 
-    def set_session(self, session):
+    def set_session(self, session: Optional[Session]) -> None:
         self.session = session
         self._session_paths_checked = False
 
     # Indirection so we can mock calls to find_spec originated from the hook during testing
     _find_spec = importlib.machinery.PathFinder.find_spec
 
-    def find_spec(self, name, path=None, target=None):
+    def find_spec(
+        self,
+        name: str,
+        path: Optional[Sequence[Union[str, bytes]]] = None,
+        target: Optional[types.ModuleType] = None,
+    ) -> Optional[importlib.machinery.ModuleSpec]:
         if self._writing_pyc:
             return None
         state = self.config._store[assertstate_key]
@@ -79,13 +92,14 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
             return None
         state.trace("find_module called for: %s" % name)
 
-        spec = self._find_spec(name, path)
+        # Type ignored because mypy is confused about the `self` binding here.
+        spec = self._find_spec(name, path)  # type: ignore
         if (
             # the import machinery could not find a file to import
             spec is None
             # this is a namespace package (without `__init__.py`)
             # there's nothing to rewrite there
-            # python3.5 - python3.6: `namespace`
+            # python3.6: `namespace`
             # python3.7+: `None`
             or spec.origin == "namespace"
             or spec.origin is None
@@ -108,10 +122,14 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
             submodule_search_locations=spec.submodule_search_locations,
         )
 
-    def create_module(self, spec):
+    def create_module(
+        self, spec: importlib.machinery.ModuleSpec
+    ) -> Optional[types.ModuleType]:
         return None  # default behaviour is fine
 
-    def exec_module(self, module):
+    def exec_module(self, module: types.ModuleType) -> None:
+        assert module.__spec__ is not None
+        assert module.__spec__.origin is not None
         fn = Path(module.__spec__.origin)
         state = self.config._store[assertstate_key]
 
@@ -131,7 +149,7 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
             ok = try_makedirs(cache_dir)
             if not ok:
                 write = False
-                state.trace("read only directory: {}".format(cache_dir))
+                state.trace(f"read only directory: {cache_dir}")
 
         cache_name = fn.name[:-3] + PYC_TAIL
         pyc = cache_dir / cache_name
@@ -139,7 +157,7 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
         # to check for a cached pyc. This may not be optimal...
         co = _read_pyc(fn, pyc, state.trace)
         if co is None:
-            state.trace("rewriting {!r}".format(fn))
+            state.trace(f"rewriting {fn!r}")
             source_stat, co = _rewrite_test(fn, self.config)
             if write:
                 self._writing_pyc = True
@@ -148,11 +166,11 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
                 finally:
                     self._writing_pyc = False
         else:
-            state.trace("found cached rewritten pyc for {}".format(fn))
+            state.trace(f"found cached rewritten pyc for {fn}")
         exec(co, module.__dict__)
 
-    def _early_rewrite_bailout(self, name, state):
-        """This is a fast way to get out of rewriting modules.
+    def _early_rewrite_bailout(self, name: str, state: "AssertionState") -> bool:
+        """A fast way to get out of rewriting modules.
 
         Profiling has shown that the call to PathFinder.find_spec (inside of
         the find_spec from this class) is a major slowdown, so, this method
@@ -161,10 +179,10 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
         """
         if self.session is not None and not self._session_paths_checked:
             self._session_paths_checked = True
-            for path in self.session._initialpaths:
+            for initial_path in self.session._initialpaths:
                 # Make something as c:/projects/my_project/path.py ->
                 #     ['c:', 'projects', 'my_project', 'path.py']
-                parts = str(path).split(os.path.sep)
+                parts = str(initial_path).split(os.path.sep)
                 # add 'path' to basenames to be checked.
                 self._basenames_to_check_rewrite.add(os.path.splitext(parts[-1])[0])
 
@@ -187,20 +205,18 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
         if self._is_marked_for_rewrite(name, state):
             return False
 
-        state.trace("early skip of rewriting module: {}".format(name))
+        state.trace(f"early skip of rewriting module: {name}")
         return True
 
-    def _should_rewrite(self, name, fn, state):
+    def _should_rewrite(self, name: str, fn: str, state: "AssertionState") -> bool:
         # always rewrite conftest files
         if os.path.basename(fn) == "conftest.py":
-            state.trace("rewriting conftest file: {!r}".format(fn))
+            state.trace(f"rewriting conftest file: {fn!r}")
             return True
 
         if self.session is not None:
-            if self.session.isinitpath(fn):
-                state.trace(
-                    "matched test file (was specified on cmdline): {!r}".format(fn)
-                )
+            if self.session.isinitpath(py.path.local(fn)):
+                state.trace(f"matched test file (was specified on cmdline): {fn!r}")
                 return True
 
         # modules not passed explicitly on the command line are only
@@ -208,20 +224,18 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
         fn_path = PurePath(fn)
         for pat in self.fnpats:
             if fnmatch_ex(pat, fn_path):
-                state.trace("matched test file {!r}".format(fn))
+                state.trace(f"matched test file {fn!r}")
                 return True
 
         return self._is_marked_for_rewrite(name, state)
 
-    def _is_marked_for_rewrite(self, name: str, state):
+    def _is_marked_for_rewrite(self, name: str, state: "AssertionState") -> bool:
         try:
             return self._marked_for_rewrite_cache[name]
         except KeyError:
             for marked in self._must_rewrite:
                 if name == marked or name.startswith(marked + "."):
-                    state.trace(
-                        "matched marked file {!r} (from {!r})".format(name, marked)
-                    )
+                    state.trace(f"matched marked file {name!r} (from {marked!r})")
                     self._marked_for_rewrite_cache[name] = True
                     return True
 
@@ -246,33 +260,37 @@ class AssertionRewritingHook(importlib.abc.MetaPathFinder, importlib.abc.Loader)
         self._must_rewrite.update(names)
         self._marked_for_rewrite_cache.clear()
 
-    def _warn_already_imported(self, name):
+    def _warn_already_imported(self, name: str) -> None:
         from _pytest.warning_types import PytestAssertRewriteWarning
-        from _pytest.warnings import _issue_warning_captured
 
-        _issue_warning_captured(
+        self.config.issue_config_time_warning(
             PytestAssertRewriteWarning(
                 "Module already imported so cannot be rewritten: %s" % name
             ),
-            self.config.hook,
             stacklevel=5,
         )
 
-    def get_data(self, pathname):
+    def get_data(self, pathname: Union[str, bytes]) -> bytes:
         """Optional PEP302 get_data API."""
         with open(pathname, "rb") as f:
             return f.read()
 
 
-def _write_pyc_fp(fp, source_stat, co):
+def _write_pyc_fp(
+    fp: IO[bytes], source_stat: os.stat_result, co: types.CodeType
+) -> None:
     # Technically, we don't have to have the same pyc format as
     # (C)Python, since these "pycs" should never be seen by builtin
-    # import. However, there's little reason deviate.
+    # import. However, there's little reason to deviate.
     fp.write(importlib.util.MAGIC_NUMBER)
+    # https://www.python.org/dev/peps/pep-0552/
+    if sys.version_info >= (3, 7):
+        flags = b"\x00\x00\x00\x00"
+        fp.write(flags)
     # as of now, bytecode header expects 32-bit numbers for size and mtime (#4903)
     mtime = int(source_stat.st_mtime) & 0xFFFFFFFF
     size = source_stat.st_size & 0xFFFFFFFF
-    # "<LL" stands for 2 unsigned longs, little-ending
+    # "<LL" stands for 2 unsigned longs, little-endian.
     fp.write(struct.pack("<LL", mtime, size))
     fp.write(marshal.dumps(co))
 
@@ -280,12 +298,17 @@ def _write_pyc_fp(fp, source_stat, co):
 if sys.platform == "win32":
     from atomicwrites import atomic_write
 
-    def _write_pyc(state, co, source_stat, pyc):
+    def _write_pyc(
+        state: "AssertionState",
+        co: types.CodeType,
+        source_stat: os.stat_result,
+        pyc: Path,
+    ) -> bool:
         try:
-            with atomic_write(fspath(pyc), mode="wb", overwrite=True) as fp:
+            with atomic_write(os.fspath(pyc), mode="wb", overwrite=True) as fp:
                 _write_pyc_fp(fp, source_stat, co)
-        except EnvironmentError as e:
-            state.trace("error writing pyc file at {}: errno={}".format(pyc, e.errno))
+        except OSError as e:
+            state.trace(f"error writing pyc file at {pyc}: {e}")
             # we ignore any failure to write the cache file
             # there are many reasons, permission-denied, pycache dir being a
             # file etc.
@@ -295,21 +318,24 @@ if sys.platform == "win32":
 
 else:
 
-    def _write_pyc(state, co, source_stat, pyc):
-        proc_pyc = "{}.{}".format(pyc, os.getpid())
+    def _write_pyc(
+        state: "AssertionState",
+        co: types.CodeType,
+        source_stat: os.stat_result,
+        pyc: Path,
+    ) -> bool:
+        proc_pyc = f"{pyc}.{os.getpid()}"
         try:
             fp = open(proc_pyc, "wb")
-        except EnvironmentError as e:
-            state.trace(
-                "error writing pyc file at {}: errno={}".format(proc_pyc, e.errno)
-            )
+        except OSError as e:
+            state.trace(f"error writing pyc file at {proc_pyc}: errno={e.errno}")
             return False
 
         try:
             _write_pyc_fp(fp, source_stat, co)
-            os.rename(proc_pyc, fspath(pyc))
-        except BaseException as e:
-            state.trace("error writing pyc file at {}: errno={}".format(pyc, e.errno))
+            os.rename(proc_pyc, os.fspath(pyc))
+        except OSError as e:
+            state.trace(f"error writing pyc file at {pyc}: {e}")
             # we ignore any failure to write the cache file
             # there are many reasons, permission-denied, pycache dir being a
             # file etc.
@@ -319,48 +345,62 @@ else:
         return True
 
 
-def _rewrite_test(fn, config):
-    """read and rewrite *fn* and return the code object."""
-    fn = fspath(fn)
-    stat = os.stat(fn)
-    with open(fn, "rb") as f:
+def _rewrite_test(fn: Path, config: Config) -> Tuple[os.stat_result, types.CodeType]:
+    """Read and rewrite *fn* and return the code object."""
+    fn_ = os.fspath(fn)
+    stat = os.stat(fn_)
+    with open(fn_, "rb") as f:
         source = f.read()
-    tree = ast.parse(source, filename=fn)
-    rewrite_asserts(tree, source, fn, config)
-    co = compile(tree, fn, "exec", dont_inherit=True)
+    tree = ast.parse(source, filename=fn_)
+    rewrite_asserts(tree, source, fn_, config)
+    co = compile(tree, fn_, "exec", dont_inherit=True)
     return stat, co
 
 
-def _read_pyc(source, pyc, trace=lambda x: None):
+def _read_pyc(
+    source: Path, pyc: Path, trace: Callable[[str], None] = lambda x: None
+) -> Optional[types.CodeType]:
     """Possibly read a pytest pyc containing rewritten code.
 
     Return rewritten code if successful or None if not.
     """
     try:
-        fp = open(fspath(pyc), "rb")
-    except IOError:
+        fp = open(os.fspath(pyc), "rb")
+    except OSError:
         return None
     with fp:
+        # https://www.python.org/dev/peps/pep-0552/
+        has_flags = sys.version_info >= (3, 7)
         try:
-            stat_result = os.stat(fspath(source))
+            stat_result = os.stat(os.fspath(source))
             mtime = int(stat_result.st_mtime)
             size = stat_result.st_size
-            data = fp.read(12)
-        except EnvironmentError as e:
-            trace("_read_pyc({}): EnvironmentError {}".format(source, e))
+            data = fp.read(16 if has_flags else 12)
+        except OSError as e:
+            trace(f"_read_pyc({source}): OSError {e}")
             return None
         # Check for invalid or out of date pyc file.
-        if (
-            len(data) != 12
-            or data[:4] != importlib.util.MAGIC_NUMBER
-            or struct.unpack("<LL", data[4:]) != (mtime & 0xFFFFFFFF, size & 0xFFFFFFFF)
-        ):
-            trace("_read_pyc(%s): invalid or out of date pyc" % source)
+        if len(data) != (16 if has_flags else 12):
+            trace("_read_pyc(%s): invalid pyc (too short)" % source)
+            return None
+        if data[:4] != importlib.util.MAGIC_NUMBER:
+            trace("_read_pyc(%s): invalid pyc (bad magic number)" % source)
+            return None
+        if has_flags and data[4:8] != b"\x00\x00\x00\x00":
+            trace("_read_pyc(%s): invalid pyc (unsupported flags)" % source)
+            return None
+        mtime_data = data[8 if has_flags else 4 : 12 if has_flags else 8]
+        if int.from_bytes(mtime_data, "little") != mtime & 0xFFFFFFFF:
+            trace("_read_pyc(%s): out of date" % source)
+            return None
+        size_data = data[12 if has_flags else 8 : 16 if has_flags else 12]
+        if int.from_bytes(size_data, "little") != size & 0xFFFFFFFF:
+            trace("_read_pyc(%s): invalid pyc (incorrect size)" % source)
             return None
         try:
             co = marshal.load(fp)
         except Exception as e:
-            trace("_read_pyc({}): marshal.load error {}".format(source, e))
+            trace(f"_read_pyc({source}): marshal.load error {e}")
             return None
         if not isinstance(co, types.CodeType):
             trace("_read_pyc(%s): not a code object" % source)
@@ -368,13 +408,18 @@ def _read_pyc(source, pyc, trace=lambda x: None):
         return co
 
 
-def rewrite_asserts(mod, source, module_path=None, config=None):
+def rewrite_asserts(
+    mod: ast.Module,
+    source: bytes,
+    module_path: Optional[str] = None,
+    config: Optional[Config] = None,
+) -> None:
     """Rewrite the assert statements in mod."""
     AssertionRewriter(module_path, config, source).run(mod)
 
 
-def _saferepr(obj):
-    """Get a safe repr of an object for assertion error messages.
+def _saferepr(obj: object) -> str:
+    r"""Get a safe repr of an object for assertion error messages.
 
     The assertion formatting (util.format_explanation()) requires
     newlines to be escaped since they are a special character for it.
@@ -382,18 +427,16 @@ def _saferepr(obj):
     custom repr it is possible to contain one of the special escape
     sequences, especially '\n{' and '\n}' are likely to be present in
     JSON reprs.
-
     """
     return saferepr(obj).replace("\n", "\\n")
 
 
-def _format_assertmsg(obj):
-    """Format the custom assertion message given.
+def _format_assertmsg(obj: object) -> str:
+    r"""Format the custom assertion message given.
 
     For strings this simply replaces newlines with '\n~' so that
     util.format_explanation() will preserve them instead of escaping
     newlines.  For other objects saferepr() is used first.
-
     """
     # reprlib appears to have a bug which means that if a string
     # contains a newline it gets escaped, however if an object has a
@@ -410,7 +453,7 @@ def _format_assertmsg(obj):
     return obj
 
 
-def _should_repr_global_name(obj):
+def _should_repr_global_name(obj: object) -> bool:
     if callable(obj):
         return False
 
@@ -420,16 +463,17 @@ def _should_repr_global_name(obj):
         return True
 
 
-def _format_boolop(explanations, is_or):
+def _format_boolop(explanations: Iterable[str], is_or: bool) -> str:
     explanation = "(" + (is_or and " or " or " and ").join(explanations) + ")"
-    if isinstance(explanation, str):
-        return explanation.replace("%", "%%")
-    else:
-        return explanation.replace(b"%", b"%%")
+    return explanation.replace("%", "%%")
 
 
-def _call_reprcompare(ops, results, expls, each_obj):
-    # type: (Tuple[str, ...], Tuple[bool, ...], Tuple[str, ...], Tuple[object, ...]) -> str
+def _call_reprcompare(
+    ops: Sequence[str],
+    results: Sequence[bool],
+    expls: Sequence[str],
+    each_obj: Sequence[object],
+) -> str:
     for i, res, expl in zip(range(len(ops)), results, expls):
         try:
             done = not res
@@ -444,16 +488,14 @@ def _call_reprcompare(ops, results, expls, each_obj):
     return expl
 
 
-def _call_assertion_pass(lineno, orig, expl):
-    # type: (int, str, str) -> None
+def _call_assertion_pass(lineno: int, orig: str, expl: str) -> None:
     if util._assertion_pass is not None:
         util._assertion_pass(lineno, orig, expl)
 
 
-def _check_if_assertion_pass_impl():
-    # type: () -> bool
-    """Checks if any plugins implement the pytest_assertion_pass hook
-    in order not to generate explanation unecessarily (might be expensive)"""
+def _check_if_assertion_pass_impl() -> bool:
+    """Check if any plugins implement the pytest_assertion_pass hook
+    in order not to generate explanation unecessarily (might be expensive)."""
     return True if util._assertion_pass else False
 
 
@@ -502,13 +544,13 @@ def set_location(node, lineno, col_offset):
 
 
 def _get_assertion_exprs(src: bytes) -> Dict[int, str]:
-    """Returns a mapping from {lineno: "assertion test expression"}"""
-    ret = {}  # type: Dict[int, str]
+    """Return a mapping from {lineno: "assertion test expression"}."""
+    ret: Dict[int, str] = {}
 
     depth = 0
-    lines = []  # type: List[str]
-    assert_lineno = None  # type: Optional[int]
-    seen_lines = set()  # type: Set[int]
+    lines: List[str] = []
+    assert_lineno: Optional[int] = None
+    seen_lines: Set[int] = set()
 
     def _write_and_reset() -> None:
         nonlocal depth, lines, assert_lineno, seen_lines
@@ -606,10 +648,11 @@ class AssertionRewriter(ast.NodeVisitor):
 
     This state is reset on every new assert statement visited and used
     by the other visitors.
-
     """
 
-    def __init__(self, module_path, config, source):
+    def __init__(
+        self, module_path: Optional[str], config: Optional[Config], source: bytes
+    ) -> None:
         super().__init__()
         self.module_path = module_path
         self.config = config
@@ -622,7 +665,7 @@ class AssertionRewriter(ast.NodeVisitor):
         self.source = source
 
     @functools.lru_cache(maxsize=1)
-    def _assert_expr_to_lineno(self):
+    def _assert_expr_to_lineno(self) -> Dict[int, str]:
         return _get_assertion_exprs(self.source)
 
     def run(self, mod: ast.Module) -> None:
@@ -630,12 +673,9 @@ class AssertionRewriter(ast.NodeVisitor):
         if not mod.body:
             # Nothing to do.
             return
-        # Insert some special imports at the top of the module but after any
-        # docstrings and __future__ imports.
-        aliases = [
-            ast.alias("builtins", "@py_builtins"),
-            ast.alias("_pytest.assertion.rewrite", "@pytest_ar"),
-        ]
+
+        # We'll insert some special imports at the top of the module, but after any
+        # docstrings and __future__ imports, so first figure out where that is.
         doc = getattr(mod, "docstring", None)
         expect_docstring = doc is None
         if doc is not None and self.is_rewrite_disabled(doc):
@@ -653,26 +693,48 @@ class AssertionRewriter(ast.NodeVisitor):
                     return
                 expect_docstring = False
             elif (
-                not isinstance(item, ast.ImportFrom)
-                or item.level > 0
-                or item.module != "__future__"
+                isinstance(item, ast.ImportFrom)
+                and item.level == 0
+                and item.module == "__future__"
             ):
-                lineno = item.lineno
+                pass
+            else:
                 break
             pos += 1
+        # Special case: for a decorated function, set the lineno to that of the
+        # first decorator, not the `def`. Issue #4984.
+        if isinstance(item, ast.FunctionDef) and item.decorator_list:
+            lineno = item.decorator_list[0].lineno
         else:
             lineno = item.lineno
+        # Now actually insert the special imports.
+        if sys.version_info >= (3, 10):
+            aliases = [
+                ast.alias("builtins", "@py_builtins", lineno=lineno, col_offset=0),
+                ast.alias(
+                    "_pytest.assertion.rewrite",
+                    "@pytest_ar",
+                    lineno=lineno,
+                    col_offset=0,
+                ),
+            ]
+        else:
+            aliases = [
+                ast.alias("builtins", "@py_builtins"),
+                ast.alias("_pytest.assertion.rewrite", "@pytest_ar"),
+            ]
         imports = [
             ast.Import([alias], lineno=lineno, col_offset=0) for alias in aliases
         ]
         mod.body[pos:pos] = imports
+
         # Collect asserts.
-        nodes = [mod]  # type: List[ast.AST]
+        nodes: List[ast.AST] = [mod]
         while nodes:
             node = nodes.pop()
             for name, field in ast.iter_fields(node):
                 if isinstance(field, list):
-                    new = []  # type: List
+                    new: List[ast.AST] = []
                     for i, child in enumerate(field):
                         if isinstance(child, ast.Assert):
                             # Transform assert.
@@ -691,51 +753,50 @@ class AssertionRewriter(ast.NodeVisitor):
                     nodes.append(field)
 
     @staticmethod
-    def is_rewrite_disabled(docstring):
+    def is_rewrite_disabled(docstring: str) -> bool:
         return "PYTEST_DONT_REWRITE" in docstring
 
-    def variable(self):
+    def variable(self) -> str:
         """Get a new variable."""
         # Use a character invalid in python identifiers to avoid clashing.
         name = "@py_assert" + str(next(self.variable_counter))
         self.variables.append(name)
         return name
 
-    def assign(self, expr):
+    def assign(self, expr: ast.expr) -> ast.Name:
         """Give *expr* a name."""
         name = self.variable()
         self.statements.append(ast.Assign([ast.Name(name, ast.Store())], expr))
         return ast.Name(name, ast.Load())
 
-    def display(self, expr):
+    def display(self, expr: ast.expr) -> ast.expr:
         """Call saferepr on the expression."""
         return self.helper("_saferepr", expr)
 
-    def helper(self, name, *args):
+    def helper(self, name: str, *args: ast.expr) -> ast.expr:
         """Call a helper in this module."""
         py_name = ast.Name("@pytest_ar", ast.Load())
         attr = ast.Attribute(py_name, name, ast.Load())
         return ast.Call(attr, list(args), [])
 
-    def builtin(self, name):
+    def builtin(self, name: str) -> ast.Attribute:
         """Return the builtin called *name*."""
         builtin_name = ast.Name("@py_builtins", ast.Load())
         return ast.Attribute(builtin_name, name, ast.Load())
 
-    def explanation_param(self, expr):
+    def explanation_param(self, expr: ast.expr) -> str:
         """Return a new named %-formatting placeholder for expr.
 
         This creates a %-formatting placeholder for expr in the
         current formatting context, e.g. ``%(py0)s``.  The placeholder
         and expr are placed in the current format context so that it
         can be used on the next call to .pop_format_context().
-
         """
         specifier = "py" + str(next(self.variable_counter))
         self.explanation_specifiers[specifier] = expr
         return "%(" + specifier + ")s"
 
-    def push_format_context(self):
+    def push_format_context(self) -> None:
         """Create a new formatting context.
 
         The format context is used for when an explanation wants to
@@ -744,19 +805,17 @@ class AssertionRewriter(ast.NodeVisitor):
         .explanation_param().  Finally .pop_format_context() is used
         to format a string of %-formatted values as added by
         .explanation_param().
-
         """
-        self.explanation_specifiers = {}  # type: Dict[str, ast.expr]
+        self.explanation_specifiers: Dict[str, ast.expr] = {}
         self.stack.append(self.explanation_specifiers)
 
-    def pop_format_context(self, expl_expr):
+    def pop_format_context(self, expl_expr: ast.expr) -> ast.Name:
         """Format the %-formatted string with current format context.
 
-        The expl_expr should be an ast.Str instance constructed from
+        The expl_expr should be an str ast.expr instance constructed from
         the %-placeholders created by .explanation_param().  This will
         add the required code to format said string to .expl_stmts and
         return the ast.Name instance of the formatted string.
-
         """
         current = self.stack.pop()
         if self.stack:
@@ -770,43 +829,44 @@ class AssertionRewriter(ast.NodeVisitor):
         self.expl_stmts.append(ast.Assign([ast.Name(name, ast.Store())], form))
         return ast.Name(name, ast.Load())
 
-    def generic_visit(self, node):
+    def generic_visit(self, node: ast.AST) -> Tuple[ast.Name, str]:
         """Handle expressions we don't have custom code for."""
         assert isinstance(node, ast.expr)
         res = self.assign(node)
         return res, self.explanation_param(self.display(res))
 
-    def visit_Assert(self, assert_):
+    def visit_Assert(self, assert_: ast.Assert) -> List[ast.stmt]:
         """Return the AST statements to replace the ast.Assert instance.
 
         This rewrites the test of an assertion to provide
         intermediate values and replace it with an if statement which
         raises an assertion error with a detailed explanation in case
         the expression is false.
-
         """
         if isinstance(assert_.test, ast.Tuple) and len(assert_.test.elts) >= 1:
             from _pytest.warning_types import PytestAssertRewriteWarning
             import warnings
 
+            # TODO: This assert should not be needed.
+            assert self.module_path is not None
             warnings.warn_explicit(
                 PytestAssertRewriteWarning(
                     "assertion is always true, perhaps remove parentheses?"
                 ),
                 category=None,
-                filename=fspath(self.module_path),
+                filename=os.fspath(self.module_path),
                 lineno=assert_.lineno,
             )
 
-        self.statements = []  # type: List[ast.stmt]
-        self.variables = []  # type: List[str]
+        self.statements: List[ast.stmt] = []
+        self.variables: List[str] = []
         self.variable_counter = itertools.count()
 
         if self.enable_assertion_pass_hook:
-            self.format_variables = []  # type: List[str]
+            self.format_variables: List[str] = []
 
-        self.stack = []  # type: List[Dict[str, ast.expr]]
-        self.expl_stmts = []  # type: List[ast.stmt]
+        self.stack: List[Dict[str, ast.expr]] = []
+        self.expl_stmts: List[ast.stmt] = []
         self.push_format_context()
         # Rewrite assert into a bunch of statements.
         top_condition, explanation = self.visit(assert_.test)
@@ -891,7 +951,7 @@ class AssertionRewriter(ast.NodeVisitor):
             set_location(stmt, assert_.lineno, assert_.col_offset)
         return self.statements
 
-    def visit_Name(self, name):
+    def visit_Name(self, name: ast.Name) -> Tuple[ast.Name, str]:
         # Display the repr of the name if it's a local variable or
         # _should_repr_global_name() thinks it's acceptable.
         locs = ast.Call(self.builtin("locals"), [], [])
@@ -901,7 +961,7 @@ class AssertionRewriter(ast.NodeVisitor):
         expr = ast.IfExp(test, self.display(name), ast.Str(name.id))
         return name, self.explanation_param(expr)
 
-    def visit_BoolOp(self, boolop):
+    def visit_BoolOp(self, boolop: ast.BoolOp) -> Tuple[ast.Name, str]:
         res_var = self.variable()
         expl_list = self.assign(ast.List([], ast.Load()))
         app = ast.Attribute(expl_list, "append", ast.Load())
@@ -913,7 +973,7 @@ class AssertionRewriter(ast.NodeVisitor):
         # Process each operand, short-circuiting if needed.
         for i, v in enumerate(boolop.values):
             if i:
-                fail_inner = []  # type: List[ast.stmt]
+                fail_inner: List[ast.stmt] = []
                 # cond is set in a prior loop iteration below
                 self.expl_stmts.append(ast.If(cond, fail_inner, []))  # noqa
                 self.expl_stmts = fail_inner
@@ -924,10 +984,10 @@ class AssertionRewriter(ast.NodeVisitor):
             call = ast.Call(app, [expl_format], [])
             self.expl_stmts.append(ast.Expr(call))
             if i < levels:
-                cond = res  # type: ast.expr
+                cond: ast.expr = res
                 if is_or:
                     cond = ast.UnaryOp(ast.Not(), cond)
-                inner = []  # type: List[ast.stmt]
+                inner: List[ast.stmt] = []
                 self.statements.append(ast.If(cond, inner, []))
                 self.statements = body = inner
         self.statements = save
@@ -936,24 +996,21 @@ class AssertionRewriter(ast.NodeVisitor):
         expl = self.pop_format_context(expl_template)
         return ast.Name(res_var, ast.Load()), self.explanation_param(expl)
 
-    def visit_UnaryOp(self, unary):
+    def visit_UnaryOp(self, unary: ast.UnaryOp) -> Tuple[ast.Name, str]:
         pattern = UNARY_MAP[unary.op.__class__]
         operand_res, operand_expl = self.visit(unary.operand)
         res = self.assign(ast.UnaryOp(unary.op, operand_res))
         return res, pattern % (operand_expl,)
 
-    def visit_BinOp(self, binop):
+    def visit_BinOp(self, binop: ast.BinOp) -> Tuple[ast.Name, str]:
         symbol = BINOP_MAP[binop.op.__class__]
         left_expr, left_expl = self.visit(binop.left)
         right_expr, right_expl = self.visit(binop.right)
-        explanation = "({} {} {})".format(left_expl, symbol, right_expl)
+        explanation = f"({left_expl} {symbol} {right_expl})"
         res = self.assign(ast.BinOp(left_expr, binop.op, right_expr))
         return res, explanation
 
-    def visit_Call(self, call):
-        """
-        visit `ast.Call` nodes
-        """
+    def visit_Call(self, call: ast.Call) -> Tuple[ast.Name, str]:
         new_func, func_expl = self.visit(call.func)
         arg_expls = []
         new_args = []
@@ -974,16 +1031,16 @@ class AssertionRewriter(ast.NodeVisitor):
         new_call = ast.Call(new_func, new_args, new_kwargs)
         res = self.assign(new_call)
         res_expl = self.explanation_param(self.display(res))
-        outer_expl = "{}\n{{{} = {}\n}}".format(res_expl, res_expl, expl)
+        outer_expl = f"{res_expl}\n{{{res_expl} = {expl}\n}}"
         return res, outer_expl
 
-    def visit_Starred(self, starred):
-        # From Python 3.5, a Starred node can appear in a function call
+    def visit_Starred(self, starred: ast.Starred) -> Tuple[ast.Starred, str]:
+        # A Starred node can appear in a function call.
         res, expl = self.visit(starred.value)
         new_starred = ast.Starred(res, starred.ctx)
         return new_starred, "*" + expl
 
-    def visit_Attribute(self, attr):
+    def visit_Attribute(self, attr: ast.Attribute) -> Tuple[ast.Name, str]:
         if not isinstance(attr.ctx, ast.Load):
             return self.generic_visit(attr)
         value, value_expl = self.visit(attr.value)
@@ -993,11 +1050,11 @@ class AssertionRewriter(ast.NodeVisitor):
         expl = pat % (res_expl, res_expl, value_expl, attr.attr)
         return res, expl
 
-    def visit_Compare(self, comp: ast.Compare):
+    def visit_Compare(self, comp: ast.Compare) -> Tuple[ast.expr, str]:
         self.push_format_context()
         left_res, left_expl = self.visit(comp.left)
         if isinstance(comp.left, (ast.Compare, ast.BoolOp)):
-            left_expl = "({})".format(left_expl)
+            left_expl = f"({left_expl})"
         res_variables = [self.variable() for i in range(len(comp.ops))]
         load_names = [ast.Name(v, ast.Load()) for v in res_variables]
         store_names = [ast.Name(v, ast.Store()) for v in res_variables]
@@ -1008,11 +1065,11 @@ class AssertionRewriter(ast.NodeVisitor):
         for i, op, next_operand in it:
             next_res, next_expl = self.visit(next_operand)
             if isinstance(next_operand, (ast.Compare, ast.BoolOp)):
-                next_expl = "({})".format(next_expl)
+                next_expl = f"({next_expl})"
             results.append(next_res)
             sym = BINOP_MAP[op.__class__]
             syms.append(ast.Str(sym))
-            expl = "{} {} {}".format(left_expl, sym, next_expl)
+            expl = f"{left_expl} {sym} {next_expl}"
             expls.append(ast.Str(expl))
             res_expr = ast.Compare(left_res, [op], [next_res])
             self.statements.append(ast.Assign([store_names[i]], res_expr))
@@ -1026,17 +1083,19 @@ class AssertionRewriter(ast.NodeVisitor):
             ast.Tuple(results, ast.Load()),
         )
         if len(comp.ops) > 1:
-            res = ast.BoolOp(ast.And(), load_names)  # type: ast.expr
+            res: ast.expr = ast.BoolOp(ast.And(), load_names)
         else:
             res = load_names[0]
         return res, self.explanation_param(self.pop_format_context(expl_call))
 
 
-def try_makedirs(cache_dir) -> bool:
-    """Attempts to create the given directory and sub-directories exist, returns True if
-    successful or it already exists"""
+def try_makedirs(cache_dir: Path) -> bool:
+    """Attempt to create the given directory and sub-directories exist.
+
+    Returns True if successful or if it already exists.
+    """
     try:
-        os.makedirs(fspath(cache_dir), exist_ok=True)
+        os.makedirs(os.fspath(cache_dir), exist_ok=True)
     except (FileNotFoundError, NotADirectoryError, FileExistsError):
         # One of the path components was not a directory:
         # - we're in a zip file
@@ -1053,7 +1112,7 @@ def try_makedirs(cache_dir) -> bool:
 
 
 def get_cache_dir(file_path: Path) -> Path:
-    """Returns the cache directory to write .pyc files for the given .py file path"""
+    """Return the cache directory to write .pyc files for the given .py file path."""
     if sys.version_info >= (3, 8) and sys.pycache_prefix:
         # given:
         #   prefix = '/tmp/pycs'
diff --git a/contrib/python/pytest/py3/_pytest/assertion/truncate.py b/contrib/python/pytest/py3/_pytest/assertion/truncate.py
index d97b05b441e..5ba9ddca75a 100644
--- a/contrib/python/pytest/py3/_pytest/assertion/truncate.py
+++ b/contrib/python/pytest/py3/_pytest/assertion/truncate.py
@@ -1,42 +1,47 @@
-"""
-Utilities for truncating assertion output.
+"""Utilities for truncating assertion output.
 
 Current default behaviour is to truncate assertion explanations at
 ~8 terminal lines, unless running in "-vv" mode or running on CI.
 """
 import os
+from typing import List
+from typing import Optional
+
+from _pytest.nodes import Item
+
 
 DEFAULT_MAX_LINES = 8
 DEFAULT_MAX_CHARS = 8 * 80
 USAGE_MSG = "use '-vv' to show"
 
 
-def truncate_if_required(explanation, item, max_length=None):
-    """
-    Truncate this assertion explanation if the given test item is eligible.
-    """
+def truncate_if_required(
+    explanation: List[str], item: Item, max_length: Optional[int] = None
+) -> List[str]:
+    """Truncate this assertion explanation if the given test item is eligible."""
     if _should_truncate_item(item):
         return _truncate_explanation(explanation)
     return explanation
 
 
-def _should_truncate_item(item):
-    """
-    Whether or not this test item is eligible for truncation.
-    """
+def _should_truncate_item(item: Item) -> bool:
+    """Whether or not this test item is eligible for truncation."""
     verbose = item.config.option.verbose
     return verbose < 2 and not _running_on_ci()
 
 
-def _running_on_ci():
+def _running_on_ci() -> bool:
     """Check if we're currently running on a CI system."""
     env_vars = ["CI", "BUILD_NUMBER"]
     return any(var in os.environ for var in env_vars)
 
 
-def _truncate_explanation(input_lines, max_lines=None, max_chars=None):
-    """
-    Truncate given list of strings that makes up the assertion explanation.
+def _truncate_explanation(
+    input_lines: List[str],
+    max_lines: Optional[int] = None,
+    max_chars: Optional[int] = None,
+) -> List[str]:
+    """Truncate given list of strings that makes up the assertion explanation.
 
     Truncates to either 8 lines, or 640 characters - whichever the input reaches
     first. The remaining lines will be replaced by a usage message.
@@ -65,15 +70,15 @@ def _truncate_explanation(input_lines, max_lines=None, max_chars=None):
     truncated_line_count += 1  # Account for the part-truncated final line
     msg = "...Full output truncated"
     if truncated_line_count == 1:
-        msg += " ({} line hidden)".format(truncated_line_count)
+        msg += f" ({truncated_line_count} line hidden)"
     else:
-        msg += " ({} lines hidden)".format(truncated_line_count)
-    msg += ", {}".format(USAGE_MSG)
+        msg += f" ({truncated_line_count} lines hidden)"
+    msg += f", {USAGE_MSG}"
     truncated_explanation.extend(["", str(msg)])
     return truncated_explanation
 
 
-def _truncate_by_char_count(input_lines, max_chars):
+def _truncate_by_char_count(input_lines: List[str], max_chars: int) -> List[str]:
     # Check if truncation required
     if len("".join(input_lines)) <= max_chars:
         return input_lines
diff --git a/contrib/python/pytest/py3/_pytest/assertion/util.py b/contrib/python/pytest/py3/_pytest/assertion/util.py
index 7d525aa4c42..da1ffd15e37 100644
--- a/contrib/python/pytest/py3/_pytest/assertion/util.py
+++ b/contrib/python/pytest/py3/_pytest/assertion/util.py
@@ -1,4 +1,4 @@
-"""Utilities for assertion debugging"""
+"""Utilities for assertion debugging."""
 import collections.abc
 import pprint
 from typing import AbstractSet
@@ -9,28 +9,26 @@ from typing import List
 from typing import Mapping
 from typing import Optional
 from typing import Sequence
-from typing import Tuple
 
 import _pytest._code
 from _pytest import outcomes
 from _pytest._io.saferepr import _pformat_dispatch
 from _pytest._io.saferepr import safeformat
 from _pytest._io.saferepr import saferepr
-from _pytest.compat import ATTRS_EQ_FIELD
 
 # The _reprcompare attribute on the util module is used by the new assertion
 # interpretation code and assertion rewriter to detect this plugin was
 # loaded and in turn call the hooks defined here as part of the
 # DebugInterpreter.
-_reprcompare = None  # type: Optional[Callable[[str, object, object], Optional[str]]]
+_reprcompare: Optional[Callable[[str, object, object], Optional[str]]] = None
 
 # Works similarly as _reprcompare attribute. Is populated with the hook call
 # when pytest_runtest_setup is called.
-_assertion_pass = None  # type: Optional[Callable[[int, str, str], None]]
+_assertion_pass: Optional[Callable[[int, str, str], None]] = None
 
 
 def format_explanation(explanation: str) -> str:
-    """This formats an explanation
+    r"""Format an explanation.
 
     Normally all embedded newlines are escaped, however there are
     three exceptions: \n{, \n} and \n~.  The first two are intended
@@ -45,7 +43,7 @@ def format_explanation(explanation: str) -> str:
 
 
 def _split_explanation(explanation: str) -> List[str]:
-    """Return a list of individual lines in the explanation
+    r"""Return a list of individual lines in the explanation.
 
     This will return a list of lines split on '\n{', '\n}' and '\n~'.
     Any other newlines will be escaped and appear in the line as the
@@ -62,11 +60,11 @@ def _split_explanation(explanation: str) -> List[str]:
 
 
 def _format_lines(lines: Sequence[str]) -> List[str]:
-    """Format the individual lines
+    """Format the individual lines.
 
-    This will replace the '{', '}' and '~' characters of our mini
-    formatting language with the proper 'where ...', 'and ...' and ' +
-    ...' text, taking care of indentation along the way.
+    This will replace the '{', '}' and '~' characters of our mini formatting
+    language with the proper 'where ...', 'and ...' and ' + ...' text, taking
+    care of indentation along the way.
 
     Return a list of formatted lines.
     """
@@ -112,6 +110,10 @@ def isset(x: Any) -> bool:
     return isinstance(x, (set, frozenset))
 
 
+def isnamedtuple(obj: Any) -> bool:
+    return isinstance(obj, tuple) and getattr(obj, "_fields", None) is not None
+
+
 def isdatacls(obj: Any) -> bool:
     return getattr(obj, "__dataclass_fields__", None) is not None
 
@@ -129,7 +131,7 @@ def isiterable(obj: Any) -> bool:
 
 
 def assertrepr_compare(config, op: str, left: Any, right: Any) -> Optional[List[str]]:
-    """Return specialised explanations for some operators/operands"""
+    """Return specialised explanations for some operators/operands."""
     verbose = config.getoption("verbose")
     if verbose > 1:
         left_repr = safeformat(left)
@@ -143,31 +145,12 @@ def assertrepr_compare(config, op: str, left: Any, right: Any) -> Optional[List[
         left_repr = saferepr(left, maxsize=maxsize)
         right_repr = saferepr(right, maxsize=maxsize)
 
-    summary = "{} {} {}".format(left_repr, op, right_repr)
+    summary = f"{left_repr} {op} {right_repr}"
 
     explanation = None
     try:
         if op == "==":
-            if istext(left) and istext(right):
-                explanation = _diff_text(left, right, verbose)
-            else:
-                if issequence(left) and issequence(right):
-                    explanation = _compare_eq_sequence(left, right, verbose)
-                elif isset(left) and isset(right):
-                    explanation = _compare_eq_set(left, right, verbose)
-                elif isdict(left) and isdict(right):
-                    explanation = _compare_eq_dict(left, right, verbose)
-                elif type(left) == type(right) and (isdatacls(left) or isattrs(left)):
-                    type_fn = (isdatacls, isattrs)
-                    explanation = _compare_eq_cls(left, right, verbose, type_fn)
-                elif verbose > 0:
-                    explanation = _compare_eq_verbose(left, right)
-                if isiterable(left) and isiterable(right):
-                    expl = _compare_eq_iterable(left, right, verbose)
-                    if explanation is not None:
-                        explanation.extend(expl)
-                    else:
-                        explanation = expl
+            explanation = _compare_eq_any(left, right, verbose)
         elif op == "not in":
             if istext(left) and istext(right):
                 explanation = _notin_text(left, right, verbose)
@@ -187,6 +170,33 @@ def assertrepr_compare(config, op: str, left: Any, right: Any) -> Optional[List[
     return [summary] + explanation
 
 
+def _compare_eq_any(left: Any, right: Any, verbose: int = 0) -> List[str]:
+    explanation = []
+    if istext(left) and istext(right):
+        explanation = _diff_text(left, right, verbose)
+    else:
+        if type(left) == type(right) and (
+            isdatacls(left) or isattrs(left) or isnamedtuple(left)
+        ):
+            # Note: unlike dataclasses/attrs, namedtuples compare only the
+            # field values, not the type or field names. But this branch
+            # intentionally only handles the same-type case, which was often
+            # used in older code bases before dataclasses/attrs were available.
+            explanation = _compare_eq_cls(left, right, verbose)
+        elif issequence(left) and issequence(right):
+            explanation = _compare_eq_sequence(left, right, verbose)
+        elif isset(left) and isset(right):
+            explanation = _compare_eq_set(left, right, verbose)
+        elif isdict(left) and isdict(right):
+            explanation = _compare_eq_dict(left, right, verbose)
+        elif verbose > 0:
+            explanation = _compare_eq_verbose(left, right)
+        if isiterable(left) and isiterable(right):
+            expl = _compare_eq_iterable(left, right, verbose)
+            explanation.extend(expl)
+    return explanation
+
+
 def _diff_text(left: str, right: str, verbose: int = 0) -> List[str]:
     """Return the explanation for the diff between text.
 
@@ -195,7 +205,7 @@ def _diff_text(left: str, right: str, verbose: int = 0) -> List[str]:
     """
     from difflib import ndiff
 
-    explanation = []  # type: List[str]
+    explanation: List[str] = []
 
     if verbose < 1:
         i = 0  # just in case left or right has zero length
@@ -240,7 +250,7 @@ def _compare_eq_verbose(left: Any, right: Any) -> List[str]:
     left_lines = repr(left).splitlines(keepends)
     right_lines = repr(right).splitlines(keepends)
 
-    explanation = []  # type: List[str]
+    explanation: List[str] = []
     explanation += ["+" + line for line in left_lines]
     explanation += ["-" + line for line in right_lines]
 
@@ -294,7 +304,7 @@ def _compare_eq_sequence(
     left: Sequence[Any], right: Sequence[Any], verbose: int = 0
 ) -> List[str]:
     comparing_bytes = isinstance(left, bytes) and isinstance(right, bytes)
-    explanation = []  # type: List[str]
+    explanation: List[str] = []
     len_left = len(left)
     len_right = len(right)
     for i in range(min(len_left, len_right)):
@@ -314,9 +324,7 @@ def _compare_eq_sequence(
                 left_value = left[i]
                 right_value = right[i]
 
-            explanation += [
-                "At index {} diff: {!r} != {!r}".format(i, left_value, right_value)
-            ]
+            explanation += [f"At index {i} diff: {left_value!r} != {right_value!r}"]
             break
 
     if comparing_bytes:
@@ -336,9 +344,7 @@ def _compare_eq_sequence(
             extra = saferepr(right[len_left])
 
         if len_diff == 1:
-            explanation += [
-                "{} contains one more item: {}".format(dir_with_more, extra)
-            ]
+            explanation += [f"{dir_with_more} contains one more item: {extra}"]
         else:
             explanation += [
                 "%s contains %d more items, first extra item: %s"
@@ -367,7 +373,7 @@ def _compare_eq_set(
 def _compare_eq_dict(
     left: Mapping[Any, Any], right: Mapping[Any, Any], verbose: int = 0
 ) -> List[str]:
-    explanation = []  # type: List[str]
+    explanation: List[str] = []
     set_left = set(left)
     set_right = set(right)
     common = set_left.intersection(set_right)
@@ -405,22 +411,19 @@ def _compare_eq_dict(
     return explanation
 
 
-def _compare_eq_cls(
-    left: Any,
-    right: Any,
-    verbose: int,
-    type_fns: Tuple[Callable[[Any], bool], Callable[[Any], bool]],
-) -> List[str]:
-    isdatacls, isattrs = type_fns
+def _compare_eq_cls(left: Any, right: Any, verbose: int) -> List[str]:
     if isdatacls(left):
         all_fields = left.__dataclass_fields__
         fields_to_check = [field for field, info in all_fields.items() if info.compare]
     elif isattrs(left):
         all_fields = left.__attrs_attrs__
-        fields_to_check = [
-            field.name for field in all_fields if getattr(field, ATTRS_EQ_FIELD)
-        ]
+        fields_to_check = [field.name for field in all_fields if getattr(field, "eq")]
+    elif isnamedtuple(left):
+        fields_to_check = left._fields
+    else:
+        assert False
 
+    indent = "  "
     same = []
     diff = []
     for field in fields_to_check:
@@ -430,6 +433,8 @@ def _compare_eq_cls(
             diff.append(field)
 
     explanation = []
+    if same or diff:
+        explanation += [""]
     if same and verbose < 2:
         explanation.append("Omitting %s identical items, use -vv to show" % len(same))
     elif same:
@@ -437,9 +442,18 @@ def _compare_eq_cls(
         explanation += pprint.pformat(same).splitlines()
     if diff:
         explanation += ["Differing attributes:"]
+        explanation += pprint.pformat(diff).splitlines()
         for field in diff:
+            field_left = getattr(left, field)
+            field_right = getattr(right, field)
+            explanation += [
+                "",
+                "Drill down into differing attribute %s:" % field,
+                ("%s%s: %r != %r") % (indent, field, field_left, field_right),
+            ]
             explanation += [
-                ("%s: %r != %r") % (field, getattr(left, field), getattr(right, field))
+                indent + line
+                for line in _compare_eq_any(field_left, field_right, verbose)
             ]
     return explanation