Handle warnings in tests

.github/workflows/main.yml

@@ -73,7 +73,7 @@ jobs:
         run: mypy jupyter_client --exclude '\/tests|kernelspecapp|ioloop|runapp' --install-types --non-interactive
 
       - name: Run the tests with coverage
-        run: pytest --cov jupyter_client -v jupyter_client
+        run: pytest --cov jupyter_client -vv jupyter_client || || pytest --cov jupyter_client -vv jupyter_client --lf
 
       - name: Build the docs
         run: |
@@ -96,7 +96,7 @@ jobs:
       - name: Install miniumum versions
         uses: jupyterlab/maintainer-tools/.github/actions/install-minimums@v1
       - name: Run the unit tests
-        run: pytest -vv jupyter_client || pytest -vv jupyter_client --lf
+        run: pytest -vv jupyter_client -W default || pytest -vv jupyter_client --lf -W default
 
   test_prereleases:
     name: Test Prereleases

jupyter_client/client.py

@@ -11,6 +11,7 @@
 
 import zmq.asyncio
 from traitlets import Any  # type: ignore
+from traitlets import Bool
 from traitlets import Instance
 from traitlets import Type
 
@@ -92,7 +93,10 @@ class KernelClient(ConnectionFileMixin):
     # The PyZMQ Context to use for communication with the kernel.
     context = Instance(zmq.asyncio.Context)
 
+    _created_context: Bool = Bool(False)
+
     def _context_default(self) -> zmq.asyncio.Context:
+        self._created_context = True
         return zmq.asyncio.Context()
 
     # The classes to use for the various channels
