Allow to set Rocky server port on launch

src/ansys/rocky/core/client.py

@@ -1,16 +1,16 @@
 import pickle
-from typing import Generator
+from typing import Final, Generator
 
 import Pyro5.api
 import serpent
 
-ROCKY_SERVER_PORT = 50615
+DEFAULT_SERVER_PORT: Final[int] = 50615
 
 _ROCKY_API = None
 
 
 def connect_to_rocky(
-    host: str = "localhost", port: int = ROCKY_SERVER_PORT
+    host: str = "localhost", port: int = DEFAULT_SERVER_PORT
 ) -> "RockyClient":
     """
     Connect to a Rocky Application instance.

src/ansys/rocky/core/launcher.py

@@ -7,15 +7,17 @@
 
 from Pyro5.errors import CommunicationError
 
-from ansys.rocky.core.client import ROCKY_SERVER_PORT, RockyClient, connect_to_rocky
+from ansys.rocky.core.client import DEFAULT_SERVER_PORT, RockyClient, connect_to_rocky
 from ansys.rocky.core.exceptions import PyRockyError
 
 _WAIT_ROCKY_START = 60
 
 
 def launch_rocky(
     rocky_exe: Optional[Union[Path, str]] = None,
+    *,
     headless: bool = True,
+    server_port: int = DEFAULT_SERVER_PORT,
 ) -> RockyClient:
     """
     Launch Rocky executable with PyRocky server enabled, wait Rocky to start up and
@@ -24,8 +26,8 @@ def launch_rocky(
     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 _is_port_busy(server_port):
+        raise RockyLaunchError(f"Port {server_port} already in use")
 
     if rocky_exe is None:
         for awp_root in ["AWP_ROOT241", "AWP_ROOT232"]:
@@ -41,7 +43,7 @@ def launch_rocky(
         if not rocky_exe.is_file():
             raise FileNotFoundError(f"Rocky executable not found at {rocky_exe}")
 
-    cmd = [rocky_exe, "--pyrocky"]
+    cmd = [rocky_exe, "--pyrocky", "--pyrocky-port", str(server_port)]
     if headless:
         cmd.append("--headless")
     with contextlib.suppress(subprocess.TimeoutExpired):
@@ -52,10 +54,11 @@ def launch_rocky(
     if rocky_process.returncode is not None:
         raise RockyLaunchError(f"Error launching Rocky:\n  {' '.join(cmd)}")
 
-    client = connect_to_rocky()
+    client = connect_to_rocky(port=server_port)
 
     # TODO: A more elegant way to find out that Rocky Pyro server started.
-    for _ in range(_WAIT_ROCKY_START):
+    now = time.time()
+    while (time.time() - now) < _WAIT_ROCKY_START:
         try:
             client.api.GetProject()
         except CommunicationError:

tests/test_pyrocky.py

@@ -1,7 +1,7 @@
 import pytest
 
 import ansys.rocky.core as pyrocky
-from ansys.rocky.core.client import ROCKY_SERVER_PORT
+from ansys.rocky.core.client import DEFAULT_SERVER_PORT
 from ansys.rocky.core.launcher import RockyLaunchError
 
 
@@ -71,8 +71,8 @@ def test_pyrocky_launch_multiple_servers():
 
     # 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.bind(("localhost", DEFAULT_SERVER_PORT))
         s.listen(10)
 
-        with pytest.raises(RockyLaunchError, match="Port \d+ already in use"):
+        with pytest.raises(RockyLaunchError, match=r"Port \d+ already in use"):
             pyrocky.launch_rocky()