Merge pull request #585 from tclose/explicit-out-files

Provide paths to ShellTask output files

Author: djarecka

pydra/engine/specs.py

@@ -450,6 +450,7 @@ def collect_additional_outputs(self, inputs, output_dir, outputs):
                 File,
                 MultiOutputFile,
                 Directory,
+                Path,
                 int,
                 float,
                 bool,
@@ -458,7 +459,12 @@ def collect_additional_outputs(self, inputs, output_dir, outputs):
             ]:
                 raise Exception("not implemented (collect_additional_output)")
             # assuming that field should have either default or metadata, but not both
-            if (
+            input_value = getattr(inputs, fld.name, attr.NOTHING)
+            if input_value is not attr.NOTHING:
+                if fld.type in (File, MultiOutputFile, Directory, Path):
+                    input_value = Path(input_value).absolute()
+                additional_out[fld.name] = input_value
+            elif (
                 fld.default is None or fld.default == attr.NOTHING
             ) and not fld.metadata:  # TODO: is it right?
                 raise AttributeError("File has to have default value or metadata")

pydra/engine/tests/test_shelltask.py

@@ -1414,6 +1414,58 @@ def test_shell_cmd_inputspec_9c(tmpdir, plugin, results_function):
     assert res.output.file_copy.parent == shelly.output_dir
 
 
+@pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
+def test_shell_cmd_inputspec_9d(tmpdir, plugin, results_function):
+    """
+    providing output name explicitly by manually setting value in input_spec
+    (instead of using default provided byoutput_file_template in metadata)
+    """
+    cmd = "cp"
+    file = tmpdir.mkdir("data_inp").join("file.txt")
+    file.write("content\n")
+
+    my_input_spec = SpecInfo(
+        name="Input",
+        fields=[
+            (
+                "file_orig",
+                attr.ib(
+                    type=File,
+                    metadata={"position": 2, "help_string": "new file", "argstr": ""},
+                ),
+            ),
+            (
+                "file_copy",
+                attr.ib(
+                    type=str,
+                    metadata={
+                        "output_file_template": "{file_orig}_copy",
+                        "help_string": "output file",
+                        "argstr": "",
+                    },
+                ),
+            ),
+        ],
+        bases=(ShellSpec,),
+    )
+
+    shelly = ShellCommandTask(
+        name="shelly",
+        executable=cmd,
+        input_spec=my_input_spec,
+        file_orig=file,
+        file_copy="my_file_copy.txt",
+        cache_dir=tmpdir,
+    )
+
+    res = results_function(shelly, plugin)
+    assert res.output.stdout == ""
+    assert res.output.file_copy.exists()
+    assert res.output.file_copy.name == "my_file_copy.txt"
+    # checking if it's created in a good place
+    assert shelly.output_dir == res.output.file_copy.parent
+
+
 @pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
 def test_shell_cmd_inputspec_10(plugin, results_function, tmpdir):
     """using input_spec, providing list of files as an input"""