tst: add prov test

Author: satra

nipype/pipeline/tests/test_utils.py

@@ -14,7 +14,7 @@
 import nipype.interfaces.base as nib
 import nipype.interfaces.utility as niu
 from ... import config
-from ..utils import merge_dict, clean_working_directory
+from ..utils import merge_dict, clean_working_directory, write_workflow_prov
 
 
 def test_identitynode_removal():
@@ -330,3 +330,15 @@ def test_multi_disconnected_iterable():
     eg = metawf.run(plugin='Linear')
     yield assert_equal, len(eg.nodes()), 60
     rmtree(out_dir)
+
+def test_provenance():
+    out_dir = mkdtemp()
+    metawf = pe.Workflow(name='meta')
+    metawf.base_dir = out_dir
+    metawf.add_nodes([create_wf('wf%d' % i) for i in range(30)])
+    eg = metawf.run(plugin='Linear')
+    prov_base = os.path.join(out_dir,
+                             'workflow_provenance_test')
+    psg = write_workflow_prov(eg, prov_base, format='all')
+    yield assert_equal, psg, 1
+    rmtree(out_dir)

nipype/pipeline/utils.py

@@ -1173,7 +1173,7 @@ def write_workflow_prov(graph, filename=None, format='all'):
                 ps.g.wasGeneratedBy(sub_bundle, process)
         else:
             process.add_attributes({pm.PROV["type"]: nipype_ns["Node"]})
-            result_bundle = ProvStore().add_results(result)
+            result_bundle = ProvStore().add_results(result).bundle()
             ps.g = merge_bundles(ps.g, result_bundle)
             ps.g.wasGeneratedBy(result_bundle, process)
         processes.append(process)