fixed all unittests outside of test_workflow that were broken in recent changes

Author: tclose

pydra/engine/environments.py

@@ -3,10 +3,12 @@
 from copy import copy
 from .helpers import execute
 from pathlib import Path
+import logging
 from fileformats.generic import FileSet
 from pydra.engine.helpers import list_fields
 from pydra.utils.typing import TypeParser
 
+logger = logging.getLogger("pydra")
 
 if ty.TYPE_CHECKING:
     from pydra.engine.core import Task
@@ -121,34 +123,50 @@ def get_bindings(
         fld: shell.arg
         for fld in list_fields(task.definition):
             if TypeParser.contains_type(FileSet, fld.type):
-                fileset: FileSet | None = task.inputs[fld.name]
-                if not fileset:
+                value: FileSet | None = task.inputs[fld.name]
+                if not value:
                     continue
-                if not isinstance(fileset, (os.PathLike, FileSet)):
-                    raise NotImplementedError(
-                        f"No support for generating bindings for {type(fileset)} types "
-                        f"({fileset})"
-                    )
+
                 copy_file = fld.copy_mode == FileSet.CopyMode.copy
 
-                host_path, env_path = fileset.parent, Path(f"{root}{fileset.parent}")
+                def map_path(fileset: os.PathLike | FileSet) -> Path:
+                    host_path, env_path = fileset.parent, Path(
+                        f"{root}{fileset.parent}"
+                    )
 
-                # Default to mounting paths as read-only, but respect existing modes
-                bindings[host_path] = (
-                    env_path,
-                    "rw" if copy_file or isinstance(fld, shell.outarg) else "ro",
-                )
+                    # Default to mounting paths as read-only, but respect existing modes
+                    bindings[host_path] = (
+                        env_path,
+                        "rw" if copy_file or isinstance(fld, shell.outarg) else "ro",
+                    )
+                    return (
+                        env_path / fileset.name
+                        if isinstance(fileset, os.PathLike)
+                        else tuple(env_path / rel for rel in fileset.relative_fspaths)
+                    )
 
                 # Provide updated in-container paths to the command to be run. If a
                 # fs-object, which resolves to a single path, just pass in the name of
                 # that path relative to the location in the mount point in the container.
                 # If it is a more complex file-set with multiple paths, then it is converted
                 # into a tuple of paths relative to the base of the fileset.
-                value_updates[fld.name] = (
-                    env_path / fileset.name
-                    if isinstance(fileset, os.PathLike)
-                    else tuple(env_path / rel for rel in fileset.relative_fspaths)
-                )
+                if TypeParser.matches(value, os.PathLike | FileSet):
+                    value_updates[fld.name] = map_path(value)
+                elif TypeParser.matches(value, ty.Sequence[FileSet | os.PathLike]):
+                    mapped_value = []
+                    for val in value:
+                        mapped_val = map_path(val)
+                        if isinstance(mapped_val, tuple):
+                            mapped_value.extend(mapped_val)
+                        else:
+                            mapped_value.append(mapped_val)
+                    value_updates[fld.name] = mapped_value
+                else:
+                    logger.debug(
+                        "No support for generating bindings for %s types " "(%s)",
+                        type(value),
+                        value,
+                    )
 
         # Add the cache directory to the list of mounts
         bindings[task.cache_dir] = (f"{self.root}/{task.cache_dir}", "rw")

pydra/engine/helpers.py

@@ -536,7 +536,12 @@ def load_and_run(task_pkl: Path, rerun: bool = False) -> Path:
 
     resultfile = task.output_dir / "_result.pklz"
     try:
-        task.run(rerun=rerun)
+        if task.submitter.worker.is_async:
+            task.submitter.loop.run_until_complete(
+                task.submitter.worker.run_async(task, rerun=rerun)
+            )
+        else:
+            task.submitter.worker.run(task, rerun=rerun)
     except Exception as e:
         # creating result and error files if missing
         errorfile = task.output_dir / "_error.pklz"

pydra/engine/submitter.py

@@ -725,7 +725,7 @@ def _split_definition(self) -> dict[int, "TaskDef[OutputType]"]:
                     resolved[inpt_name] = value._get_value(
                         workflow=self.workflow,
                         graph=self.graph,
-                        state_index=input_ind[state_key],
+                        state_index=input_ind.get(state_key),
                     )
                 elif state_key in input_ind:
                     resolved[inpt_name] = self.node.state._get_element(
@@ -786,7 +786,7 @@ def get_runnable_tasks(self, graph: DiGraph) -> list["Task[DefType]"]:
             if is_runnable:
                 runnable.append(self.blocked.pop(index))
         self.queued.update({t.state_index: t for t in runnable})
-        return runnable
+        return list(self.queued.values())
 
 
 async def prepare_runnable(runnable):

pydra/engine/tests/test_helpers.py

@@ -230,7 +230,8 @@ def test_load_and_run_exception_run(tmpdir):
     assert result_1.outputs.out == 2
 
 
-def test_load_and_run_wf(tmpdir):
+@pytest.mark.parametrize("worker", ["cf", "debug"])
+def test_load_and_run_wf(tmpdir, worker):
     """testing load_and_run for pickled task"""
     wf_pkl = Path(tmpdir.join("wf_main.pkl"))
 
@@ -242,7 +243,7 @@ def Workflow(x, y=10):
     task = Task(
         name="mult",
         definition=Workflow(x=2),
-        submitter=Submitter(cache_dir=tmpdir, worker="cf"),
+        submitter=Submitter(cache_dir=tmpdir, worker=worker),
     )
 
     with wf_pkl.open("wb") as fp:

pydra/engine/tests/test_singularity.py

@@ -89,7 +89,7 @@ def test_singularity_2a(plugin, tmp_path):
     assert singu.cmdline == f"{cmd_exec} {' '.join(cmd_args)}"
 
     with Submitter(
-        worker=plugin, environment=Singularity(image=image), cache_dir=tmp_path
+        worker="debug", environment=Singularity(image=image), cache_dir=tmp_path
     ) as sub:
         res = sub(singu)
 

pydra/engine/tests/test_specs.py

@@ -97,12 +97,12 @@ def test_input_file_hash_1(tmp_path):
     def A(in_file: File) -> File:
         return in_file
 
-    assert A(in_file=outfile)._hash == "9644d3998748b339819c23ec6abec520"
+    assert A(in_file=outfile)._hash == "e708da65b720212c5ce9ed2c65aae59c"
 
     with open(outfile, "w") as fp:
         fp.write("test")
 
-    assert A(in_file=outfile)._hash == "9f7f9377ddef6d8c018f1bf8e89c208c"
+    assert A(in_file=outfile)._hash == "f726a193430352bb3b92dccf5eccff3a"
 
 
 def test_input_file_hash_2(tmp_path):
@@ -117,7 +117,7 @@ def A(in_file: File) -> File:
 
     # checking specific hash value
     hash1 = A(in_file=file)._hash
-    assert hash1 == "179bd3cbdc747edc4957579376fe8c7d"
+    assert hash1 == "eba2fafb8df4bae94a7aa42bb159b778"
 
     # checking if different name doesn't affect the hash
     file_diffname = tmp_path / "in_file_2.txt"
@@ -146,7 +146,7 @@ def A(in_file: ty.Union[File, int]) -> File:
 
     # checking specific hash value
     hash1 = A(in_file=file)._hash
-    assert hash1 == "179bd3cbdc747edc4957579376fe8c7d"
+    assert hash1 == "eba2fafb8df4bae94a7aa42bb159b778"
 
     # checking if different name doesn't affect the hash
     file_diffname = tmp_path / "in_file_2.txt"
@@ -234,7 +234,7 @@ def A(in_file: ty.List[ty.List[ty.Union[int, File]]]) -> File:
 
     # checking specific hash value
     hash1 = A(in_file=[[file, 3]])._hash
-    assert hash1 == "ffd7afe0ca9d4585518809a509244b4b"
+    assert hash1 == "b583e0fd5501d3bed9bf510ce2a9e379"
 
     # the same file, but int field changes
     hash1a = A(in_file=[[file, 5]])._hash
@@ -268,7 +268,7 @@ def A(in_file: ty.List[ty.Dict[ty.Any, ty.Union[File, int]]]) -> File:
 
     # checking specific hash value
     hash1 = A(in_file=[{"file": file, "int": 3}])._hash
-    assert hash1 == "ba884a74e33552854271f55b03e53947"
+    assert hash1 == "aa2d4b708ed0dd8340582a6514bfd5ce"
 
     # the same file, but int field changes
     hash1a = A(in_file=[{"file": file, "int": 5}])._hash