Clean up testing and coverage (#1030)

Author: blink1073

.coveragerc

@@ -3,7 +3,9 @@
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+import asyncio
 import os
+import warnings
 
 import pytest
 
@@ -72,6 +74,69 @@ async def test_direct_debug_request(kernel):
     assert reply["header"]["msg_type"] == "debug_reply"
 
 
+async def test_deprecated_features(kernel):
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        header = kernel._parent_header
+        assert isinstance(header, dict)
+        shell_streams = kernel.shell_streams
+        assert len(shell_streams) == 1
+        assert shell_streams[0] == kernel.shell_stream
+        warnings.simplefilter("ignore", RuntimeWarning)
+        kernel.shell_streams = [kernel.shell_stream, kernel.shell_stream]
+
+
+async def test_process_control(kernel):
+    from jupyter_client.session import DELIM
+
+    class FakeMsg:
+        def __init__(self, bytes):
+            self.bytes = bytes
+
+    await kernel.process_control([FakeMsg(DELIM), 1])
+    msg = kernel._prep_msg("does_not_exist")
+    await kernel.process_control(msg)
+
+
+def test_should_handle(kernel):
+    msg = kernel.session.msg("debug_request", {})
+    kernel.aborted.add(msg["header"]["msg_id"])
+    assert not kernel.should_handle(kernel.control_stream, msg, [])
+
+
+async def test_dispatch_shell(kernel):
+    from jupyter_client.session import DELIM
+
+    class FakeMsg:
+        def __init__(self, bytes):
+            self.bytes = bytes
+
+    await kernel.dispatch_shell([FakeMsg(DELIM), 1])
+    msg = kernel._prep_msg("does_not_exist")
+    await kernel.dispatch_shell(msg)
+
+
+async def test_enter_eventloop(kernel):
+    kernel.eventloop = None
+    kernel.enter_eventloop()
+    kernel.eventloop = asyncio.get_running_loop()
+    kernel.enter_eventloop()
+    called = 0
+
+    def check_status():
+        nonlocal called
+        if called == 0:
+            msg = kernel.session.msg("debug_request", {})
+            kernel.msg_queue.put(msg)
+        called += 1
+        kernel.io_loop.call_later(0.001, check_status)
+
+    kernel.io_loop.call_later(0.001, check_status)
+    kernel.start()
+    while called < 2:
+        await asyncio.sleep(0.1)
+
+
 # TODO: this causes deadlock
 # async def test_direct_usage_request(kernel):
 #     reply = await kernel.test_control_message("usage_request", {})

ipykernel/tests/test_kernel_direct.py

@@ -81,7 +81,7 @@ nowarn = "test -W default {args}"
 
 [tool.hatch.envs.cov]
 features = ["test"]
-dependencies = ["coverage", "pytest-cov", "matplotlib", "curio", "trio"]
+dependencies = ["coverage[toml]", "pytest-cov", "matplotlib", "curio", "trio"]
 [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}"
@@ -143,6 +143,13 @@ exclude_lines = [
   "@(abc\\.)?abstractmethod",
 ]
 
+[tool.coverage.run]
+omit = [
+  "ipykernel/tests/*",
+  "ipykernel/debugger.py",
+  "ipykernel/eventloops.py",
+]
+
 [tool.flake8]
 ignore = "E501, W503, E402"
 builtins = "c, get_config"