cleanup

ipykernel/eventloops.py

@@ -429,7 +429,7 @@ async def close_loop():
         close_loop()
 
     elif not loop.is_closed():
-        loop.run_until_complete(close_loop)  # type:ignore
+        loop.run_until_complete(close_loop)  # type:ignore[call-overload]
         loop.close()
 
 
@@ -566,8 +566,10 @@ def enable_gui(gui, kernel=None):
             make_qt_app_for_kernel(gui, kernel)
 
     loop = loop_map[gui]
-    if loop and kernel.eventloop is not None and kernel.eventloop is not loop:
-        msg = "Cannot activate multiple GUI eventloops"
+    if (
+        loop and kernel.eventloop is not None and kernel.eventloop is not loop
+    ):  # type:ignore[unreachable]
+        msg = "Cannot activate multiple GUI eventloops"  # type:ignore[unreachable]
         raise RuntimeError(msg)
     kernel.eventloop = loop
     # We set `eventloop`; the function the user chose is executed in `Kernel.enter_eventloop`, thus

ipykernel/inprocess/client.py

@@ -193,7 +193,7 @@ def _dispatch_to_kernel(self, msg):
             dispatch_shell = run_sync(kernel.dispatch_shell)
             dispatch_shell(msg_parts)
         else:
-            loop = asyncio.get_event_loop()
+            loop = asyncio.get_event_loop()  # type:ignore[unreachable]
             loop.run_until_complete(kernel.dispatch_shell(msg_parts))
         idents, reply_msg = self.session.recv(stream, copy=False)
         self.shell_channel.call_handlers_later(reply_msg)

ipykernel/inprocess/ipkernel.py

@@ -111,7 +111,7 @@ def _input_request(self, prompt, ident, parent, password=False):
         # Await a response.
         while self.raw_input_str is None:
             frontend.stdin_channel.process_events()
-        return self.raw_input_str
+        return self.raw_input_str  # type:ignore[unreachable]
 
     # -------------------------------------------------------------------------
     # Protected interface

ipykernel/iostream.py

@@ -93,7 +93,7 @@ def _start_event_gc():
         if self._event_pipe_gc_task is not None:
             # cancel gc task to avoid pending task warnings
             async def _cancel():
-                self._event_pipe_gc_task.cancel()  # type:ignore
+                self._event_pipe_gc_task.cancel()  # type:ignore[union-attr]
 
             if not self._stopped:
                 self.io_loop.run_sync(_cancel)
