Add more coverage and add Readme badges (#1031)

Author: blink1073

README.md

@@ -1,5 +1,9 @@
 # IPython Kernel for Jupyter
 
+[![Build Status](https://github.com/ipython/ipykernel/actions/workflows/ci.yml/badge.svg?query=branch%3Amain++)](https://github.com/ipython/ipykernel/actions/workflows/ci.yml/badge.svg?query=branch%3Amain++)
+[![codecov](https://codecov.io/gh/ipython/ipykernel/branch/main/graph/badge.svg?token=SyksDOcIJa)](https://codecov.io/gh/ipython/ipykernel)
+[![Documentation Status](https://readthedocs.org/projects/ipython/badge/?version=latest)](http://ipython.readthedocs.io/en/latest/?badge=latest)
+
 This package provides the IPython kernel for Jupyter.
 
 ## Installation from source

ipykernel/ipkernel.py

@@ -659,7 +659,7 @@ def do_clear(self):
 
 
 class Kernel(IPythonKernel):
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs):  # pragma: no cover
         import warnings
 
         warnings.warn(

ipykernel/tests/conftest.py

@@ -8,7 +8,9 @@
 from tornado.ioloop import IOLoop
 from zmq.eventloop.zmqstream import ZMQStream
 
+from ipykernel.ipkernel import IPythonKernel
 from ipykernel.kernelbase import Kernel
+from ipykernel.zmqshell import ZMQInteractiveShell
 
 try:
     import resource
@@ -37,20 +39,10 @@
     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"
+class KernelMixin:
     log = logging.getLogger()
 
-    def __init__(self, *args, **kwargs):
+    def _initialize(self):
         self.context = context = zmq.Context()
         self.iopub_socket = context.socket(zmq.PUB)
         self.session = Session()
@@ -64,7 +56,6 @@ def __init__(self, *args, **kwargs):
             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
@@ -123,9 +114,41 @@ def _send_interupt_children(self):
         pass
 
 
+class TestKernel(KernelMixin, 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"
+
+    def __init__(self, *args, **kwargs):
+        self._initialize()
+        super().__init__(*args, **kwargs)
+
+
+class TestIPyKernel(KernelMixin, IPythonKernel):
+    def __init__(self, *args, **kwargs):
+        self._initialize()
+        super().__init__(*args, **kwargs)
+
+
 @pytest.fixture
 async def kernel():
     kernel = TestKernel()
     kernel.io_loop = IOLoop.current()
     yield kernel
     kernel.destroy()
+
+
+@pytest.fixture
+async def ipkernel():
+    kernel = TestIPyKernel()
+    kernel.io_loop = IOLoop.current()
+    yield kernel
+    kernel.destroy()
+    ZMQInteractiveShell.clear_instance()

ipykernel/tests/test_ipkernel_direct.py

@@ -0,0 +1,98 @@
+"""Test IPythonKernel directly"""
+
+import asyncio
+import os
+
+import pytest
+
+from ipykernel.ipkernel import IPythonKernel
+
+if os.name == "nt":
+    pytest.skip("skipping tests on windows", allow_module_level=True)
+
+
+class user_mod:
+    __dict__ = {}
+
+
+async def test_properities(ipkernel: IPythonKernel):
+    ipkernel.user_module = user_mod()
+    ipkernel.user_ns = {}
+
+
+async def test_direct_kernel_info_request(ipkernel):
+    reply = await ipkernel.test_shell_message("kernel_info_request", {})
+    assert reply["header"]["msg_type"] == "kernel_info_reply"
+
+
+async def test_direct_execute_request(ipkernel):
+    reply = await ipkernel.test_shell_message("execute_request", dict(code="hello", silent=False))
+    assert reply["header"]["msg_type"] == "execute_reply"
+
+
+async def test_direct_execute_request_aborting(ipkernel):
+    ipkernel._aborting = True
+    reply = await ipkernel.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(ipkernel):
+    reply = await ipkernel.test_shell_message("complete_request", dict(code="hello", cursor_pos=0))
+    assert reply["header"]["msg_type"] == "complete_reply"
+
+
+async def test_inspect_request(ipkernel):
+    reply = await ipkernel.test_shell_message("inspect_request", dict(code="hello", cursor_pos=0))
+    assert reply["header"]["msg_type"] == "inspect_reply"
+
+
+async def test_history_request(ipkernel):
+    reply = await ipkernel.test_shell_message(
+        "history_request", dict(hist_access_type="", output="", raw="")
+    )
+    assert reply["header"]["msg_type"] == "history_reply"
+
+
+async def test_comm_info_request(ipkernel):
+    reply = await ipkernel.test_shell_message("comm_info_request")
+    assert reply["header"]["msg_type"] == "comm_info_reply"
+
+
+async def test_direct_interrupt_request(ipkernel):
+    reply = await ipkernel.test_shell_message("interrupt_request", {})
+    assert reply["header"]["msg_type"] == "interrupt_reply"
+
+
+# TODO: this causes deadlock
+# async def test_direct_shutdown_request(ipkernel):
+#     reply = await ipkernel.test_shell_message("shutdown_request", dict(restart=False))
+#     assert reply["header"]["msg_type"] == "shutdown_reply"
+#     reply = await ipkernel.test_shell_message("shutdown_request", dict(restart=True))
+#     assert reply["header"]["msg_type"] == "shutdown_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'
+
+
+async def test_is_complete_request(ipkernel):
+    reply = await ipkernel.test_shell_message("is_complete_request", dict(code="hello"))
+    assert reply["header"]["msg_type"] == "is_complete_reply"
+
+
+async def test_direct_debug_request(ipkernel):
+    reply = await ipkernel.test_control_message("debug_request", {})
+    assert reply["header"]["msg_type"] == "debug_reply"
+
+
+async def test_direct_clear(ipkernel):
+    ipkernel.do_clear()
+
+
+async def test_cancel_on_sigint(ipkernel: IPythonKernel):
+    future = asyncio.Future()
+    with ipkernel._cancel_on_sigint(future):
+        pass
+    future.set_result(None)

pyproject.toml

@@ -55,6 +55,13 @@ test = [
     "pytest-asyncio",
     "pytest-timeout"
 ]
+cov = [
+  "coverage[toml]",
+  "pytest-cov",
+  "matplotlib",
+  "curio",
+  "trio",
+]
 
 [tool.hatch.version]
 path = "ipykernel/_version.py"
@@ -80,8 +87,7 @@ test = "python -m pytest -vv {args}"
 nowarn = "test -W default {args}"
 
 [tool.hatch.envs.cov]
-features = ["test"]
-dependencies = ["coverage[toml]", "pytest-cov", "matplotlib", "curio", "trio"]
+features = ["test", "cov"]
 [tool.hatch.envs.cov.scripts]
 test = "python -m pytest -vv --cov ipykernel --cov-branch --cov-report term-missing:skip-covered {args}"
 nowarn = "test -W default {args}"
@@ -148,6 +154,7 @@ omit = [
   "ipykernel/tests/*",
   "ipykernel/debugger.py",
   "ipykernel/eventloops.py",
+  "ipykernel/pickleutil.py"
 ]
 
 [tool.flake8]