🚸 Deduce shell in add_process if not specified

Author: kzscisoft

simvue/executor.py

@@ -35,6 +35,15 @@ class CompletionCallback(typing.Protocol):
     def __call__(self, *, status_code: int, std_out: str, std_err: str) -> None: ...
 
 
+def get_current_shell() -> str | None:
+    """Return the users current shell executable."""
+    try:
+        _process = psutil.Process(os.getppid())
+        return _process.exe()
+    except psutil.Error:
+        return None
+
+
 def _execute_process(
     proc_id: str,
     command: list[str],
@@ -231,9 +240,12 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
 
         if executable:
             command += [f"{executable}"]
-        else:
+        elif pos_args:
             command += [pos_args[0]]
             pos_args.pop(0)
+        elif executable := get_current_shell():
+            command += [f"{executable}"]
+
         if script:
             command += [f"{script}"]
 

simvue/run.py

@@ -38,7 +38,7 @@
 from .config.user import SimvueConfiguration
 
 from .factory.dispatch import Dispatcher
-from .executor import Executor
+from .executor import Executor, get_current_shell
 from .metrics import SystemResourceMeasurement
 from .models import FOLDER_REGEX, NAME_REGEX, MetricKeyString
 from .system import get_system
@@ -877,10 +877,12 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
         if executable:
             executable_str = f"{executable}"
             cmd_list += [executable_str]
-        else:
+        elif pos_args:
             cmd_list += [pos_args[0]]
             executable = pos_args[0]
             pos_args.pop(0)
+        else:
+            executable = get_current_shell()
 
         for kwarg, val in cmd_kwargs.items():
             _quoted_val: str = f'"{val}"'

tests/functional/test_executor.py

@@ -165,9 +165,11 @@ def completion_callback(*_, success: dict[str, bool]=success, **__):
             folder=f"/simvue_unit_testing/{folder_id}",
             tags=[platform.system(), "simvue_client_tests", request.node.name]
         )
+        print(os.environ.get("SHELL"))
         run.add_process(
             f"test_completion_callbacks_var_change_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
-            "exit 0",
+            c="exit 0",
+            executable="bash" if any(shell in os.environ.get("SHELL", "") for shell in ("zsh", "bash")) else None,
             completion_callback=completion_callback
         )