FIX: Use traits to provide type hints when modifying paths

nipype/pipeline/engine/tests/test_utils.py

@@ -16,7 +16,7 @@
 from ....interfaces import base as nib
 from ....interfaces import utility as niu
 from .... import config
-from ..utils import clean_working_directory, write_workflow_prov
+from ..utils import clean_working_directory, write_workflow_prov, modify_paths
 
 
 class InputSpec(nib.TraitedSpec):
@@ -224,3 +224,60 @@ def test_mapnode_crash3(tmpdir):
     wf.config["execution"]["crashdump_dir"] = os.getcwd()
     with pytest.raises(RuntimeError):
         wf.run(plugin='Linear')
+
+
+class StrPathConfuser(nib.SimpleInterface):
+    class input_spec(nib.TraitedSpec):
+        in_str = nib.traits.String()
+
+    class output_spec(nib.TraitedSpec):
+        out_tuple = nib.traits.Tuple(nib.File, nib.traits.String)
+        out_dict_path = nib.traits.Dict(nib.traits.String, nib.File(exists=True))
+        out_dict_str = nib.traits.DictStrStr()
+        out_list = nib.traits.List(nib.traits.String)
+        out_str = nib.traits.String()
+        out_path = nib.File(exists=True)
+
+    def _run_interface(self, runtime):
+        out_path = os.path.abspath(os.path.basename(self.inputs.in_str) + '_path')
+        open(out_path, 'w').close()
+
+        self._results['out_str'] = self.inputs.in_str
+        self._results['out_path'] = out_path
+        self._results['out_tuple'] = (out_path, self.inputs.in_str)
+        self._results['out_dict_path'] = {self.inputs.in_str: out_path}
+        self._results['out_dict_str'] = {self.inputs.in_str: self.inputs.in_str}
+        self._results['out_list'] = [self.inputs.in_str] * 2
+        return runtime
+
+
+def test_modify_paths_bug(tmpdir):
+    """
+    There was a bug in which, if the current working directory contained a file with the name
+    of an output String, the string would get transformed into a path, and generally wreak havoc.
+
+    This attempts to replicate that condition, using an object with strings and paths in various
+    trait configurations, to ensure that the guards added resolve the issue.
+
+    Please see https://github.com/nipy/nipype/issues/2944 for more details.
+    """
+    tmpdir.chdir()
+
+    spc = pe.Node(StrPathConfuser(in_str='2'), name='spc')
+
+    open('2', 'w').close()
+
+    results = spc.run()
+
+    # Basic check that string was not manipulated and path exists
+    out_path = results.outputs.out_path
+    out_str = results.outputs.out_str
+    assert out_str == '2'
+    assert os.path.basename(out_path) == '2_path'
+    assert os.path.isfile(out_path)
+
+    # Assert data structures pass through correctly
+    assert results.outputs.out_tuple == (out_path, out_str)
+    assert results.outputs.out_dict_path == {out_str: out_path}
+    assert results.outputs.out_dict_str == {out_str: out_str}
+    assert results.outputs.out_list == [out_str] * 2

nipype/pipeline/engine/utils.py

@@ -25,7 +25,7 @@
 
 from ... import logging, config, LooseVersion
 from ...utils.filemanip import (
-    relpath,
+    resolve_path,
     makedirs,
     fname_presuffix,
     to_str,
@@ -41,7 +41,7 @@
 from ...utils.misc import str2bool
 from ...utils.functions import create_function_from_source
 from ...interfaces.base import (Bunch, CommandLine, isdefined, Undefined,
-                                InterfaceResult, traits)
+                                InterfaceResult, traits, File)
 from ...interfaces.utility import IdentityInterface
 from ...utils.provenance import ProvStore, pm, nipype_ns, get_id
 
@@ -297,7 +297,15 @@ def save_resultfile(result, cwd, name):
             tosave = _uncollapse(outputs.copy(), collapsed)
         except AttributeError:
             tosave = outputs = result.outputs.dictcopy()  # outputs was a bunch
-        for k, v in list(modify_paths(tosave, relative=True, basedir=cwd).items()):
+        type_hints = {}
+        try:
+            type_hints = result.outputs.traits()
+        except AttributeError:
+            pass
+        for k, v in modify_paths(tosave,
+                                 relative=config.getboolean('execution', 'use_relative_paths'),
+                                 type_hints=type_hints,
+                                 basedir=cwd).items():
             setattr(result.outputs, k, v)
 
     savepkl(resultsfile, result)
@@ -352,10 +360,13 @@ def load_resultfile(path, name):
             if result.outputs:
                 try:
                     outputs = _protect_collapses(result.outputs)
+                    type_hints = result.outputs.traits()
                 except AttributeError:
                     outputs = result.outputs.dictcopy()  # outputs == Bunch
