Add unit test for a connection file with ports set to 0

Before ipykernel 6.23.3, i.e., before #1127, a kernel manager could specify a channel port of 0, and ipykernel would pick a random port and rewrite the connection file with the actual port used. This provided a nice way to address the natural race condition between a kernel manager picking a port and ipykernel actually connecting to it and using it.

This unit test tests that this port 0 connection file behavior works, and also tests that existing information in the connection file is not overwritten.

ipykernel/tests/test_kernelapp.py

@@ -1,13 +1,17 @@
+import json
 import os
 import threading
 import time
 from unittest.mock import patch
 
 import pytest
+from jupyter_core.paths import secure_write
+from traitlets.config.loader import Config
 
 from ipykernel.kernelapp import IPKernelApp
 
 from .conftest import MockKernel
+from .utils import TemporaryWorkingDirectory
 
 try:
     import trio
@@ -47,6 +51,71 @@ def trigger_stop():
     app.close()
 
 
+def test_merge_connection_file():
+    cfg = Config()
+    with TemporaryWorkingDirectory() as d:
+        cfg.ProfileDir.location = d
+        cf = os.path.join(d, "kernel.json")
+        initial_connection_info = {
+            "ip": "*",
+            "transport": "tcp",
+            "shell_port": 0,
+            "hb_port": 0,
+            "iopub_port": 0,
+            "stdin_port": 0,
+            "control_port": 53555,
+            "key": "abc123",
+            "signature_scheme": "hmac-sha256",
+            "kernel_name": "My Kernel",
+        }
+        # We cannot use connect.write_connection_file since
+        # it replaces port number 0 with a random port
+        # and we want IPKernelApp to do that replacement.
+        with secure_write(cf) as f:
+            json.dump(initial_connection_info, f)
+        assert os.path.exists(cf)
+
+        app = IPKernelApp(config=cfg, connection_file=cf)
+
+        # Calling app.initialize() does not work in the test, so we call the relevant functions that initialize() calls
+        super(IPKernelApp, app).initialize(argv=None)
+        app.init_connection_file()
+        app.init_sockets()
+        app.init_heartbeat()
+        app.write_connection_file()
+
+        # Initialize should have merged the actual connection info
+        # with the connection info in the file
+        assert cf == app.abs_connection_file
+        assert os.path.exists(cf)
+
+        with open(cf) as f:
+            new_connection_info = json.load(f)
+
+        # ports originally set as 0 have been replaced
+        for port in ("shell", "hb", "iopub", "stdin"):
+            key = f"{port}_port"
+            # We initially had the port as 0
+            assert initial_connection_info[key] == 0
+            # the port is not 0 now
+            assert new_connection_info[key] > 0
+            # the port matches the port the kernel actually used
+            assert new_connection_info[key] == getattr(app, key), f"{key}"
+            del new_connection_info[key]
+            del initial_connection_info[key]
+
+        # The wildcard ip address was also replaced
+        assert(new_connection_info["ip"] != "*")
+        del new_connection_info["ip"]
+        del initial_connection_info["ip"]
+
+        # everything else in the connection file is the same
+        assert initial_connection_info == new_connection_info
+
+        app.close()
+        os.remove(cf)
+
+
 @pytest.mark.skipif(trio is None, reason="requires trio")
 def test_trio_loop():
     app = IPKernelApp(trio_loop=True)