@@ -282,6 +286,9 @@ def start_channels(
         :meth:`start_kernel`. If the channels have been stopped and you
         call this, :class:`RuntimeError` will be raised.
         """
+        # Re-create the context if needed.
+        if self._created_context:
+            self.context = self._context_default()
         if iopub:
             self.iopub_channel.start()
         if shell:
@@ -311,6 +318,8 @@ def stop_channels(self) -> None:
             self.hb_channel.stop()
         if self.control_channel.is_alive():
             self.control_channel.stop()
+        if self._created_context:
+            self.context.destroy()
 
     @property
     def channels_running(self) -> bool:

jupyter_client/kernelspec.py

@@ -31,6 +31,8 @@
 
 class KernelSpec(HasTraits):
     argv = List()
+    name = Unicode()
+    mimetype = Unicode()
     display_name = Unicode()
     language = Unicode()
     env = Dict()

jupyter_client/manager.py

@@ -52,33 +52,38 @@ class _ShutdownStatus(Enum):
     SigkillRequest = "SigkillRequest"
 
 
-def in_pending_state(method):
+def in_pending_state(attr='_ready'):
     """Sets the kernel to a pending state by
     creating a fresh Future for the KernelManager's `ready`
     attribute. Once the method is finished, set the Future's results.
     """
 
-    @functools.wraps(method)
-    async def wrapper(self, *args, **kwargs):
-        # Create a future for the decorated method
-        try:
-            self._ready = Future()
-        except RuntimeError:
-            # No event loop running, use concurrent future
-            self._ready = CFuture()
-        try:
-            # call wrapped method, await, and set the result or exception.
-            out = await method(self, *args, **kwargs)
-            # Add a small sleep to ensure tests can capture the state before done
-            await asyncio.sleep(0.01)
-            self._ready.set_result(None)
-            return out
-        except Exception as e:
-            self._ready.set_exception(e)
-            self.log.exception(self._ready.exception())
-            raise e
+    def inner(method):
+        @functools.wraps(method)
+        async def wrapper(self, *args, **kwargs):
+            # Create a future for the decorated method
+            try:
+                asyncio.get_running_loop()
+                fut = Future()
+            except RuntimeError:
+                # No event loop running, use concurrent future
+                fut = CFuture()
+            setattr(self, attr, fut)
+            try:
+                # call wrapped method, await, and set the result or exception.
+                out = await method(self, *args, **kwargs)
+                # Add a small sleep to ensure tests can capture the state before done
+                await asyncio.sleep(0.01)
+                fut.set_result(None)
+                return out
+            except Exception as e:
+                fut.set_exception(e)
+                self.log.exception(fut.exception())
+                raise e
+
+        return wrapper
 
-    return wrapper
+    return inner
 
 
 class KernelManager(ConnectionFileMixin):
@@ -90,8 +95,10 @@ class KernelManager(ConnectionFileMixin):
     def __init__(self, *args, **kwargs):
         super().__init__(**kwargs)
         self._shutdown_status = _ShutdownStatus.Unset
+        self._shutdown_ready = None
         # Create a place holder future.
         try:
+            asyncio.get_running_loop()
             self._ready = Future()
         except RuntimeError:
             # No event loop running, use concurrent future
@@ -182,6 +189,11 @@ def ready(self) -> Future:
         """A future that resolves when the kernel process has started for the first time"""
         return self._ready
 
+    @property
+    def shutdown_ready(self) -> Future:
+        """A future that resolves when a shutdown has completed"""
+        return self._shutdown_ready
+
     @property
     def ipykernel(self) -> bool:
         return self.kernel_name in {"python", "python2", "python3"}
@@ -360,7 +372,7 @@ async def _async_post_start_kernel(self, **kw) -> None:
 
     post_start_kernel = run_sync(_async_post_start_kernel)
 
-    @in_pending_state
+    @in_pending_state()
     async def _async_start_kernel(self, **kw):
         """Starts a kernel on this host in a separate process.
 
@@ -387,8 +399,9 @@ async def _async_request_shutdown(self, restart: bool = False) -> None:
         content = dict(restart=restart)
         msg = self.session.msg("shutdown_request", content=content)
         # ensure control socket is connected
-        self._connect_control_socket()
-        self.session.send(self._control_socket, msg)
+        if self._control_socket and not self._control_socket.closed:
+            self._connect_control_socket()
+            self.session.send(self._control_socket, msg)
         assert self.provisioner is not None
         await self.provisioner.shutdown_requested(restart=restart)
         self._shutdown_status = _ShutdownStatus.ShutdownRequest
@@ -453,7 +466,7 @@ async def _async_cleanup_resources(self, restart: bool = False) -> None:
 
     cleanup_resources = run_sync(_async_cleanup_resources)
 
-    @in_pending_state
+    @in_pending_state('_shutdown_ready')
     async def _async_shutdown_kernel(self, now: bool = False, restart: bool = False):
         """Attempts to stop the kernel process cleanly.
 
@@ -476,7 +489,10 @@ async def _async_shutdown_kernel(self, now: bool = False, restart: bool = False)
         # Stop monitoring for restarting while we shutdown.
         self.stop_restarter()
 
-        await ensure_async(self.interrupt_kernel())
+        # If the kernel has already started, interrupt it to give it a
+        # chance to clean up.
+        if self.has_kernel:
+            await ensure_async(self.interrupt_kernel())
 
         if now:
             await ensure_async(self._kill_kernel())

jupyter_client/provisioning/local_provisioner.py

@@ -60,6 +60,11 @@ async def wait(self) -> Optional[int]:
 
             # Process is no longer alive, wait and clear
             ret = self.process.wait()
+            # Make sure all the fds get closed.
+            for attr in ['stdout', 'stderr', 'stdin']:
+                fid = getattr(self.process, attr)
+                if fid:
+                    fid.close()
             self.process = None  # allow has_process to now return False
         return ret
 

jupyter_client/ssh/tunnel.py

@@ -36,8 +36,6 @@ class SSHException(Exception):  # type: ignore
 except ImportError:
     pexpect = None
 
-from zmq.utils.strtypes import b
-
 
 def select_random_ports(n):
     """Select and return n random ports that are available."""
@@ -56,7 +54,7 @@ def select_random_ports(n):
 # -----------------------------------------------------------------------------
 # Check for passwordless login
 # -----------------------------------------------------------------------------
-_password_pat = re.compile(b(r"pass(word|phrase):"), re.IGNORECASE)
+_password_pat = re.compile(r"pass(word|phrase):".encode("utf8"), re.IGNORECASE)
 
 
 def try_passwordless_ssh(server, keyfile, paramiko=None):

jupyter_client/tests/conftest.py

@@ -7,9 +7,31 @@
 
 from .utils import test_env
 
+try:
+    import resource
+except ImportError:
+    # Windows
+    resource = None
+
 pjoin = os.path.join
 
 
+# Handle resource limit
+if resource is not None:
+    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+
+    DEFAULT_SOFT = 4096
+    if hard >= DEFAULT_SOFT:
+        soft = DEFAULT_SOFT
+
+    old_soft, old_hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    hard = old_hard
+    if old_soft < soft:
+        if hard < soft:
+            hard = soft
+        resource.setrlimit(resource.RLIMIT_NOFILE, (soft, hard))
+
+
 if os.name == "nt" and sys.version_info >= (3, 7):
     asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
 
@@ -22,6 +44,7 @@ def event_loop():
     if os.name == "nt" and sys.version_info >= (3, 7):
         asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
     loop = asyncio.SelectorEventLoop()
+
     try:
         yield loop
     finally:

jupyter_client/tests/test_client.py

@@ -22,14 +22,16 @@ class TestKernelClient(TestCase):
     def setUp(self):
         self.env_patch = test_env()
         self.env_patch.start()
-        self.addCleanup(self.env_patch.stop)
         try:
             KernelSpecManager().get_kernel_spec(NATIVE_KERNEL_NAME)
         except NoSuchKernel:
             pytest.skip()
         self.km, self.kc = start_new_kernel(kernel_name=NATIVE_KERNEL_NAME)
-        self.addCleanup(self.kc.stop_channels)
-        self.addCleanup(self.km.shutdown_kernel)
+
+    def tearDown(self):
+        self.env_patch.stop()
+        self.km.shutdown_kernel()
+        self.kc.stop_channels()
 
     def test_execute_interactive(self):
         kc = self.kc

jupyter_client/tests/test_kernelmanager.py

@@ -441,6 +441,8 @@ def execute(cmd):
         km.shutdown_kernel()
         assert km.context.closed
 
+        kc.stop_channels()
+
 
 @pytest.mark.asyncio
 class TestAsyncKernelManager:
@@ -525,6 +527,8 @@ async def execute(cmd):
         # verify that subprocesses were interrupted
         assert reply["user_expressions"]["poll"] == [-signal.SIGINT] * N
 
+        await km.shutdown_kernel()
+
     @pytest.mark.timeout(10)
     async def test_start_new_async_kernel(self, install_kernel, start_async_kernel):
         km, kc = start_async_kernel

jupyter_client/tests/test_multikernelmanager.py

@@ -44,6 +44,10 @@ def setUp(self):
         self.env_patch.start()
         super().setUp()
 
+    def tearDown(self) -> None:
+        self.env_patch.stop()
+        return super().tearDown()
+
     # static so picklable for multiprocessing on Windows
     @staticmethod
     def _get_tcp_km():
@@ -243,6 +247,10 @@ def setUp(self):
         self.env_patch.start()
         super().setUp()
 
+    def tearDown(self):
+        self.env_patch.stop()
+        super().setUp()
+
     # static so picklable for multiprocessing on Windows
     @staticmethod
     def _get_tcp_km():
@@ -382,9 +390,8 @@ async def test_use_pending_kernels(self):
         k = km.get_kernel(kid)
         assert isinstance(k, AsyncKernelManager)
         await ensure_future(km.shutdown_kernel(kid, now=True))
-        # Wait for the kernel to shutdown
-        await kernel.ready
-        assert kid not in km, f"{kid} not in {km}"
+        await kernel.shutdown_ready
+        assert kid not in km, f"{kid} aws in {km}"
 
     @gen_test
     async def test_use_pending_kernels_early_restart(self):
@@ -396,9 +403,8 @@ async def test_use_pending_kernels_early_restart(self):
             await km.restart_kernel(kid, now=True)
         await kernel.ready
         await ensure_future(km.shutdown_kernel(kid, now=True))
-        # Wait for the kernel to shutdown
-        await kernel.ready
-        assert kid not in km, f"{kid} not in {km}"
+        await kernel.shutdown_ready
+        assert kid not in km, f"{kid} was in {km}"
 
     @gen_test
     async def test_use_pending_kernels_early_shutdown(self):
@@ -412,9 +418,8 @@ async def test_use_pending_kernels_early_shutdown(self):
         await kernel.ready
         # Shutdown once the kernel is ready
         await ensure_future(km.shutdown_kernel(kid, now=True))
-        # Wait for the kernel to shutdown
-        await kernel.ready
-        assert kid not in km, f"{kid} not in {km}"
+        await kernel.shutdown_ready
+        assert kid not in km, f"{kid} was in {km}"
 
     @gen_test
     async def test_use_pending_kernels_early_interrupt(self):
@@ -427,9 +432,8 @@ async def test_use_pending_kernels_early_interrupt(self):
         # Now wait for the kernel to be ready.
         await kernel.ready
         await ensure_future(km.shutdown_kernel(kid, now=True))
-        # Wait for the kernel to shutdown
-        await kernel.ready
-        assert kid not in km, f"{kid} not in {km}"
+        await kernel.shutdown_ready
+        assert kid not in km, f"{kid} was in {km}"
 
     @gen_test
     async def test_tcp_cinfo(self):
@@ -465,8 +469,9 @@ async def test_start_sequence_ipc_kernels(self):
 
     def tcp_lifecycle_with_loop(self):
         # Ensure each thread has an event loop
-        asyncio.set_event_loop(asyncio.new_event_loop())
-        asyncio.get_event_loop().run_until_complete(self.raw_tcp_lifecycle())
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.raw_tcp_lifecycle())
 
     # static so picklable for multiprocessing on Windows
     @classmethod
@@ -479,11 +484,8 @@ async def raw_tcp_lifecycle(cls, test_kid=None):
     # static so picklable for multiprocessing on Windows
     @classmethod
     def raw_tcp_lifecycle_sync(cls, test_kid=None):
-        loop = asyncio.get_event_loop()
-        if loop.is_running():
-            # Forked MP, make new loop
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
         loop.run_until_complete(cls.raw_tcp_lifecycle(test_kid=test_kid))
 
     @gen_test