+                    type_hints = {}
                 try:
                     for k, v in list(modify_paths(outputs, relative=False,
+                                                  type_hints=type_hints,
                                                   basedir=path).items()):
                         setattr(result.outputs, k, v)
                 except FileNotFoundError:
@@ -455,7 +466,7 @@ def format_node(node, format='python', include_config=False):
     return lines
 
 
-def modify_paths(object, relative=True, basedir=None):
+def modify_paths(object, relative=True, basedir=None, type_hints=None):
     """Convert paths in data structure to either full paths or relative paths
 
     Supports combinations of lists, dicts, tuples, strs
@@ -472,34 +483,44 @@ def modify_paths(object, relative=True, basedir=None):
         basedir = os.getcwd()
     if isinstance(object, dict):
         out = {}
+        if isinstance(type_hints, traits.CTrait) and isinstance(type_hints.trait_type, traits.Dict):
+            type_hints_ = type_hints.trait_type.value_trait
         for key, val in sorted(object.items()):
+            if isinstance(type_hints, dict):
+                type_hints_ = type_hints.get(key)
             if isdefined(val):
                 out[key] = modify_paths(
-                    val, relative=relative, basedir=basedir)
+                    val, relative=relative, basedir=basedir,
+                    type_hints=type_hints_)
     elif isinstance(object, (list, tuple)):
         out = []
-        for val in object:
+        type_hints_ = [None] * len(object)
+        if type_hints is not None:
+            if isinstance(type_hints.trait_type, traits.List):
+                type_hints_ = [type_hints.trait_type.item_trait] * len(object)
+            elif isinstance(type_hints.trait_type, traits.Tuple):
+                type_hints_ = type_hints.trait_type.types
+        for val, type_hint in zip(object, type_hints_):
             if isdefined(val):
                 out.append(
-                    modify_paths(val, relative=relative, basedir=basedir))
+                    modify_paths(val, relative=relative, basedir=basedir,
+                                 type_hints=type_hint))
         if isinstance(object, tuple):
             out = tuple(out)
     else:
-        if isdefined(object):
-            if isinstance(object, (str, bytes)) and os.path.isfile(object):
-                if relative:
-                    if config.getboolean('execution', 'use_relative_paths'):
-                        out = relpath(object, start=basedir)
-                    else:
-                        out = object
-                else:
-                    out = os.path.abspath(os.path.join(basedir, object))
-                if not os.path.exists(out):
-                    raise IOError('File %s not found' % out)
-            else:
-                out = object
-        else:
+        if not isdefined(object):
             raise TypeError("Object {} is undefined".format(object))
+        out = object
+        check_path = type_hints is None or isinstance(type_hints.trait_type, File)
+        if (check_path and
+                isinstance(object, (str, bytes)) and
+                os.path.isfile(resolve_path(object, start=basedir))):
+            out = resolve_path(
+                object, start=basedir,
+                relative=relative and config.getboolean('execution', 'use_relative_paths'))
+            # I don't think this can be reached
+            if not os.path.exists(out):
+                raise IOError('File %s not found' % out)
     return out
 
 

nipype/utils/filemanip.py

@@ -961,40 +961,22 @@ def canonicalize_env(env):
     return out_env
 
 
-def relpath(path, start=None):
-    """Return a relative version of a path"""
-
-    try:
-        return op.relpath(path, start)
-    except AttributeError:
-        pass
+def resolve_path(path, start=None, relative=False):
+    if relative:
+        return op.relpath(resolve_path(path, start), start)
 
+    if op.isabs(path):
+        return path
     if start is None:
-        start = os.curdir
-    if not path:
-        raise ValueError("no path specified")
-    start_list = op.abspath(start).split(op.sep)
-    path_list = op.abspath(path).split(op.sep)
-    if start_list[0].lower() != path_list[0].lower():
-        unc_path, rest = op.splitunc(path)
-        unc_start, rest = op.splitunc(start)
-        if bool(unc_path) ^ bool(unc_start):
-            raise ValueError(("Cannot mix UNC and non-UNC paths "
-                              "(%s and %s)") % (path, start))
-        else:
-            raise ValueError("path is on drive %s, start on drive %s" %
-                             (path_list[0], start_list[0]))
-    # Work out how much of the filepath is shared by start and path.
-    for i in range(min(len(start_list), len(path_list))):
-        if start_list[i].lower() != path_list[i].lower():
-            break
-    else:
-        i += 1
+        start = os.getcwd()
+    return op.abspath(op.join(start, path))
+
 
-    rel_list = [op.pardir] * (len(start_list) - i) + path_list[i:]
-    if not rel_list:
-        return os.curdir
-    return op.join(*rel_list)
+def relpath(path, start=None):
+    """Return a relative version of a path"""
+    warnings.warn('nipype.utils.filemanip.relpath will be removed; please us os.path.relpath',
+                  FutureWarning)
+    return op.relpath(path, start)
 
 
 @contextlib.contextmanager