Fix kernel lifecycle with warnings treated as errors

ipykernel/ipkernel.py

@@ -645,7 +645,7 @@ def do_history(
             "history": list(hist),
         }
 
-    def do_shutdown(self, restart):
+    async def do_shutdown(self, restart):
         """Handle kernel shutdown."""
         if self.shell:
             self.shell.exit_now = True

ipykernel/kernelapp.py

@@ -4,6 +4,7 @@
 # Distributed under the terms of the Modified BSD License.
 from __future__ import annotations
 
+import asyncio
 import atexit
 import errno
 import logging
@@ -12,6 +13,7 @@
 import sys
 import traceback
 import typing as t
+import warnings
 from functools import partial
 from io import FileIO, TextIOWrapper
 from logging import StreamHandler
@@ -758,6 +760,12 @@ def start(self):
             return self.subapp.start()
         if self.poller is not None:
             self.poller.start()
+        with warnings.catch_warnings():
+            warnings.filterwarnings("error", "There is no current event loop", DeprecationWarning)
+            try:
+                asyncio.get_event_loop()
+            except (RuntimeError, DeprecationWarning):
+                asyncio.set_event_loop(asyncio.new_event_loop())
         self.kernel.start()
         self.io_loop = ioloop.IOLoop.current()
         if self.trio_loop:

tests/test_ipkernel_direct.py

@@ -108,6 +108,12 @@ def raiseOSError():
     assert len(reply["content"]["traceback"]) > 0
 
 
+@pytest.mark.parametrize("restart", [False, True])
+async def test_do_shutdown(ipkernel, restart):
+    assert await ipkernel.do_shutdown(restart) == {"status": "ok", "restart": restart}
+    assert ipkernel.shell.exit_now
+
+
 # TODO: this causes deadlock
 # async def test_direct_shutdown_request(ipkernel):
 #     reply = await ipkernel.test_shell_message("shutdown_request", dict(restart=False))

tests/test_kernel.py

@@ -661,6 +661,22 @@ def test_shutdown():
         assert not km.is_alive()
 
 
+@pytest.mark.parametrize("restart", [False, True])
+def test_lifecycle_with_warnings_as_errors(monkeypatch, restart):
+    # Avoid jupyter_core's unrelated path-migration warning in the child.
+    monkeypatch.setenv("JUPYTER_PLATFORM_DIRS", "1")
+    monkeypatch.setenv("PYTHONWARNINGS", "error")
+    with new_kernel() as kc:
+        _msg_id, content = execute("answer = 6 * 7", kc=kc)
+        assert content["status"] == "ok"
+        wait_for_idle(kc)
+        msg_id = kc.shutdown(restart=restart)
+        reply = get_reply(kc, msg_id, timeout=10, channel="control")
+        assert reply["content"] == {"status": "ok", "restart": restart}
+        kc.parent.provisioner.process.wait(timeout=10)
+        assert kc.parent.provisioner.process.returncode == 0
+
+
 def test_interrupt_during_input():
     """
     The kernel exits after being interrupted while waiting in input().

tests/test_kernelapp.py

@@ -1,7 +1,10 @@
 import json
 import os
+import subprocess
+import sys
 import threading
 import time
+from textwrap import dedent
 from unittest.mock import patch
 
 import pytest
@@ -54,6 +57,40 @@ def trigger_stop():
     app.close()
 
 
+@pytest.mark.parametrize("loop_state", ["unset", "cleared", "existing"])
+def test_start_app_event_loop(loop_state, monkeypatch):
+    # Avoid jupyter_core's unrelated path-migration warning in the child.
+    monkeypatch.setenv("JUPYTER_PLATFORM_DIRS", "1")
+    code = dedent(
+        """
+        import asyncio
+        import sys
+        from unittest.mock import Mock
+        from tornado.ioloop import IOLoop
+        from ipykernel.kernelapp import IPKernelApp
+
+        expected = None
+        if sys.argv[1] == "cleared":
+            asyncio.set_event_loop(None)
+        elif sys.argv[1] == "existing":
+            expected = asyncio.new_event_loop()
+            asyncio.set_event_loop(expected)
+        app = IPKernelApp()
+        app.kernel = Mock()
+        def start_kernel():
+            loop = IOLoop.current().asyncio_loop
+            loop.call_soon(loop.stop)
+        app.kernel.start.side_effect = start_kernel
+        app.start()
+        assert app.io_loop.asyncio_loop is asyncio.get_event_loop()
+        if expected is not None:
+            assert app.io_loop.asyncio_loop is expected
+        app.io_loop.close()
+        """
+    )
+    subprocess.run([sys.executable, "-Werror", "-c", code, loop_state], check=True, timeout=10)
+
+
 @pytest.mark.skipif(os.name == "nt", reason="permission errors on windows")
 def test_merge_connection_file():
     cfg = Config()