A test case for output removal

Author: chrisgorgo

nipype/pipeline/tests/test_utils.py

@@ -86,6 +86,99 @@ def test_function(arg1):
  'file2.txt'))
  rmtree(out_dir)
 
+
+def test_outputs_removal_wf():
+
+ def test_function(arg1):
+ import os
+ file1 = os.path.join(os.getcwd(), 'file1.txt')
+ file2 = os.path.join(os.getcwd(), 'file2.txt')
+ fp = open(file1, 'wt')
+ fp.write('%d' % arg1)
+ fp.close()
+ fp = open(file2, 'wt')
+ fp.write('%d' % arg1)
+ fp.close()
+ return file1, file2
+
+ def test_function2(in_file, arg):
+ import os
+ in_arg = open(in_file).read()
+ file1 = os.path.join(os.getcwd(), 'file1.txt')
+ file2 = os.path.join(os.getcwd(), 'file2.txt')
+ fp = open(file1, 'wt')
+ fp.write('%d' % arg + in_arg)
+ fp.close()
+ fp = open(file2, 'wt')
+ fp.write('%d' % arg + in_arg)
+ fp.close()
+ return file1, file2, 1
+ 
+ def test_function3(arg):
+ import os
+ return arg
+
+ out_dir = mkdtemp()
+
+ for plugin in ('Linear', 'MultiProc'):
+ n1 = pe.Node(niu.Function(input_names=['arg1'],
+ output_names=['out_file1', 'out_file2'],
+ function=test_function),
+ name='n1')
+ n1.inputs.arg1 = 1
+
+ n2 = pe.Node(niu.Function(input_names=['in_file', 'arg'],
+ output_names=['out_file1', 'out_file2', 'n'],
+ function=test_function2),
+ name='n2')
+ n2.inputs.arg = 2
+
+ n3 = pe.Node(niu.Function(input_names=['arg'],
+ output_names=['n'],
+ function=test_function3),
+ name='n3')
+
+ wf = pe.Workflow(name="node_rem_test" + plugin, base_dir=out_dir)
+ wf.connect(n1, "out_file1", n2, "in_file")
+
+ for remove_unnecessary_outputs in [True, False]:
+ wf.config = {'execution': {'remove_unnecessary_outputs': remove_unnecessary_outputs}}
+ wf.config = merge_dict(deepcopy(config._sections), wf.config)
+ wf.run(plugin=plugin)
+
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n1.name,
+ 'file2.txt')) != remove_unnecessary_outputs
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n1.name,
+ 'file1.txt'))
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n2.name,
+ 'file1.txt'))
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n2.name,
+ 'file2.txt'))
+ wf.connect(n2, "n", n3, "arg")
+ for remove_unnecessary_outputs in [True, False]:
+ wf.config = {'execution': {'remove_unnecessary_outputs': remove_unnecessary_outputs}}
+ wf.config = merge_dict(deepcopy(config._sections), wf.config)
+ wf.run(plugin=plugin)
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n2.name,
+ 'file1.txt')) != remove_unnecessary_outputs
+ yield assert_true, os.path.exists(os.path.join(wf.base_dir,
+ wf.name,
+ n2.name,
+ 'file2.txt')) != remove_unnecessary_outputs
+ 
+ rmtree(out_dir)
+
+
 class InputSpec(nib.TraitedSpec):
  in_file = nib.File(exists=True, copyfile=True)