Block launch rocky server multiple times

ROCKY-20938

src/ansys/rocky/core/client.py

@@ -4,10 +4,12 @@
 import Pyro5.api
 import serpent
 
+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.
 

src/ansys/rocky/core/launcher.py

@@ -7,10 +7,19 @@
 
 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 PyRockyError
 
 
+def is_rocky_server_running() -> 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', ROCKY_SERVER_PORT)) == 0
+
+
 def launch_rocky(
     rocky_exe: Optional[Path] = None,
     headless: bool = True,
@@ -19,6 +28,9 @@ def launch_rocky(
     Launch Rocky executable with PyRocky server enabled, wait Rocky to start up and
     return a `RockyClient` instance.
     """
+    if is_rocky_server_running():
+        raise RockyLaunchError("Cannot start Rocky server. There is already a Rocky server running.")
+
     if rocky_exe is None:
         for awp_root in ["AWP_ROOT241", "AWP_ROOT232"]:
             if awp_root not in os.environ:

tests/test_pyrocky.py

@@ -1,6 +1,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()
@@ -59,3 +61,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="Cannot start Rocky server. There is already a Rocky server running."):
+            pyrocky.launch_rocky()