Increase coverage

codecov.yml

@@ -0,0 +1,9 @@
+coverage:
+  status:
+    project:
+      default:
+        target: auto
+        threshold: 1
+    patch:
+      default:
+        target: 0%

ipykernel/kernelbase.py

@@ -102,7 +102,7 @@ def _update_eventloop(self, change):
     banner: str
 
     @default("shell_streams")
-    def _shell_streams_default(self):
+    def _shell_streams_default(self):  # pragma: no cover
         warnings.warn(
             "Kernel.shell_streams is deprecated in ipykernel 6.0. Use Kernel.shell_stream",
             DeprecationWarning,
@@ -114,7 +114,7 @@ def _shell_streams_default(self):
             return []
 
     @observe("shell_streams")
-    def _shell_streams_changed(self, change):
+    def _shell_streams_changed(self, change):  # pragma: no cover
         warnings.warn(
             "Kernel.shell_streams is deprecated in ipykernel 6.0. Use Kernel.shell_stream",
             DeprecationWarning,
@@ -683,7 +683,6 @@ def finish_metadata(self, parent, metadata, reply_content):
 
     async def execute_request(self, stream, ident, parent):
         """handle an execute_request"""
-
         try:
             content = parent["content"]
             code = content["code"]
@@ -947,7 +946,6 @@ def do_is_complete(self, code):
 
     async def debug_request(self, stream, ident, parent):
         content = parent["content"]
-
         reply_content = self.do_debug_request(content)
         if inspect.isawaitable(reply_content):
             reply_content = await reply_content
@@ -1006,7 +1004,7 @@ async def do_debug_request(self, msg):
     # Engine methods (DEPRECATED)
     # ---------------------------------------------------------------------------
 
-    async def apply_request(self, stream, ident, parent):
+    async def apply_request(self, stream, ident, parent):  # pragma: no cover
         self.log.warning("apply_request is deprecated in kernel_base, moving to ipyparallel.")
         try:
             content = parent["content"]
@@ -1044,7 +1042,7 @@ def do_apply(self, content, bufs, msg_id, reply_metadata):
     # Control messages (DEPRECATED)
     # ---------------------------------------------------------------------------
 
-    async def abort_request(self, stream, ident, parent):
+    async def abort_request(self, stream, ident, parent):  # pragma: no cover
         """abort a specific msg by id"""
         self.log.warning(
             "abort_request is deprecated in kernel_base. It is only part of IPython parallel"
@@ -1063,7 +1061,7 @@ async def abort_request(self, stream, ident, parent):
         )
         self.log.debug("%s", reply_msg)
 
-    async def clear_request(self, stream, idents, parent):
+    async def clear_request(self, stream, idents, parent):  # pragma: no cover
         """Clear our namespace."""
         self.log.warning(
             "clear_request is deprecated in kernel_base. It is only part of IPython parallel"

ipykernel/tests/conftest.py

@@ -1,6 +1,15 @@
 import asyncio
+import logging
 import os
 
+import pytest
+import zmq
+from jupyter_client.session import Session
+from tornado.ioloop import IOLoop
+from zmq.eventloop.zmqstream import ZMQStream
+
+from ipykernel.kernelbase import Kernel
+
 try:
     import resource
 except ImportError:
@@ -26,3 +35,97 @@
 # Enforce selector event loop on Windows.
 if os.name == "nt":
     asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+
+class TestKernel(Kernel):
+    implementation = "test"
+    implementation_version = "1.0"
+    language = "no-op"
+    language_version = "0.1"
+    language_info = {
+        "name": "test",
+        "mimetype": "text/plain",
+        "file_extension": ".txt",
+    }
+    banner = "test kernel"
+    log = logging.getLogger()
+
+    def __init__(self, *args, **kwargs):
+        self.context = context = zmq.Context()
+        self.iopub_socket = context.socket(zmq.PUB)
+        self.session = Session()
+        self.test_sockets = [self.iopub_socket]
+        self.test_streams = []
+
+        for name in ["shell", "control"]:
+            socket = context.socket(zmq.ROUTER)
+            stream = ZMQStream(socket)
+            stream.on_send(self._on_send)
+            self.test_sockets.append(socket)
+            self.test_streams.append(stream)
+            setattr(self, f"{name}_stream", stream)
+        super().__init__(*args, **kwargs)
+
+    def do_execute(
+        self, code, silent, store_history=True, user_expressions=None, allow_stdin=False
+    ):
+        if not silent:
+            stream_content = {"name": "stdout", "text": code}
+            self.send_response(self.iopub_socket, "stream", stream_content)
+
+        return {
+            "status": "ok",
+            # The base class increments the execution count
+            "execution_count": self.execution_count,
+            "payload": [],
+            "user_expressions": {},
+        }
+
+    async def do_debug_request(self, msg):
+        return {}
+
+    async def test_shell_message(self, *args, **kwargs):
+        msg_list = self._prep_msg(*args, **kwargs)
+        await self.dispatch_shell(msg_list)
+        self.shell_stream.flush()
+        return await self._wait_for_msg()
+
+    async def test_control_message(self, *args, **kwargs):
+        msg_list = self._prep_msg(*args, **kwargs)
+        await self.process_control(msg_list)
+        self.control_stream.flush()
+        return await self._wait_for_msg()
+
+    def destroy(self):
+        for stream in self.test_streams:
+            stream.close()
+        for socket in self.test_sockets:
+            socket.close()
+        self.context.destroy()
+
+    def _on_send(self, msg, *args, **kwargs):
+        self._reply = msg
+
+    def _prep_msg(self, *args, **kwargs):
+        self._reply = None
+        msg = self.session.msg(*args, **kwargs)
+        msg = self.session.serialize(msg)
+        return [zmq.Message(m) for m in msg]
+
+    async def _wait_for_msg(self):
+        while not self._reply:
+            await asyncio.sleep(0.1)
+        _, msg = self.session.feed_identities(self._reply)
+        return self.session.deserialize(msg)
+
+    def _send_interupt_children(self):
+        # override to prevent deadlock
+        pass
+
+
+@pytest.fixture
+async def kernel():
+    kernel = TestKernel()
+    kernel.io_loop = IOLoop.current()
+    yield kernel
+    kernel.destroy()

ipykernel/tests/test_eventloop.py

@@ -1,8 +1,15 @@
 """Test eventloop integration"""
 
+import asyncio
+import os
+import threading
+import time
+
 import pytest
 import tornado
 
+from ipykernel.eventloops import enable_gui, loop_asyncio, loop_tk
+
 from .utils import execute, flush_channels, start_new_kernel
 
 KC = KM = None
@@ -41,3 +48,41 @@ def test_asyncio_interrupt():
     flush_channels(KC)
     msg_id, content = execute(async_code, KC)
     assert content["status"] == "ok"
+
+
+windows_skip = pytest.mark.skipif(os.name == "nt", reason="causing failures on windows")
+
+
+@windows_skip
+def test_tk_loop(kernel):
+    def do_thing():
+        time.sleep(1)
+        try:
+            kernel.app_wrapper.app.quit()
+        # guard for tk failing to start (if there is no display)
+        except AttributeError:
+            pass
+
+    t = threading.Thread(target=do_thing)
+    t.start()
+    # guard for tk failing to start (if there is no display)
+    try:
+        loop_tk(kernel)
+    except Exception:
+        pass
+    t.join()
+
+
+@windows_skip
+def test_asyncio_loop(kernel):
+    def do_thing():
+        loop.call_soon(loop.stop)
+
+    loop = asyncio.get_event_loop()
+    loop.call_soon(do_thing)
+    loop_asyncio(kernel)
+
+
+@windows_skip
+def test_enable_gui(kernel):
+    enable_gui("tk", kernel)

ipykernel/tests/test_kernel_direct.py

@@ -0,0 +1,78 @@
+"""test the IPython Kernel"""
+
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
+
+import os
+
+import pytest
+
+if os.name == "nt":
+    pytest.skip("skipping tests on windows", allow_module_level=True)
+
+
+async def test_direct_kernel_info_request(kernel):
+    reply = await kernel.test_shell_message("kernel_info_request", {})
+    assert reply["header"]["msg_type"] == "kernel_info_reply"
+
+
+async def test_direct_execute_request(kernel):
+    reply = await kernel.test_shell_message("execute_request", dict(code="hello", silent=False))
+    assert reply["header"]["msg_type"] == "execute_reply"
+
+
+async def test_direct_execute_request_aborting(kernel):
+    kernel._aborting = True
+    reply = await kernel.test_shell_message("execute_request", dict(code="hello", silent=False))
+    assert reply["header"]["msg_type"] == "execute_reply"
+    assert reply["content"]["status"] == "aborted"
+
+
+async def test_complete_request(kernel):
+    reply = await kernel.test_shell_message("complete_request", dict(code="hello", cursor_pos=0))
+    assert reply["header"]["msg_type"] == "complete_reply"
+
+
+async def test_inspect_request(kernel):
+    reply = await kernel.test_shell_message("inspect_request", dict(code="hello", cursor_pos=0))
+    assert reply["header"]["msg_type"] == "inspect_reply"
+
+
+async def test_history_request(kernel):
+    reply = await kernel.test_shell_message(
+        "history_request", dict(hist_access_type="", output="", raw="")
+    )
+    assert reply["header"]["msg_type"] == "history_reply"
+
+
+async def test_comm_info_request(kernel):
+    reply = await kernel.test_shell_message("comm_info_request")
+    assert reply["header"]["msg_type"] == "comm_info_reply"
+
+
+async def test_direct_interrupt_request(kernel):
+    reply = await kernel.test_shell_message("interrupt_request", {})
+    assert reply["header"]["msg_type"] == "interrupt_reply"
+
+
+async def test_direct_shutdown_request(kernel):
+    reply = await kernel.test_shell_message("shutdown_request", dict(restart=False))
+    assert reply["header"]["msg_type"] == "shutdown_reply"
+    reply = await kernel.test_shell_message("shutdown_request", dict(restart=True))
+    assert reply["header"]["msg_type"] == "shutdown_reply"
+
+
+async def test_is_complete_request(kernel):
+    reply = await kernel.test_shell_message("is_complete_request", dict(code="hello"))
+    assert reply["header"]["msg_type"] == "is_complete_reply"
+
+
+async def test_direct_debug_request(kernel):
+    reply = await kernel.test_control_message("debug_request", {})
+    assert reply["header"]["msg_type"] == "debug_reply"
+
+
+# TODO: this causes deadlock
+# async def test_direct_usage_request(kernel):
+#     reply = await kernel.test_control_message("usage_request", {})
+#     assert reply['header']['msg_type'] == 'usage_reply'

pyproject.toml

@@ -51,7 +51,8 @@ test = [
     "flaky",
     "ipyparallel",
     "pre-commit",
-    "pytest-timeout",
+    "pytest-asyncio",
+    "pytest-timeout"
 ]
 
 [tool.hatch.version]
@@ -110,6 +111,7 @@ testpaths = [
     "ipykernel/tests",
     "ipykernel/inprocess/tests"
 ]
+asyncio_mode = "auto"
 timeout = 300
 # Restore this setting to debug failures
 # timeout_method = "thread"
@@ -127,6 +129,20 @@ filterwarnings= [
   "module:Jupyter is migrating its paths to use standard platformdirs:DeprecationWarning",
 ]
 
+[tool.coverage.report]
+exclude_lines = [
+  "pragma: no cover",
+  "def __repr__",
+  "if self.debug:",
+  "if settings.DEBUG",
+  "raise AssertionError",
+  "raise NotImplementedError",
+  "if 0:",
+  "if __name__ == .__main__.:",
+  "class .*\bProtocol\\):",
+  "@(abc\\.)?abstractmethod",
+]
+
 [tool.flake8]
 ignore = "E501, W503, E402"
 builtins = "c, get_config"