From ad97a60002e93f3b260f4f69df5e20b2652019f8 Mon Sep 17 00:00:00 2001
From: jakkdl <h6+github@pm.me>
Date: Wed, 19 Jul 2023 14:05:24 +0200
Subject: [PATCH] add KernelManager.exit_status

---
 jupyter_client/manager.py   | 11 ++++++++++
 tests/test_kernelmanager.py | 40 +++++++++++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+)

diff --git a/jupyter_client/manager.py b/jupyter_client/manager.py
index f04bd987..8626407d 100644
--- a/jupyter_client/manager.py
+++ b/jupyter_client/manager.py
@@ -655,6 +655,17 @@ async def _async_is_alive(self) -> bool:
 
     is_alive = run_sync(_async_is_alive)
 
+    async def _async_exit_status(self) -> int | None:
+        """Returns 0 if there's no kernel or it exited gracefully,
+        None if the kernel is running, or a negative value `-N` if the
+        kernel was killed by signal `N` (posix only)."""
+        if not self.has_kernel:
+            return 0
+        assert self.provisioner is not None
+        return await self.provisioner.poll()
+
+    exit_status = run_sync(_async_exit_status)
+
     async def _async_wait(self, pollinterval: float = 0.1) -> None:
         # Use busy loop at 100ms intervals, polling until the process is
         # not alive.  If we find the process is no longer alive, complete
diff --git a/tests/test_kernelmanager.py b/tests/test_kernelmanager.py
index f2d749eb..989d5198 100644
--- a/tests/test_kernelmanager.py
+++ b/tests/test_kernelmanager.py
@@ -160,6 +160,46 @@ async def test_async_signal_kernel_subprocesses(self, name, install, expected):
         assert km._shutdown_status in expected
 
 
+class TestKernelManagerExitStatus:
+    @pytest.mark.skipif(sys.platform == "win32", reason="Windows doesn't support signals")
+    @pytest.mark.parametrize('_signal', [signal.SIGHUP, signal.SIGTERM, signal.SIGKILL])
+    async def test_exit_status(self, _signal):
+        # install kernel
+        _install_kernel(name="test_exit_status")
+
+        # start kernel
+        km, kc = start_new_kernel(kernel_name="test_exit_status")
+
+        # stop restarter - not needed?
+        # km.stop_restarter()
+
+        # check that process is running
+        assert km.exit_status() is None
+
+        # get the provisioner
+        # send signal
+        provisioner = km.provisioner
+        assert provisioner is not None
+        assert provisioner.has_process
+        await provisioner.send_signal(_signal)
+
+        # wait for the process to exit
+        try:
+            await asyncio.wait_for(km._async_wait(), timeout=3.0)
+        except TimeoutError:
+            assert False, f'process never stopped for signal {signal}'
+
+        # check that the signal is correct
+        assert km.exit_status() == -_signal
+
+        # doing a proper shutdown now wipes the status, might be bad?
+        km.shutdown_kernel(now=True)
+        assert km.exit_status() == 0
+
+        # stop channels so cleanup doesn't complain
+        kc.stop_channels()
+
+
 class TestKernelManager:
     def test_lifecycle(self, km):
         km.start_kernel(stdout=PIPE, stderr=PIPE)