Merge pull request #195 from djarecka/fix/filelist_hash_recurs

hash for containers (closes #178)

Author: satra

pydra/engine/helpers.py

@@ -10,6 +10,7 @@
 import subprocess as sp
 
 from .specs import Runtime, File, attr_fields
+from .helpers_file import is_existing_file, hash_file
 
 
 def ensure_list(obj, tuple2list=False):
@@ -396,6 +397,27 @@ def hash_function(obj):
     return sha256(str(obj).encode()).hexdigest()
 
 
+def hash_value(value, tp=None, metadata=None):
+    """calculating hash or returning values recursively"""
+    if isinstance(value, (tuple, list)):
+        return [hash_value(el, tp, metadata) for el in value]
+    elif isinstance(value, dict):
+        dict_hash = {k: hash_value(v, tp, metadata) for (k, v) in value.items()}
+        # returning a sorted object
+        return sorted(dict_hash.items(), key=lambda x: x[0])
+    else:  # not a container
+        if (
+            (tp is File or "pydra.engine.specs.File" in str(tp))
+            and is_existing_file(value)
+            and "container_path" not in metadata
+        ):
+            return hash_file(value)
+        elif isinstance(value, tuple):
+            return list(value)
+        else:
+            return value
+
+
 def output_names_from_inputfields(inputs):
     """
     Collect outputs from input fields with output_file_template.

pydra/engine/helpers_file.py

@@ -110,11 +110,6 @@ def fname_presuffix(fname, prefix="", suffix="", newpath=None, use_ext=True):
 def hash_file(afile, chunk_len=8192, crypto=sha256, raise_notfound=True):
     """Compute hash of a file using 'crypto' module."""
     from .specs import LazyField
-    from .helpers import hash_function
-
-    # adding option for tasks with splitter over list of files
-    if isinstance(afile, list):
-        return hash_function([hash_file(el) for el in afile])
 
     if afile is None or isinstance(afile, LazyField) or isinstance(afile, list):
         return None
@@ -517,3 +512,11 @@ def is_local_file(f):
     from .specs import File
 
     return f.type is File and "container_path" not in f.metadata
+
+
+def is_existing_file(f):
+    """ checking if an object is an existing file"""
+    try:
+        return Path(f).exists()
+    except TypeError:
+        return False

pydra/engine/specs.py

@@ -39,21 +39,17 @@ def collect_additional_outputs(self, input_spec, inputs, output_dir):
     @property
     def hash(self):
         """Compute a basic hash for any given set of fields."""
-        from .helpers import hash_function
-        from .helpers_file import hash_file
+        from .helpers import hash_value, hash_function
 
         inp_dict = {}
         for field in attr_fields(self):
             if field.name in ["_graph_checksums", "bindings"] or field.metadata.get(
                 "output_file_template"
             ):
                 continue
-            if field.type == File and "container_path" not in field.metadata:
-                inp_dict[field.name] = hash_file(getattr(self, field.name))
-            elif isinstance(getattr(self, field.name), tuple):
-                inp_dict[field.name] = list(getattr(self, field.name))
-            else:
-                inp_dict[field.name] = getattr(self, field.name)
+            inp_dict[field.name] = hash_value(
+                value=getattr(self, field.name), tp=field.type, metadata=field.metadata
+            )
         inp_hash = hash_function(inp_dict)
         if hasattr(self, "_graph_checksums"):
             return hash_function((inp_hash, self._graph_checksums))

pydra/engine/task.py

@@ -391,7 +391,7 @@ def _run_task(self):
             args = self.command_args
         if args:
             # removing emty strings
-            args = [el for el in args if el not in ["", " "]]
+            args = [str(el) for el in args if el not in ["", " "]]
             keys = ["return_code", "stdout", "stderr"]
             values = execute(args, strip=self.strip)
             self.output_ = dict(zip(keys, values))

pydra/engine/tests/test_node_task.py

@@ -3,7 +3,15 @@
 import numpy as np
 import pytest
 
-from .utils import fun_addtwo, fun_addvar, moment, fun_div
+from .utils import (
+    fun_addtwo,
+    fun_addvar,
+    moment,
+    fun_div,
+    fun_dict,
+    fun_file,
+    fun_file_list,
+)
 
 from ..core import TaskBase
 from ..submitter import Submitter
@@ -283,6 +291,32 @@ def test_task_init_6():
     assert nn.state.states_val == []
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_init_7(plugin, tmpdir):
+    """ task with a dictionary of files as an input, checking checksum"""
+    file1 = tmpdir.join("file1.txt")
+    with open(file1, "w") as f:
+        f.write("hello")
+
+    file2 = tmpdir.join("file2.txt")
+    with open(file2, "w") as f:
+        f.write("from pydra\n")
+
+    nn1 = fun_file_list(name="NA", filename_list=[file1, file2])
+    output_dir1 = nn1.output_dir
+
+    # changing the content of the file
+    file2 = tmpdir.join("file2.txt")
+    with open(file2, "w") as f:
+        f.write("from pydra")
+
+    nn2 = fun_file_list(name="NA", filename_list=[file1, file2])
+    output_dir2 = nn2.output_dir
+
+    # the checksum should be different - content of file2 is different
+    assert output_dir1.name != output_dir2.name
+
+
 def test_task_error():
     func = fun_div(name="div", a=1, b=0)
     with pytest.raises(ZeroDivisionError):
@@ -382,6 +416,64 @@ def test_task_nostate_2(plugin):
     assert nn.output_dir.exists()
 
 
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_nostate_3(plugin):
+    """ task with a dictionary as an input"""
+    nn = fun_dict(name="NA", d={"a": "ala", "b": "bala"})
+    assert nn.inputs.d == {"a": "ala", "b": "bala"}
+
+    with Submitter(plugin=plugin) as sub:
+        sub(nn)
+
+    # checking the results
+    results = nn.result()
+    assert results.output.out == "a:ala_b:bala"
+    # checking the output_dir
+    assert nn.output_dir.exists()
+
+
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_nostate_4(plugin, tmpdir):
+    """ task with a dictionary as an input"""
+    file1 = tmpdir.join("file.txt")
+    with open(file1, "w") as f:
+        f.write("hello from pydra\n")
+
+    nn = fun_file(name="NA", filename=file1)
+
+    with Submitter(plugin=plugin) as sub:
+        sub(nn)
+
+    # checking the results
+    results = nn.result()
+    assert results.output.out == "hello from pydra\n"
+    # checking the output_dir
+    assert nn.output_dir.exists()
+
+
+@pytest.mark.parametrize("plugin", Plugins)
+def test_task_nostate_5(plugin, tmpdir):
+    """ task with a dictionary of files as an input"""
+    file1 = tmpdir.join("file1.txt")
+    with open(file1, "w") as f:
+        f.write("hello")
+
+    file2 = tmpdir.join("file2.txt")
+    with open(file2, "w") as f:
+        f.write("from pydra\n")
+
+    nn = fun_file_list(name="NA", filename_list=[file1, file2])
+
+    with Submitter(plugin=plugin) as sub:
+        sub(nn)
+
+    # checking the results
+    results = nn.result()
+    assert results.output.out == "hello from pydra\n"
+    # checking the output_dir
+    assert nn.output_dir.exists()
+
+
 # Testing caching for tasks without states
 
 

pydra/engine/tests/test_shelltask.py

@@ -910,6 +910,48 @@ def test_shell_cmd_inputspec_7a(plugin, results_function):
     assert res.output.out1_changed.exists()
 
 
+@pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
+@pytest.mark.parametrize("plugin", Plugins)
+def test_shell_cmd_inputspec_8(plugin, results_function, tmpdir):
+    """ using input_spec, providing list of files as an input """
+
+    file_1 = tmpdir.join("file_1.txt")
+    file_2 = tmpdir.join("file_2.txt")
+    with open(file_1, "w") as f:
+        f.write("hello ")
+    with open(file_2, "w") as f:
+        f.write("from boston")
+
+    cmd_exec = "cat"
+    files_list = [file_1, file_2]
+
+    my_input_spec = SpecInfo(
+        name="Input",
+        fields=[
+            (
+                "files",
+                attr.ib(
+                    type=ty.List[File],
+                    metadata={
+                        "position": 1,
+                        "help_string": "list of files",
+                        "mandatory": True,
+                    },
+                ),
+            )
+        ],
+        bases=(ShellSpec,),
+    )
+
+    shelly = ShellCommandTask(
+        name="shelly", executable=cmd_exec, files=files_list, input_spec=my_input_spec
+    )
+
+    assert shelly.inputs.executable == cmd_exec
+    res = results_function(shelly, plugin)
+    assert res.output.stdout == "hello from boston"
+
+
 @pytest.mark.parametrize("results_function", [result_no_submitter, result_submitter])
 @pytest.mark.parametrize("plugin", Plugins)
 def test_shell_cmd_inputspec_copyfile_1(plugin, results_function, tmpdir):