[fix](cli): detect bound ports before launch (#2071)

* [fix](cli): detect bound ports before launch

* [fix](cli): align port reuse check by platform

kt-kernel/python/cli/utils/port_checker.py

@@ -3,6 +3,7 @@ Port availability checking utilities.
 """
 
 import socket
+import sys
 from typing import Tuple
 
 
@@ -17,22 +18,14 @@ def is_port_available(host: str, port: int) -> bool:
         True if port is available, False if occupied
     """
     try:
-        # Try to bind to the port
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        sock.settimeout(1)
+        bind_host = "" if host == "0.0.0.0" else host
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            if sys.platform != "win32":
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            sock.bind((bind_host, port))
+        return True
 
-        # Use SO_REUSEADDR to allow binding to recently closed ports
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-
-        # Try to bind
-        result = sock.connect_ex((host if host != "0.0.0.0" else "127.0.0.1", port))
-        sock.close()
-
-        # If connect_ex returns 0, port is occupied
-        # If it returns error (non-zero), port is available
-        return result != 0
-
-    except Exception:
+    except OSError:
         # If any error occurs, assume port is not available
         return False
 

kt-kernel/test/per_commit/test_port_checker.py

@@ -0,0 +1,45 @@
+import importlib.util
+import socket
+from pathlib import Path
+import unittest
+from unittest.mock import MagicMock, patch
+
+from ci.ci_register import register_cpu_ci
+
+
+register_cpu_ci(est_time=0.1, suite="default")
+
+
+PORT_CHECKER_PATH = Path(__file__).resolve().parents[2] / "python" / "cli" / "utils" / "port_checker.py"
+SPEC = importlib.util.spec_from_file_location("port_checker", PORT_CHECKER_PATH)
+assert SPEC is not None and SPEC.loader is not None
+port_checker = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(port_checker)
+
+
+class TestPortChecker(unittest.TestCase):
+    def test_bound_port_is_not_available_before_listen(self):
+        holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            holder.bind(("127.0.0.1", 0))
+            port = holder.getsockname()[1]
+
+            self.assertFalse(port_checker.is_port_available("127.0.0.1", port))
+            self.assertEqual(port_checker.find_available_port("127.0.0.1", port, max_attempts=1), (False, port))
+        finally:
+            holder.close()
+
+    def test_non_windows_bind_check_uses_reuseaddr(self):
+        sock = MagicMock()
+        sock.__enter__.return_value = sock
+
+        with patch.object(port_checker.sys, "platform", "linux"):
+            with patch.object(port_checker.socket, "socket", return_value=sock):
+                self.assertTrue(port_checker.is_port_available("127.0.0.1", 12345))
+
+        sock.setsockopt.assert_called_once_with(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.bind.assert_called_once_with(("127.0.0.1", 12345))
+
+
+if __name__ == "__main__":
+    unittest.main()