Pass only user inputs to prepare_shell_job_inputs

aiida_workgraph/engine/utils.py

@@ -136,8 +136,26 @@ def prepare_for_shell_task(task: dict, inputs: dict) -> dict:
 
     # Retrieve the signature of `prepare_shell_job_inputs` to determine expected input parameters.
     signature = inspect.signature(prepare_shell_job_inputs)
-    kwargs = {key: inputs.pop(key, None) for key in signature.parameters.keys()}
-    inputs.update(prepare_shell_job_inputs(**kwargs))
+    aiida_shell_input_keys = signature.parameters.keys()
+
+    # Iterate over all WorkGraph `inputs`, and extract the ones which are expected by `prepare_shell_job_inputs`
+    inputs_aiida_shell_subset = {
+        key: inputs[key] for key in inputs.keys() if key in aiida_shell_input_keys
+    }
+
+    try:
+        aiida_shell_inputs = prepare_shell_job_inputs(**inputs_aiida_shell_subset)
+    except ValueError:
+        raise
+
+    # We need to remove the original input-keys, as they might be offending for the call to `launch_shell_job`
+    # E.g., `inputs` originally can contain `command`, which gets, however, transformed to #
+    # `code` by `prepare_shell_job_inputs`
+    for key in inputs_aiida_shell_subset.keys():
+        inputs.pop(key)
+
+    # Finally, we update the original `inputs` with the modified ones from the call to `prepare_shell_job_inputs`
+    inputs = {**inputs, **aiida_shell_inputs}
 
     inputs.setdefault("metadata", {})
     inputs["metadata"].update({"call_link_label": task["name"]})

tests/test_shell.py

@@ -4,6 +4,16 @@
 from aiida.orm import SinglefileData, load_computer
 
 
+def test_prepare_for_shell_task_nonexistent():
+    """Check that the `ValueError` raised by `aiida-shell` for a non-extistent executable is captured by WorkGraph."""
+    from aiida_workgraph.engine.utils import prepare_for_shell_task
+
+    task = {"name": "test"}
+    inputs = {"command": "abc42"}
+    with pytest.raises(ValueError, match="failed to determine the absolute path"):
+        prepare_for_shell_task(task=task, inputs=inputs)
+
+
 @pytest.mark.usefixtures("started_daemon_client")
 def test_shell_command(fixture_localhost):
     """Test the ShellJob with command as a string."""