@@ -634,7 +634,7 @@ def write(self, string: str) -> Optional[int]:  # type:ignore[override]
         """
 
         if not isinstance(string, str):
-            msg = f"write() argument must be str, not {type(string)}"
+            msg = f"write() argument must be str, not {type(string)}"  # type:ignore[unreachable]
             raise TypeError(msg)
 
         if self.echo is not None:

ipykernel/ipkernel.py

@@ -29,7 +29,7 @@
 try:
     from IPython.core.interactiveshell import _asyncio_runner  # type:ignore[attr-defined]
 except ImportError:
-    _asyncio_runner = None  # type:ignore
+    _asyncio_runner = None
 
 try:
     from IPython.core.completer import provisionalcompleter as _provisionalcompleter
@@ -625,7 +625,7 @@ def do_apply(self, content, bufs, msg_id, reply_metadata):
         try:
             from ipyparallel.serialize import serialize_object, unpack_apply_message
         except ImportError:
-            from .serialize import serialize_object, unpack_apply_message
+            from .serialize import serialize_object, unpack_apply_message  # type:ignore[no-redef]
 
         shell = self.shell
         try:

ipykernel/kernelapp.py

@@ -460,9 +460,9 @@ def init_blackhole(self):
         if self.no_stdout or self.no_stderr:
             blackhole = open(os.devnull, "w")  # noqa
             if self.no_stdout:
-                sys.stdout = sys.__stdout__ = blackhole  # type:ignore
+                sys.stdout = sys.__stdout__ = blackhole
             if self.no_stderr:
-                sys.stderr = sys.__stderr__ = blackhole  # type:ignore
+                sys.stderr = sys.__stderr__ = blackhole
 
     def init_io(self):
         """Redirect input streams and set a display hook."""
@@ -670,8 +670,8 @@ def init_pdb(self):
 
         if hasattr(debugger, "InterruptiblePdb"):
             # Only available in newer IPython releases:
-            debugger.Pdb = debugger.InterruptiblePdb  # type:ignore
-            pdb.Pdb = debugger.Pdb  # type:ignore
+            debugger.Pdb = debugger.InterruptiblePdb  # type:ignore[misc]
+            pdb.Pdb = debugger.Pdb  # type:ignore[assignment,misc]
             pdb.set_trace = debugger.set_trace  # type:ignore[assignment]
 
     @catch_config_error

ipykernel/parentpoller.py

@@ -66,7 +66,7 @@ def __init__(self, interrupt_handle=None, parent_handle=None):
         assert interrupt_handle or parent_handle
         super().__init__()
         if ctypes is None:
-            msg = "ParentPollerWindows requires ctypes"
+            msg = "ParentPollerWindows requires ctypes"  # type:ignore[unreachable]
             raise ImportError(msg)
         self.daemon = True
         self.interrupt_handle = interrupt_handle

ipykernel/trio_runner.py

@@ -65,6 +65,6 @@ async def loc(coro):
             self._cell_cancel_scope = trio.CancelScope()
             with self._cell_cancel_scope:
                 return await coro
-            self._cell_cancel_scope = None
+            self._cell_cancel_scope = None  # type:ignore[unreachable]
 
         return trio.from_thread.run(loc, async_fn, trio_token=self._trio_token)

pyproject.toml

@@ -86,7 +86,7 @@ path = "ipykernel/_version.py"
 features = ["docs"]
 [tool.hatch.envs.docs.scripts]
 build = "make -C docs html SPHINXOPTS='-W'"
-api = "sphinx-apidoc -o docs/api -f -E ipykernel ipykernel/tests ipykernel/inprocess/tests ipykernel/datapub.py ipykernel/pickleutil.py ipykernel/serialize.py ipykernel/gui ipykernel/pylab"
+api = "sphinx-apidoc -o docs/api -f -E ipykernel tests ipykernel/datapub.py ipykernel/pickleutil.py ipykernel/serialize.py ipykernel/gui ipykernel/pylab"
 
 [tool.hatch.envs.test]
 features = ["test"]
@@ -151,6 +151,11 @@ warn_return_any = true
 warn_unreachable = true
 warn_unused_ignores = true
 
+[[tool.mypy.overrides]]
+module = "tests.*"
+disable_error_code = ["ignore-without-code"]
+warn_unreachable = false
+
 [tool.pytest.ini_options]
 minversion = "6.0"
 xfail_strict = true

tests/test_debugger.py

@@ -121,7 +121,11 @@ def test_set_breakpoints(kernel_with_debug):
     assert reply["body"]["breakpoints"][0]["source"]["path"] == source
 
     r = wait_for_debug_request(kernel_with_debug, "debugInfo")
-    assert source in map(lambda b: b["source"], r["body"]["breakpoints"])  # noqa
+
+    def func(b):
+        return b["source"]
+
+    assert source in map(func, r["body"]["breakpoints"])
 
     r = wait_for_debug_request(kernel_with_debug, "configurationDone")
     assert r["success"]
@@ -208,7 +212,11 @@ def test_rich_inspect_not_at_breakpoint(kernel_with_debug):
     get_reply(kernel_with_debug, msg_id)
 
     r = wait_for_debug_request(kernel_with_debug, "inspectVariables")
-    assert var_name in list(map(lambda v: v["name"], r["body"]["variables"]))  # noqa
+
+    def func(v):
+        return v["name"]
+
+    assert var_name in list(map(func, r["body"]["variables"]))
 
     reply = wait_for_debug_request(
         kernel_with_debug,

tests/test_eventloop.py

@@ -57,7 +57,7 @@ def teardown():
 
 
 async_code = """
-from ipykernel.tests._asyncio_utils import async_func
+from tests._asyncio_utils import async_func
 async_func()
 """