tests: added unit tests for connect_to_mapdl

Author: germa89

src/ansys/mapdl/core/launcher/remote.py

@@ -33,14 +33,50 @@
 )
 from ansys.mapdl.core.mapdl_grpc import MapdlGrpc
 
+_NON_VALID_ARGS = (
+    "add_env_vars",
+    "additional_switches",
+    "exec_file",
+    "jobname",
+    "launch_on_hpc",
+    "license_server_check",
+    "license_type",
+    "mapdl_output",
+    "mode",
+    "nproc",
+    "override",
+    "ram",
+    "remove_temp_dir_on_exit",
+    "replace_env_vars",
+    "run_location",
+    "running_on_hpc",
+    "start_instance",
+    "version",
+)
+
+
+def check_remote_args(args):
+    for each_arg in _NON_VALID_ARGS:
+        if each_arg in args:
+            raise ValueError(
+                f"'connect_to_mapdl' does not accept '{each_arg}' argument."
+            )
+        else:
+            if each_arg == "mode":
+                args[each_arg] = "grpc"
+            elif each_arg == "start_instance":
+                args[each_arg] = False
+            else:
+                args[each_arg] = None  # setting everything as None.
+
 
 def connect_to_mapdl(
+    port: Optional[int] = None,
+    ip: Optional[str] = None,
     *,
     loglevel: str = "ERROR",
     start_timeout: Optional[int] = None,
-    port: Optional[int] = None,
     cleanup_on_exit: bool = True,
-    ip: Optional[str] = None,
     clear_on_connect: bool = True,
     log_apdl: Optional[Union[bool, str]] = None,
     print_com: bool = False,
@@ -55,31 +91,16 @@ def connect_to_mapdl(
 
     check_kwargs(args)  # check if passing wrong arguments
 
-    if args.get("start_instance"):
-        raise ValueError(
-            "'connect_to_mapdl' only accept 'start_instance' equals 'False'. "
-            "If you intend to launch locally an instance use either "
-            "'launch_mapdl_grpc' or the infamous 'launch_mapdl(start_instance=True)'."
-        )
+    check_remote_args(args)
 
     pre_check_args(args)
 
     get_ip(args)
 
     args["port"] = get_port(args["port"], args["start_instance"])
 
-    # Check for a valid connection mode
-    # args["mode"] = check_mode(args["mode"], args["version"])
-    if args.get("mode", "grpc") != "grpc":
-        raise ValueError("Only a 'grpc' instance can be connected to remotely.")
-
     start_parm = generate_start_parameters(args)
 
-    # Early exit for debugging.
-    if args["_debug_no_launch"]:
-        # Early exit, just for testing
-        return args  # type: ignore
-
     ########################################
     # Connecting to a remote instance
     # -------------------------------
@@ -90,7 +111,7 @@ def connect_to_mapdl(
     start_parm["launched"] = False
 
     mapdl = MapdlGrpc(
-        cleanup_on_exit=False,
+        cleanup_on_exit=args["cleanup_on_exit"],
         loglevel=args["loglevel"],
         set_no_abort=args["set_no_abort"],
         use_vtk=args["use_vtk"],

src/ansys/mapdl/core/launcher/tools.py

@@ -1176,8 +1176,6 @@ def pack_arguments(locals_):
     args["_debug_no_launch"] = locals_.get(
         "_debug_no_launch", locals_["kwargs"].get("_debug_no_launch", None)
     )
-    args.setdefault("launch_on_hpc", False)
-    args.setdefault("ip", None)
 
     return args
 
@@ -1565,6 +1563,9 @@ def pre_check_args(args: dict[str, Any]):
             "the argument 'nproc' in 'launch_mapdl'."
         )
 
+    args.setdefault("launch_on_hpc", False)
+    args.setdefault("ip", None)
+
 
 def get_cpus(args: Dict[str, Any]):
     """Get number of CPUs

src/ansys/mapdl/core/mapdl_core.py

@@ -169,15 +169,16 @@
 _ALLOWED_START_PARM = [
     "additional_switches",
     "check_parameter_names",
+    "clear_on_connect",
     "env_vars",
-    "launched",
     "exec_file",
     "finish_job_on_exit",
     "hostname",
     "ip",
     "jobid",
     "jobname",
     "launch_on_hpc",
+    "launched",
     "mode",
     "nproc",
     "override",

tests/test_launcher/test_remote.py

@@ -19,3 +19,62 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
+from unittest.mock import patch
+
+import pytest
+
+from ansys.mapdl.core.launcher.remote import _NON_VALID_ARGS, connect_to_mapdl
+
+
+def test_connect_to_mapdl():
+    mapdl = connect_to_mapdl()
+
+    assert "PREP7" in mapdl.prep7()
+
+    assert not mapdl._start_instance
+    assert not mapdl._launched
+
+    mapdl.exit(force=False)
+
+
+@pytest.mark.parametrize("arg", _NON_VALID_ARGS)
+def test_connect_to_mapdl_exceptions(arg):
+    with pytest.raises(
+        ValueError, match=f"'connect_to_mapdl' does not accept '{arg}' argument."
+    ):
+        connect_to_mapdl(**{arg: True})
+
+
+_IP_TEST = "my_ip"
+
+
+@pytest.mark.parametrize(
+    "arg,value",
+    (
+        ("ip", _IP_TEST),
+        ("port", 50053),
+        ("loglevel", "DEBUG"),
+        ("loglevel", "ERROR"),
+        ("start_timeout", 12),
+        ("start_timeout", 15),
+        ("cleanup_on_exit", True),
+        ("cleanup_on_exit", False),
+        ("clear_on_connect", True),
+        ("clear_on_connect", False),
+        ("log_apdl", True),
+        ("log_apdl", False),
+        ("log_apdl", "log.out"),
+        ("print_com", False),
+    ),
+)
+@patch("socket.gethostbyname", lambda *args, **kwargs: _IP_TEST)
+@patch("socket.inet_aton", lambda *args, **kwargs: _IP_TEST)
+def test_connect_to_mapdl_kwargs(arg, value):
+    with patch("ansys.mapdl.core.launcher.remote.MapdlGrpc") as mock_mg:
+        args = {arg: value}
+        mapdl = connect_to_mapdl(**args)
+
+        mock_mg.assert_called_once()
+        kw = mock_mg.call_args_list[0].kwargs
+        assert "ip" in kw and kw["ip"] == _IP_TEST
+        assert arg in kw and kw[arg] == value