Another try at tracking down ResourceWarning with tracemalloc.

pyproject.toml

@@ -173,6 +173,10 @@ filterwarnings= [
   # Ignore jupyter_client warnings
   "module:Jupyter is migrating its paths to use standard platformdirs:DeprecationWarning",
 
+  # we do not want to raise on resources warnings, or we will not have a chance to
+  # collect the messages and print the location of the leak
+  "always::ResourceWarning",
+
   # ignore unclosed sqlite in traits
   "ignore:unclosed database in <sqlite3.Connection:ResourceWarning",
 

tests/conftest.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import os
+import warnings
 from math import inf
 from typing import Any, Callable, no_type_check
 from unittest.mock import MagicMock
@@ -22,6 +23,11 @@
     # Windows
     resource = None  # type:ignore
 
+try:
+    import tracemalloc
+except ModuleNotFoundError:
+    tracemalloc = None
+
 
 @pytest.fixture()
 def anyio_backend():
@@ -207,3 +213,38 @@ async def ipkernel(anyio_backend):
         yield kernel
         kernel.destroy()
         ZMQInteractiveShell.clear_instance()
+
+
+@pytest.fixture()
+def tracemalloc_resource_warning(recwarn, N=10):
+    """fixture to enable tracemalloc for a single test, and report the
+    location of the leaked resource
+
+    We cannot only enable tracemalloc, as otherwise it is stopped just after the
+    test, the frame cache is cleared by tracemalloc.stop() and thus the warning
+    printing code get None when doing
+    `tracemalloc.get_object_traceback(r.source)`.
+
+    So we need to both filter the warnings to enable ResourceWarning, and loop
+    through it print the stack before we stop tracemalloc and continue.
+
+    """
+    if tracemalloc is None:
+        yield
+        return
+
+    tracemalloc.start(N)
+    with warnings.catch_warnings():
+        warnings.simplefilter("always", category=ResourceWarning)
+        yield None
+    try:
+        for r in recwarn:
+            if r.category is ResourceWarning and r.source is not None:
+                tb = tracemalloc.get_object_traceback(r.source)
+                if tb:
+                    info = f"Leaking resource:{r}\n |" + "\n |".join(tb.format())
+                    # technically an Error and not a failure as we fail in the fixture
+                    # and not the test
+                    pytest.fail(info)
+    finally:
+        tracemalloc.stop()

tests/test_connect.py

@@ -19,21 +19,6 @@
 
 from .utils import TemporaryWorkingDirectory
 
-
-@pytest.fixture(scope="module", autouse=True)
-def _enable_tracemalloc():
-    try:
-        import tracemalloc
-    except ModuleNotFoundError:
-        # pypy
-        tracemalloc = None
-    if tracemalloc is not None:
-        tracemalloc.start()
-    yield
-    if tracemalloc is not None:
-        tracemalloc.stop()
-
-
 sample_info: dict = {
     "ip": "1.2.3.4",
     "transport": "ipc",
@@ -133,7 +118,7 @@ def test_port_bind_failure_recovery(request):
             app.init_sockets()
 
 
-def test_port_bind_failure_gives_up_retries(request):
+def test_port_bind_failure_gives_up_retries(request, tracemalloc_resource_warning):
     cfg = Config()
     with TemporaryWorkingDirectory() as d:
         cfg.ProfileDir.location = d

tests/test_ipkernel_direct.py

@@ -51,7 +51,7 @@ async def test_direct_execute_request_aborting(ipkernel):
     assert reply["content"]["status"] == "aborted"
 
 
-async def test_complete_request(ipkernel):
+async def test_complete_request(ipkernel, tracemalloc_resource_warning):
     reply = await ipkernel.test_shell_message("complete_request", dict(code="hello", cursor_pos=0))
     assert reply["header"]["msg_type"] == "complete_reply"
     ipkernel.use_experimental_completions = False