diff --git a/pydra/engine/tests/test_shelltask.py b/pydra/engine/tests/test_shelltask.py index c967e3f765..ba8baf11c4 100644 --- a/pydra/engine/tests/test_shelltask.py +++ b/pydra/engine/tests/test_shelltask.py @@ -14,6 +14,7 @@ SpecInfo, File, Directory, + MultiInputFile, MultiOutputFile, MultiInputObj, ) @@ -1503,6 +1504,57 @@ def test_shell_cmd_inputspec_10_err(tmpdir): res = shelly() +def test_shell_cmd_inputsspec_11(): + input_fields = [ + ( + "inputFiles", + attr.ib( + type=MultiInputFile, + metadata={ + "argstr": "...", + "help_string": "The list of input image files to be segmented.", + }, + ), + ) + ] + + output_fields = [ + ( + "outputFiles", + attr.ib( + type=MultiOutputFile, + metadata={ + "help_string": "Corrected Output Images: should specify the same number of images as inputVolume, if only one element is given, then it is used as a file pattern where %s is replaced by the imageVolumeType, and %d by the index list location.", + "output_file_template": "{inputFiles}", + }, + ), + ) + ] + + input_spec = SpecInfo(name="Input", fields=input_fields, bases=(ShellSpec,)) + output_spec = SpecInfo(name="Output", fields=output_fields, bases=(ShellOutSpec,)) + + task = ShellCommandTask( + name="echoMultiple", + executable="touch", + input_spec=input_spec, + output_spec=output_spec, + ) + wf = Workflow(name="wf", input_spec=["inputFiles"], inputFiles=["test1", "test2"]) + + task.inputs.inputFiles = wf.lzin.inputFiles + + wf.add(task) + wf.set_output([("out", wf.echoMultiple.lzout.outputFiles)]) + + with Submitter(plugin="cf") as sub: + sub(wf) + result = wf.result() + + for out_file in result.output.out: + assert out_file.name == "test1" or out_file.name == "test2" + + @pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter]) def test_shell_cmd_inputspec_copyfile_1(plugin, results_function, tmpdir): """shelltask changes a file in place, @@ -4509,3 +4561,412 @@ def test_shell_cmd_non_existing_outputs_multi_2(tmpdir): # checking if the outputs are Nothing assert res.output.out_list[0] == Path(shelly.output_dir) / "test_1_real.nii" assert res.output.out_list[1] == attr.NOTHING + + +def test_shellspec_cmd_absolute_path(tmpdir): + """test the 'absolute_path' metadata option. An output path marked with this should not be appended + to the nodes output_dir.""" + input_spec = SpecInfo( + name="Input", + fields=[ + ( + "out_dir", + attr.ib( + type=str, + metadata={ + "help_string": """ + base name of the pretend outputs. + """, + "mandatory": True, + "argstr": "{out_dir}/test_1.nii", + }, + ), + ) + ], + bases=(ShellSpec,), + ) + out_spec = SpecInfo( + name="Output", + fields=[ + ( + "out_1", + attr.ib( + type=File, + metadata={ + "help_string": "fictional output #1", + "output_file_template": "{out_dir}/test_1.nii", + "absolute_path": True, + }, + ), + ), + ], + bases=(ShellOutSpec,), + ) + + shelly = ShellCommandTask( + executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir + ) + shelly() + res = shelly.result() + # checking if the output was created + assert (Path(tmpdir) / Path("test_1.nii")).exists() + # check if path of the output is correct + assert res.output.out_1 == Path(tmpdir) / Path("test_1.nii") + + +def test_shellspec_cmd_absolute_path_not_absolute(tmpdir): + """test the 'absolute_path' metadata option. An output path marked with this should not be appended + to the nodes output_dir. This test checks if an output named 'test_1.nii' that is an absolute_path + is set to NOTHING in case that file exists in the output_dir.""" + input_spec = SpecInfo( + name="Input", + fields=[ + ( + "out_dir", + attr.ib( + type=str, + metadata={ + "help_string": """ + base name of the pretend outputs. + """, + "mandatory": True, + # creating the file in out_dir and the nodes working directory + "argstr": "{out_dir}/test_1.nii test_1.nii", + }, + ), + ) + ], + bases=(ShellSpec,), + ) + out_spec = SpecInfo( + name="Output", + fields=[ + ( + "out_1", + attr.ib( + type=File, + metadata={ + "help_string": "fictional output #1", + "output_file_template": "test_1.nii", + "absolute_path": True, + }, + ), + ), + ], + bases=(ShellOutSpec,), + ) + + shelly = ShellCommandTask( + executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir + ) + shelly() + res = shelly.result() + # checking if the output was created + assert (Path(tmpdir) / Path("test_1.nii")).exists() + assert (Path(shelly.output_dir) / Path("test_1.nii")).exists() + # check if path of the output is correct + assert res.output.out_1 == attr.NOTHING + + +def test_shellspec_cmd_absolute_path_non_existing(tmpdir): + """test the 'absolute_path' metadata option. An output path marked with this should not be appended + to the nodes output_dir. Testing for a existing and non existing file.""" + input_spec = SpecInfo( + name="Input", + fields=[ + ( + "out_dir", + attr.ib( + type=str, + metadata={ + "help_string": """ + base name of the pretend outputs. + """, + "mandatory": True, + "argstr": "{out_dir}/test.nii", + }, + ), + ) + ], + bases=(ShellSpec,), + ) + out_spec = SpecInfo( + name="Output", + fields=[ + ( + "out_1", + attr.ib( + type=File, + metadata={ + "help_string": "fictional output #1", + "output_file_template": "{out_dir}/test_1.nii", + "absolute_path": True, + }, + ), + ), + ( + "out_2", + attr.ib( + type=File, + metadata={ + "help_string": "fictional output #1", + "output_file_template": "{out_dir}/test.nii", + "absolute_path": True, + }, + ), + ), + ], + bases=(ShellOutSpec,), + ) + + shelly = ShellCommandTask( + executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir + ) + shelly() + res = shelly.result() + # checking if the output was created + assert (Path(tmpdir) / Path("test.nii")).exists() + # check if path of the out_1 is correct + assert res.output.out_1 == attr.NOTHING + # check if path of the out_1 is correct + assert res.output.out_2 == (Path(tmpdir) / Path("test.nii")) + + +def test_shellspec_cmd_absolute_path_non_existing_multi(tmpdir): + """test the 'absolute_path' metadata option. An output path marked with this should not be appended + to the nodes output_dir. Testing for a existing and non existing file in multiOutputObj.""" + input_spec = SpecInfo( + name="Input", + fields=[ + ( + "out_name", + attr.ib( + type=MultiInputObj, + metadata={ + "help_string": """ + base name of the pretend outputs. + """, + "mandatory": True, + "argstr": "...", + }, + ), + ) + ], + bases=(ShellSpec,), + ) + out_spec = SpecInfo( + name="Output", + fields=[ + ( + "out_list", + attr.ib( + type=MultiOutputFile, + metadata={ + "help_string": "fictional output #1", + "output_file_template": "{out_name}", + "absolute_path": True, + }, + ), + ), + ], + bases=(ShellOutSpec,), + ) + + shelly = ShellCommandTask( + executable="touch", + input_spec=input_spec, + output_spec=out_spec, + out_name=[str(Path(tmpdir) / "test_1.nii"), "test_2.nii"], + ) + shelly() + res = shelly.result() + # checking if the outputs are Nothing + assert res.output.out_list[0] == Path(tmpdir) / "test_1.nii" + assert res.output.out_list[1] == attr.NOTHING + + +def test_shellspec_formatter_1(tmpdir): + """test the input callable 'formatter'.""" + + def spec_info(formatter): + return SpecInfo( + name="Input", + fields=[ + ( + "in1", + attr.ib( + type=str, + metadata={ + "help_string": """ + just a dummy name + """, + "mandatory": True, + }, + ), + ), + ( + "in2", + attr.ib( + type=str, + metadata={ + "help_string": """ + just a dummy name + """, + "mandatory": True, + }, + ), + ), + ( + "together", + attr.ib( + type=ty.List, + metadata={ + "help_string": """ + combines in1 and in2 into a list + """, + # When providing a formatter all other metadata options are discarded. + "formatter": formatter, + }, + ), + ), + ], + bases=(ShellSpec,), + ) + + def formatter_1(inputs): + print("FORMATTER:", inputs) + return f"-t [{inputs['in1']}, {inputs['in2']}]" + + input_spec = spec_info(formatter_1) + shelly = ShellCommandTask( + executable="exec", input_spec=input_spec, in1="i1", in2="i2" + ) + assert shelly.cmdline == "exec -t [i1, i2]" + + # testing that the formatter can overwrite a provided value for together. + shelly = ShellCommandTask( + executable="exec", + input_spec=input_spec, + in1="i1", + in2="i2", + together=[1], + ) + assert shelly.cmdline == "exec -t [i1, i2]" + + # asking for specific inputs + def formatter_2(in1, in2): + print("FORMATTER:", in1, in2) + return f"-t [{in1}, {in2}]" + + input_spec = spec_info(formatter_2) + + shelly = ShellCommandTask( + executable="exec", input_spec=input_spec, in1="i1", in2="i2" + ) + assert shelly.cmdline == "exec -t [i1, i2]" + + def formatter_3(in1, in3): + print("FORMATTER:", in1, in3) + return f"-t [{in1}, {in3}]" + + input_spec = spec_info(formatter_3) + + shelly = ShellCommandTask( + executable="exec", input_spec=input_spec, in1="i1", in2="i2" + ) + with pytest.raises(Exception) as excinfo: + shelly.cmdline + assert ( + "arguments of the formatter function from together has to be in inputs or be field or output_dir, but in3 is used" + == str(excinfo.value) + ) + + # chcking if field value is accessible when None + def formatter_5(field): + assert field == "-t test" + # formatter must return a string + return field + + input_spec = spec_info(formatter_5) + + shelly = ShellCommandTask( + executable="exec", + input_spec=input_spec, + in1="i1", + in2="i2", + together="-t test", + ) + assert shelly.cmdline == "exec -t test" + + # chcking if field value is accessible when None + def formatter_4(field): + assert field == None + # formatter must return a string + return "" + + input_spec = spec_info(formatter_4) + + shelly = ShellCommandTask( + executable="exec", input_spec=input_spec, in1="i1", in2="i2" + ) + assert shelly.cmdline == "exec" + + +def test_shellspec_formatter_splitter_2(tmpdir): + """test the input callable 'formatter' when a splitter is used on an argument of the formatter.""" + + def spec_info(formatter): + return SpecInfo( + name="Input", + fields=[ + ( + "in1", + attr.ib( + type=str, + metadata={ + "help_string": "in1", + }, + ), + ), + ( + "in2", + attr.ib( + type=str, + metadata={ + "help_string": "in2", + }, + ), + ), + ( + "together", + attr.ib( + type=ty.List, + metadata={ + "help_string": """ + uses in1 + """, + # When providing a formatter all other metadata options are discarded. + "formatter": formatter, + }, + ), + ), + ], + bases=(ShellSpec,), + ) + + # asking for specific inputs + def formatter_1(in1, in2): + return f"-t [{in1} {in2}]" + + input_spec = spec_info(formatter_1) + in1 = ["in11", "in12"] + shelly = ShellCommandTask( + name="f", executable="executable", input_spec=input_spec, in1=in1, in2="in2" + ).split("in1") + assert shelly != None + + results = shelly.cmdline + assert len(results) == 2 + com_results = ["executable -t [in11 in2]", "executable -t [in12 in2]"] + for i, cr in enumerate(com_results): + assert results[i] == cr diff --git a/pydra/engine/tests/test_shelltask_inputspec.py b/pydra/engine/tests/test_shelltask_inputspec.py index 4e31918eb9..364b12cd57 100644 --- a/pydra/engine/tests/test_shelltask_inputspec.py +++ b/pydra/engine/tests/test_shelltask_inputspec.py @@ -1720,57 +1720,6 @@ def test_shell_cmd_inputs_template_10(): assert shelly.output_names == ["return_code", "stdout", "stderr", "outA"] -def test_shell_cmd_inputs_template_11(): - input_fields = [ - ( - "inputFiles", - attr.ib( - type=MultiInputFile, - metadata={ - "argstr": "...", - "help_string": "The list of input image files to be segmented.", - }, - ), - ) - ] - - output_fields = [ - ( - "outputFiles", - attr.ib( - type=MultiOutputFile, - metadata={ - "help_string": "Corrected Output Images: should specify the same number of images as inputVolume, if only one element is given, then it is used as a file pattern where %s is replaced by the imageVolumeType, and %d by the index list location.", - "output_file_template": "{inputFiles}", - }, - ), - ) - ] - - input_spec = SpecInfo(name="Input", fields=input_fields, bases=(ShellSpec,)) - output_spec = SpecInfo(name="Output", fields=output_fields, bases=(ShellOutSpec,)) - - task = ShellCommandTask( - name="echoMultiple", - executable="touch", - input_spec=input_spec, - output_spec=output_spec, - ) - wf = Workflow(name="wf", input_spec=["inputFiles"], inputFiles=["test1", "test2"]) - - task.inputs.inputFiles = wf.lzin.inputFiles - - wf.add(task) - wf.set_output([("out", wf.echoMultiple.lzout.outputFiles)]) - - with Submitter(plugin="cf") as sub: - sub(wf) - result = wf.result() - - for out_file in result.output.out: - assert out_file.name == "test1" or out_file.name == "test2" - - def test_shell_cmd_inputs_template_1_st(): """additional inputs, one uses output_file_template (and argstr) testing cmdline when splitter defined @@ -2073,412 +2022,3 @@ def test_shell_cmd_inputs_di(tmpdir, use_validator): image_dimensionality=5, ) assert "value of image_dimensionality" in str(excinfo.value) - - -def test_shellspec_cmd_absolute_path(tmpdir): - """test the 'absolute_path' metadata option. An output path marked with this should not be appended - to the nodes output_dir.""" - input_spec = SpecInfo( - name="Input", - fields=[ - ( - "out_dir", - attr.ib( - type=str, - metadata={ - "help_string": """ - base name of the pretend outputs. - """, - "mandatory": True, - "argstr": "{out_dir}/test_1.nii", - }, - ), - ) - ], - bases=(ShellSpec,), - ) - out_spec = SpecInfo( - name="Output", - fields=[ - ( - "out_1", - attr.ib( - type=File, - metadata={ - "help_string": "fictional output #1", - "output_file_template": "{out_dir}/test_1.nii", - "absolute_path": True, - }, - ), - ), - ], - bases=(ShellOutSpec,), - ) - - shelly = ShellCommandTask( - executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir - ) - shelly() - res = shelly.result() - # checking if the output was created - assert (Path(tmpdir) / Path("test_1.nii")).exists() - # check if path of the output is correct - assert res.output.out_1 == Path(tmpdir) / Path("test_1.nii") - - -def test_shellspec_cmd_absolute_path_not_absolute(tmpdir): - """test the 'absolute_path' metadata option. An output path marked with this should not be appended - to the nodes output_dir. This test checks if an output named 'test_1.nii' that is an absolute_path - is set to NOTHING in case that file exists in the output_dir.""" - input_spec = SpecInfo( - name="Input", - fields=[ - ( - "out_dir", - attr.ib( - type=str, - metadata={ - "help_string": """ - base name of the pretend outputs. - """, - "mandatory": True, - # creating the file in out_dir and the nodes working directory - "argstr": "{out_dir}/test_1.nii test_1.nii", - }, - ), - ) - ], - bases=(ShellSpec,), - ) - out_spec = SpecInfo( - name="Output", - fields=[ - ( - "out_1", - attr.ib( - type=File, - metadata={ - "help_string": "fictional output #1", - "output_file_template": "test_1.nii", - "absolute_path": True, - }, - ), - ), - ], - bases=(ShellOutSpec,), - ) - - shelly = ShellCommandTask( - executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir - ) - shelly() - res = shelly.result() - # checking if the output was created - assert (Path(tmpdir) / Path("test_1.nii")).exists() - assert (Path(shelly.output_dir) / Path("test_1.nii")).exists() - # check if path of the output is correct - assert res.output.out_1 == attr.NOTHING - - -def test_shellspec_cmd_absolute_path_non_existing(tmpdir): - """test the 'absolute_path' metadata option. An output path marked with this should not be appended - to the nodes output_dir. Testing for a existing and non existing file.""" - input_spec = SpecInfo( - name="Input", - fields=[ - ( - "out_dir", - attr.ib( - type=str, - metadata={ - "help_string": """ - base name of the pretend outputs. - """, - "mandatory": True, - "argstr": "{out_dir}/test.nii", - }, - ), - ) - ], - bases=(ShellSpec,), - ) - out_spec = SpecInfo( - name="Output", - fields=[ - ( - "out_1", - attr.ib( - type=File, - metadata={ - "help_string": "fictional output #1", - "output_file_template": "{out_dir}/test_1.nii", - "absolute_path": True, - }, - ), - ), - ( - "out_2", - attr.ib( - type=File, - metadata={ - "help_string": "fictional output #1", - "output_file_template": "{out_dir}/test.nii", - "absolute_path": True, - }, - ), - ), - ], - bases=(ShellOutSpec,), - ) - - shelly = ShellCommandTask( - executable="touch", input_spec=input_spec, output_spec=out_spec, out_dir=tmpdir - ) - shelly() - res = shelly.result() - # checking if the output was created - assert (Path(tmpdir) / Path("test.nii")).exists() - # check if path of the out_1 is correct - assert res.output.out_1 == attr.NOTHING - # check if path of the out_1 is correct - assert res.output.out_2 == (Path(tmpdir) / Path("test.nii")) - - -def test_shellspec_cmd_absolute_path_non_existing_multi(tmpdir): - """test the 'absolute_path' metadata option. An output path marked with this should not be appended - to the nodes output_dir. Testing for a existing and non existing file in multiOutputObj.""" - input_spec = SpecInfo( - name="Input", - fields=[ - ( - "out_name", - attr.ib( - type=MultiInputObj, - metadata={ - "help_string": """ - base name of the pretend outputs. - """, - "mandatory": True, - "argstr": "...", - }, - ), - ) - ], - bases=(ShellSpec,), - ) - out_spec = SpecInfo( - name="Output", - fields=[ - ( - "out_list", - attr.ib( - type=MultiOutputFile, - metadata={ - "help_string": "fictional output #1", - "output_file_template": "{out_name}", - "absolute_path": True, - }, - ), - ), - ], - bases=(ShellOutSpec,), - ) - - shelly = ShellCommandTask( - executable="touch", - input_spec=input_spec, - output_spec=out_spec, - out_name=[str(Path(tmpdir) / "test_1.nii"), "test_2.nii"], - ) - shelly() - res = shelly.result() - # checking if the outputs are Nothing - assert res.output.out_list[0] == Path(tmpdir) / "test_1.nii" - assert res.output.out_list[1] == attr.NOTHING - - -def test_shellspec_formatter_1(tmpdir): - """test the input callable 'formatter'.""" - - def spec_info(formatter): - return SpecInfo( - name="Input", - fields=[ - ( - "in1", - attr.ib( - type=str, - metadata={ - "help_string": """ - just a dummy name - """, - "mandatory": True, - }, - ), - ), - ( - "in2", - attr.ib( - type=str, - metadata={ - "help_string": """ - just a dummy name - """, - "mandatory": True, - }, - ), - ), - ( - "together", - attr.ib( - type=ty.List, - metadata={ - "help_string": """ - combines in1 and in2 into a list - """, - # When providing a formatter all other metadata options are discarded. - "formatter": formatter, - }, - ), - ), - ], - bases=(ShellSpec,), - ) - - def formatter_1(inputs): - print("FORMATTER:", inputs) - return f"-t [{inputs['in1']}, {inputs['in2']}]" - - input_spec = spec_info(formatter_1) - shelly = ShellCommandTask( - executable="exec", input_spec=input_spec, in1="i1", in2="i2" - ) - assert shelly.cmdline == "exec -t [i1, i2]" - - # testing that the formatter can overwrite a provided value for together. - shelly = ShellCommandTask( - executable="exec", - input_spec=input_spec, - in1="i1", - in2="i2", - together=[1], - ) - assert shelly.cmdline == "exec -t [i1, i2]" - - # asking for specific inputs - def formatter_2(in1, in2): - print("FORMATTER:", in1, in2) - return f"-t [{in1}, {in2}]" - - input_spec = spec_info(formatter_2) - - shelly = ShellCommandTask( - executable="exec", input_spec=input_spec, in1="i1", in2="i2" - ) - assert shelly.cmdline == "exec -t [i1, i2]" - - def formatter_3(in1, in3): - print("FORMATTER:", in1, in3) - return f"-t [{in1}, {in3}]" - - input_spec = spec_info(formatter_3) - - shelly = ShellCommandTask( - executable="exec", input_spec=input_spec, in1="i1", in2="i2" - ) - with pytest.raises(Exception) as excinfo: - shelly.cmdline - assert ( - "arguments of the formatter function from together has to be in inputs or be field or output_dir, but in3 is used" - == str(excinfo.value) - ) - - # chcking if field value is accessible when None - def formatter_5(field): - assert field == "-t test" - # formatter must return a string - return field - - input_spec = spec_info(formatter_5) - - shelly = ShellCommandTask( - executable="exec", - input_spec=input_spec, - in1="i1", - in2="i2", - together="-t test", - ) - assert shelly.cmdline == "exec -t test" - - # chcking if field value is accessible when None - def formatter_4(field): - assert field == None - # formatter must return a string - return "" - - input_spec = spec_info(formatter_4) - - shelly = ShellCommandTask( - executable="exec", input_spec=input_spec, in1="i1", in2="i2" - ) - assert shelly.cmdline == "exec" - - -def test_shellspec_formatter_splitter_2(tmpdir): - """test the input callable 'formatter' when a splitter is used on an argument of the formatter.""" - - def spec_info(formatter): - return SpecInfo( - name="Input", - fields=[ - ( - "in1", - attr.ib( - type=str, - metadata={ - "help_string": "in1", - }, - ), - ), - ( - "in2", - attr.ib( - type=str, - metadata={ - "help_string": "in2", - }, - ), - ), - ( - "together", - attr.ib( - type=ty.List, - metadata={ - "help_string": """ - uses in1 - """, - # When providing a formatter all other metadata options are discarded. - "formatter": formatter, - }, - ), - ), - ], - bases=(ShellSpec,), - ) - - # asking for specific inputs - def formatter_1(in1, in2): - return f"-t [{in1} {in2}]" - - input_spec = spec_info(formatter_1) - in1 = ["in11", "in12"] - shelly = ShellCommandTask( - name="f", executable="executable", input_spec=input_spec, in1=in1, in2="in2" - ).split("in1") - assert shelly != None - - results = shelly.cmdline - assert len(results) == 2 - com_results = ["executable -t [in11 in2]", "executable -t [in12 in2]"] - for i, cr in enumerate(com_results): - assert results[i] == cr