Block launch rocky server multiple times

ROCKY-20938

src/ansys/rocky/core/client.py

@@ -28,10 +28,12 @@
 
 from ansys.rocky.core.exceptions import RockyApiError
 
+ROCKY_SERVER_PORT = 50615
+
 _ROCKY_API = None
 
 
-def connect_to_rocky(host: str = "localhost", port: int = 50615) -> "RockyClient":
+def connect_to_rocky(host: str = "localhost", port: int = ROCKY_SERVER_PORT) -> "RockyClient":
     """Connect to a Rocky Application instance.
 
     Parameters

src/ansys/rocky/core/launcher.py

@@ -25,16 +25,18 @@
 from pathlib import Path
 import subprocess
 import time
-from typing import Optional
+from typing import Optional, Union
 
 from Pyro5.errors import CommunicationError
 
-from ansys.rocky.core.client import RockyClient, connect_to_rocky
+from ansys.rocky.core.client import ROCKY_SERVER_PORT, RockyClient, connect_to_rocky
 from ansys.rocky.core.exceptions import RockyLaunchError
 
+_WAIT_ROCKY_START = 60
+
 
 def launch_rocky(
-    rocky_exe: Optional[Path] = None,
+    rocky_exe: Optional[Union[Path, str]] = None,
     headless: bool = True,
 ) -> RockyClient:
     """
@@ -54,6 +56,12 @@ def launch_rocky(
     RockyClient
         A `RockyClient` instance connected to the launched Rocky application.
     """
+    if isinstance(rocky_exe, str):
+        rocky_exe = Path(rocky_exe)
+
+    if _is_port_busy(ROCKY_SERVER_PORT):
+        raise RockyLaunchError(f"Port {ROCKY_SERVER_PORT} already in use")
+
     if rocky_exe is None:
         for awp_root in ["AWP_ROOT241", "AWP_ROOT232"]:
             if awp_root not in os.environ:
@@ -96,4 +104,11 @@ def launch_rocky(
     return client
 
 
-_WAIT_ROCKY_START = 60
+def _is_port_busy(port: int) -> bool:
+    """
+    Check if there is already a Rocky server running.
+    """
+    import socket
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        return s.connect_ex(("localhost", port)) == 0

tests/test_pyrocky.py

@@ -23,6 +23,8 @@
 import pytest
 
 import ansys.rocky.core as pyrocky
+from ansys.rocky.core.client import ROCKY_SERVER_PORT
+from ansys.rocky.core.launcher import RockyLaunchError
 
 
 @pytest.fixture()
@@ -81,3 +83,18 @@ def test_sequences_interface(rocky_session):
     # Test __del__
     del inlets_outlets[0]
     assert {e.GetName() for e in inlets_outlets} == {"Inlet2"}
+
+
+def test_pyrocky_launch_multiple_servers():
+    """
+    Test that start multiple rocky servers is not allowed.
+    """
+    import socket
+
+    # Emulating Rocky server already running by binding socket to the server address.
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("localhost", ROCKY_SERVER_PORT))
+        s.listen(10)
+
+        with pytest.raises(RockyLaunchError, match="Port \d+ already in use"):
+            pyrocky.launch_rocky()