diff --git a/jupyter_client/tests/test_client.py b/jupyter_client/tests/test_client.py index a422462b3..87cfd23fd 100644 --- a/jupyter_client/tests/test_client.py +++ b/jupyter_client/tests/test_client.py @@ -1,12 +1,15 @@ """Tests for the KernelClient""" # Copyright (c) Jupyter Development Team. # Distributed under the terms of the Modified BSD License. +import asyncio import os +from textwrap import dedent from unittest import TestCase import pytest from IPython.utils.capture import capture_output +from ..manager import start_new_async_kernel from ..manager import start_new_kernel from .utils import test_env from jupyter_client.kernelspec import KernelSpecManager @@ -18,6 +21,40 @@ pjoin = os.path.join +@pytest.mark.asyncio +async def test_interrupt_coroutine(): + try: + KernelSpecManager().get_kernel_spec(NATIVE_KERNEL_NAME) + except NoSuchKernel: + pytest.skip() + + km, kc = await start_new_async_kernel(kernel_name=NATIVE_KERNEL_NAME) + + code = dedent( + """ + import asyncio + + async def main(): + print("sleeping") + await asyncio.sleep(0.5) + print("done") + + await main() + """ + ) + with capture_output() as io: + asyncio.create_task(kc.execute_interactive(code, timeout=TIMEOUT)) + # wait for coroutine to start, this should print "sleeping" + await asyncio.sleep(0.2) + # interrupt kernel before coroutine completes execution, "done" should not be printed + await km.interrupt_kernel() + + assert "sleeping" in io.stdout + assert "done" not in io.stdout + kc.stop_channels() + await km.shutdown_kernel() + + class TestKernelClient(TestCase): def setUp(self): self.env_patch = test_env()