fix: check existence of connection_file before writing (ipython#1127)

Co-authored-by: Steven Silvester <[email protected]>

Author: fecet

ipykernel/eventloops.py

@@ -429,7 +429,7 @@ async def close_loop():
         close_loop()
 
     elif not loop.is_closed():
-        loop.run_until_complete(close_loop)  # type:ignore[call-overload]
+        loop.run_until_complete(close_loop)  # type:ignore
         loop.close()
 
 

ipykernel/inprocess/tests/test_kernelmanager.py

@@ -3,6 +3,8 @@
 
 import unittest
 
+from flaky import flaky
+
 from ipykernel.inprocess.manager import InProcessKernelManager
 
 # -----------------------------------------------------------------------------
@@ -18,6 +20,7 @@ def tearDown(self):
         if self.km.has_kernel:
             self.km.shutdown_kernel()
 
+    @flaky
     def test_interface(self):
         """Does the in-process kernel manager implement the basic KM interface?"""
         km = self.km

ipykernel/kernelapp.py

@@ -260,6 +260,9 @@ def _bind_socket(self, s, port):
     def write_connection_file(self):
         """write connection info to JSON file"""
         cf = self.abs_connection_file
+        if os.path.exists(cf):
+            self.log.debug("Connection file %s already exists", cf)
+            return
         self.log.debug("Writing connection file: %s", cf)
         write_connection_file(
             cf,

ipykernel/tests/test_debugger.py

@@ -121,7 +121,7 @@ def test_set_breakpoints(kernel_with_debug):
     assert reply["body"]["breakpoints"][0]["source"]["path"] == source
 
     r = wait_for_debug_request(kernel_with_debug, "debugInfo")
-    assert source in map(lambda b: b["source"], r["body"]["breakpoints"])  # type:ignore # noqa
+    assert source in map(lambda b: b["source"], r["body"]["breakpoints"])  # noqa
 
     r = wait_for_debug_request(kernel_with_debug, "configurationDone")
     assert r["success"]
@@ -208,7 +208,7 @@ def test_rich_inspect_not_at_breakpoint(kernel_with_debug):
     get_reply(kernel_with_debug, msg_id)
 
     r = wait_for_debug_request(kernel_with_debug, "inspectVariables")
-    assert var_name in list(map(lambda v: v["name"], r["body"]["variables"]))  # type:ignore  # noqa
+    assert var_name in list(map(lambda v: v["name"], r["body"]["variables"]))  # noqa
 
     reply = wait_for_debug_request(
         kernel_with_debug,