fix: tests for mapnode crash

Author: satra

nipype/pipeline/engine/nodes.py

@@ -36,7 +36,7 @@
 from hashlib import sha1
 
 from ... import config, logging
-from ...utils.misc import (flatten, unflatten, package_check, str2bool)
+from ...utils.misc import (flatten, unflatten, str2bool)
 from ...utils.filemanip import (save_json, FileNotFoundError,
                                 filename_to_list, list_to_filename,
                                 copyfiles, fnames_presuffix, loadpkl,
@@ -54,7 +54,6 @@
                     get_print_name, merge_dict, evaluate_connect_function)
 from .base import EngineBase
 
-package_check('networkx', '1.3')
 logger = logging.getLogger('workflow')
 
 class Node(EngineBase):
@@ -607,6 +606,7 @@ def _run_command(self, execute, copyfiles=True):
             try:
                 result = self._interface.run()
             except Exception as msg:
+                self._save_results(result, cwd)
                 self._result.runtime.stderr = msg
                 raise
 
@@ -1124,6 +1124,7 @@ def _make_nodes(self, cwd=None):
             yield i, node
 
     def _node_runner(self, nodes, updatehash=False):
+        old_cwd = os.getcwd()
         for i, node in nodes:
             err = None
             try:
@@ -1133,6 +1134,7 @@ def _node_runner(self, nodes, updatehash=False):
                 if str2bool(self.config['execution']['stop_on_first_crash']):
                     raise
             finally:
+                os.chdir(old_cwd)
                 yield i, node, err
 
     def _collate_results(self, nodes):
@@ -1275,12 +1277,8 @@ def _run_interface(self, execute=True, updatehash=False):
                 nitems = len(filename_to_list(getattr(self.inputs,
                                                       self.iterfield[0])))
             nodenames = ['_' + self.name + str(i) for i in range(nitems)]
-            nodes = list(self._make_nodes(cwd))
-            node_results = list(self._node_runner(nodes))
-            self._collate_results(node_results)
-            # map-reduce formulation
-            #self._collate_results(self._node_runner(self._make_nodes(cwd),
-            #                                        updatehash=updatehash))
+            self._collate_results(self._node_runner(self._make_nodes(cwd),
+                                                    updatehash=updatehash))
             self._save_results(self._result, cwd)
             # remove any node directories no longer required
             dirs2remove = []

nipype/pipeline/engine/tests/test_utils.py

@@ -338,36 +338,37 @@ def test_provenance(tmpdir):
     assert len(psg.bundles) == 2
     assert len(psg.get_records()) == 7
 
+
+def dummy_func(value):
+    return value + 1
+
+
 def test_mapnode_crash(tmpdir):
     """Test mapnode crash when stop_on_first_crash is True"""
     cwd = os.getcwd()
-    def myfunction(string):
-        return string + 'meh'
     node = pe.MapNode(niu.Function(input_names=['WRONG'],
                                    output_names=['newstring'],
-                                   function=myfunction),
+                                   function=dummy_func),
                       iterfield=['WRONG'],
                       name='myfunc')
-    node.inputs.WRONG = ['string' + str(i) for i in range(3)]
+    node.inputs.WRONG = ['string{}'.format(i) for i in range(3)]
     node.config = deepcopy(config._sections)
     node.config['execution']['stop_on_first_crash'] = True
     node.base_dir = str(tmpdir)
     with pytest.raises(TypeError):
         node.run()
     os.chdir(cwd)
 
+
 def test_mapnode_crash2(tmpdir):
     """Test mapnode crash when stop_on_first_crash is False"""
     cwd = os.getcwd()
-    def myfunction(string):
-        return string + 'meh'
-
     node = pe.MapNode(niu.Function(input_names=['WRONG'],
                                    output_names=['newstring'],
-                                   function=myfunction),
+                                   function=dummy_func),
                       iterfield=['WRONG'],
                       name='myfunc')
-    node.inputs.WRONG = ['string' + str(i) for i in range(3)]
+    node.inputs.WRONG = ['string{}'.format(i) for i in range(3)]
     node.base_dir = str(tmpdir)
 
     with pytest.raises(Exception):
@@ -377,15 +378,12 @@ def myfunction(string):
 
 def test_mapnode_crash3(tmpdir):
     """Test mapnode crash when mapnode is embedded in a workflow"""
-
-    def myfunction(string):
-        return string + 'meh'
     node = pe.MapNode(niu.Function(input_names=['WRONG'],
                                    output_names=['newstring'],
-                                   function=myfunction),
+                                   function=dummy_func),
                       iterfield=['WRONG'],
                       name='myfunc')
-    node.inputs.WRONG = ['string' + str(i) for i in range(3)]
+    node.inputs.WRONG = ['string{}'.format(i) for i in range(3)]
     wf = pe.Workflow('testmapnodecrash')
     wf.add_nodes([node])
     wf.base_dir = str(tmpdir)