debugged test_environments (except singularity tests)

Author: tclose

pydra/design/shell.py

@@ -216,10 +216,15 @@ def _validate_path_template(self, attribute, value):
                 f"path_template ({value!r}) can only be provided when no default "
                 f"({self.default!r}) is provided"
             )
+        if value and not is_fileset_or_union(self.type):
+            raise ValueError(
+                f"path_template ({value!r}) can only be provided when type is a FileSet, "
+                f"or union thereof, not {self.type!r}"
+            )
 
     @keep_extension.validator
     def _validate_keep_extension(self, attribute, value):
-        if value and self.path_template is not None:
+        if value and self.path_template is None:
             raise ValueError(
                 f"keep_extension ({value!r}) can only be provided when path_template "
                 f"is provided"

pydra/engine/environments.py

@@ -10,7 +10,6 @@
 if ty.TYPE_CHECKING:
     from pydra.engine.core import Task
     from pydra.engine.specs import ShellDef
-    from pydra.design import shell
 
 
 class Environment:
@@ -94,7 +93,7 @@ def bind(self, loc, mode="ro"):
         return f"{loc_abs}:{self.root}{loc_abs}:{mode}"
 
     def _get_bindings(
-        self, definition: "ShellDef", root: str | None = None
+        self, task: "Task", root: str | None = None
     ) -> tuple[dict[str, tuple[str, str]], dict[str, tuple[Path, ...]]]:
         """Return bindings necessary to run task in an alternative root.
 
@@ -111,27 +110,32 @@ def _get_bindings(
         bindings: dict
           Mapping from paths in the host environment to the target environment
         """
+        from pydra.design import shell
+
         bindings: dict[str, tuple[str, str]] = {}
         input_updates: dict[str, tuple[Path, ...]] = {}
         if root is None:
             return bindings
         fld: shell.arg
-        for fld in list_fields(definition):
+        for fld in list_fields(task.definition):
             if TypeParser.contains_type(FileSet, fld.type):
-                fileset: FileSet | None = definition[fld.name]
-                if fileset is None:
+                fileset: FileSet | None = task.inputs[fld.name]
+                if not fileset:
                     continue
                 if not isinstance(fileset, (os.PathLike, FileSet)):
                     raise NotImplementedError(
-                        "Generating environment bindings for nested FileSets is not "
-                        "supported yet"
+                        f"No support for generating bindings for {type(fileset)} types "
+                        f"({fileset})"
                     )
                 copy = fld.copy_mode == FileSet.CopyMode.copy
 
                 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 else "ro")
+                bindings[host_path] = (
+                    env_path,
+                    "rw" if copy or isinstance(fld, shell.outarg) else "ro",
+                )
 
                 # 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
@@ -152,9 +156,7 @@ class Docker(Container):
     def execute(self, task: "Task[ShellDef]") -> dict[str, ty.Any]:
         docker_img = f"{self.image}:{self.tag}"
         # mounting all input locations
-        mounts, input_updates = self._get_bindings(
-            definition=task.definition, root=self.root
-        )
+        mounts, input_updates = self._get_bindings(task=task, root=self.root)
 
         docker_args = [
             "docker",
@@ -193,9 +195,7 @@ class Singularity(Container):
     def execute(self, task: "Task[ShellDef]") -> dict[str, ty.Any]:
         singularity_img = f"{self.image}:{self.tag}"
         # mounting all input locations
-        mounts, input_updates = self._get_bindings(
-            definition=task.definition, root=self.root
-        )
+        mounts, input_updates = self._get_bindings(task=task, root=self.root)
 
         # todo adding xargsy etc
         singularity_args = [

pydra/engine/helpers_file.py

@@ -157,7 +157,7 @@ def template_update_single(
     input_values: dict[str, ty.Any] = None,
     output_dir: Path | None = None,
     spec_type: str = "input",
-):
+) -> Path | None:
     """Update a single template from the input_spec or output_spec
     based on the value from inputs_dict
     (checking the types of the fields, that have "output_file_template)"
@@ -198,9 +198,9 @@ def template_update_single(
     if output_dir and value is not None:
         # should be converted to str, it is also used for input fields that should be str
         if type(value) is list:
-            return [str(output_dir / Path(val).name) for val in value]
+            return [output_dir / val.name for val in value]
         else:
-            return str(output_dir / Path(value).name)
+            return output_dir / value.name
     else:
         return None
 
@@ -243,7 +243,7 @@ def _template_formatting(field, definition, input_values):
         formatted = _string_template_formatting(
             field, template, definition, input_values
         )
-    return formatted
+    return Path(formatted)
 
 
 def _string_template_formatting(field, template, definition, input_values):
@@ -252,6 +252,14 @@ def _string_template_formatting(field, template, definition, input_values):
     inp_fields = re.findall(r"{\w+}", template)
     inp_fields_fl = re.findall(r"{\w+:[0-9.]+f}", template)
     inp_fields += [re.sub(":[0-9.]+f", "", el) for el in inp_fields_fl]
+
+    # FIXME: This would be a better solution, and would allow you to explicitly specify
+    # whether you want to use the extension of the input file or not, by referencing
+    # the "ext" attribute of the input file. However, this would require a change in the
+    # way the element formatting is done
+    #
+    # inp_fields = set(re.findall(r"{(\w+)(?:\.\w+)?(?::[0-9.]+f)?}", template))
+
     if len(inp_fields) == 0:
         return template
 

pydra/engine/specs.py

@@ -521,7 +521,7 @@ def _check_rules(self):
             if is_lazy(value):
                 continue
 
-            if value is attrs.NOTHING:
+            if value is attrs.NOTHING and not getattr(field, "path_template", False):
                 errors.append(f"Mandatory field {field.name!r} is not set")
 
             # Collect alternative fields associated with this field.
@@ -1043,10 +1043,9 @@ def _command_args(
             output_dir = Path.cwd()
         self._check_resolved()
         inputs = attrs_values(self)
-        modified_inputs = template_update(self, output_dir=output_dir)
+        inputs.update(template_update(self, output_dir=output_dir))
         if input_updates:
             inputs.update(input_updates)
-        inputs.update(modified_inputs)
         pos_args = []  # list for (position, command arg)
         positions_provided = []
         for field in list_fields(self):

pydra/engine/tests/test_environments.py

@@ -43,7 +43,7 @@ def newcache(x):
     outputs = shelly(cache_dir=newcache("shelly-exec"))
     assert drop_stderr(env_outputs) == drop_stderr(attrs_values(outputs))
 
-    outputs = shelly(environment=Native())
+    outputs = shelly(environment=Native(), cache_dir=newcache("shelly-call"))
     assert env_outputs == attrs_values(outputs)
 
     with Submitter(cache_dir=newcache("shelly-submitter"), environment=Native()) as sub:
@@ -59,11 +59,11 @@ def test_docker_1(tmp_path):
     def newcache(x):
         makedir(tmp_path, x)
 
-    cmd = ["whoami"]
+    cmd = "whoami"
     docker = Docker(image="busybox")
     Shelly = shell.define(cmd)
     shelly = Shelly()
-    assert shelly.cmdline == " ".join(cmd)
+    assert shelly.cmdline == cmd
 
     shelly_job = Task(
         definition=shelly,
@@ -267,7 +267,7 @@ def newcache(x):
         f.write("hello ")
 
     shelly = shelly_with_input_factory(filename=filename, executable="cat")
-    outputs_dict = docker.execute(shelly)
+    outputs_dict = docker.execute(make_job(shelly, tmp_path, "shelly"))
 
     with Submitter(environment=docker, cache_dir=newcache("shell_sub")) as sub:
         results = sub(shelly)
@@ -340,7 +340,7 @@ def newcache(x):
     outputs = shelly.split(file=filename)(
         environment=docker, cache_dir=newcache("shelly_call")
     )
-    assert [s.strip for s in outputs.stdout] == ["hello", "hi"]
+    assert [s.strip() for s in outputs.stdout] == ["hello", "hi"]
 
 
 def shelly_outputfile_factory(filename, executable="cp"):
@@ -351,16 +351,20 @@ def shelly_outputfile_factory(filename, executable="cp"):
             shell.arg(
                 name="file_orig",
                 type=File,
-                position=2,
+                position=1,
                 help="new file",
                 argstr="",
             ),
-            shell.arg(
+        ],
+        outputs=[
+            shell.outarg(
                 name="file_copy",
-                type=str,
-                output_file_template="{file_orig}_copy",
+                type=File,
+                path_template="{file_orig}_copy",
                 help="output file",
                 argstr="",
+                position=2,
+                keep_extension=True,
             ),
         ],
     )
@@ -390,11 +394,10 @@ def newcache(x):
         result = sub(shelly)
     assert Path(result.outputs.file_copy) == result.output_dir / "file_copy.txt"
 
-    outputs = shelly(environment=Native(), cache_dir=newcache("shelly_call"))
-    assert (
-        Path(outputs.file_copy)
-        == newcache("shelly_call") / shelly._checksum / "file_copy.txt"
-    )
+    call_cache = newcache("shelly_call")
+
+    outputs = shelly(environment=Native(), cache_dir=call_cache)
+    assert Path(outputs.file_copy) == call_cache / shelly._checksum / "file_copy.txt"
 
 
 def test_shell_fileout_st(tmp_path):
@@ -414,34 +417,21 @@ def newcache(x):
 
     filename = [filename_1, filename_2]
 
-    shelly = shelly_outputfile_factory(
-        tempdir=tmp_path, filename=None, name="shelly_sub"
-    )
+    shelly = shelly_outputfile_factory(filename=None)
     with Submitter(environment=Native(), cache_dir=newcache("shelly")) as sub:
         results = sub(shelly.split(file_orig=filename))
 
-    assert results.outputs.file_copy == [
-        File(results.output_dir / "file_1_copy.txt"),
-        File(results.output_dir / "file_2_copy.txt"),
+    assert [f.name for f in results.outputs.file_copy] == [
+        "file_1_copy.txt",
+        "file_2_copy.txt",
     ]
 
     call_cache = newcache("shelly_call")
 
     outputs = shelly.split(file_orig=filename)(
         environment=Native(), cache_dir=call_cache
     )
-    assert outputs.file_copy == [
-        File(
-            call_cache
-            / attrs.evolve(shelly, file_orig=filename_1)._checksum
-            / "file_1_copy.txt"
-        ),
-        File(
-            call_cache
-            / attrs.evolve(shelly, file_orig=filename_1)._checksum
-            / "file_1_copy.txt"
-        ),
-    ]
+    assert [f.name for f in outputs.file_copy] == ["file_1_copy.txt", "file_2_copy.txt"]
 
 
 @no_win
@@ -459,15 +449,11 @@ def newcache(x):
     with open(filename, "w") as f:
         f.write("hello ")
 
-    shelly_sub = shelly_outputfile_factory(
-        tempdir=tmp_path, filename=filename, name="shelly_sub"
-    )
-    shelly_sub.environment = docker_env
-    shelly_sub()
-    assert (
-        Path(shelly_sub.result().output.file_copy)
-        == shelly_sub.output_dir / "file_copy.txt"
-    )
+    shelly = shelly_outputfile_factory(filename=filename)
+
+    with Submitter(environment=docker_env, cache_dir=newcache("shelly")) as sub:
+        results = sub(shelly)
+    assert results.outputs.file_copy == File(results.output_dir / "file_copy.txt")
 
 
 @no_win
@@ -495,15 +481,7 @@ def newcache(x):
 
     with Submitter(environment=docker_env, cache_dir=newcache("shelly_sub")) as sub:
         results = sub(shelly.split(file_orig=filename))
-    assert results.outputs.file_copy == [
-        File(
-            results.output_dir
-            / attrs.evolve(shelly, file_orig=filename_1)._checksum
-            / "file_1_copy.txt"
-        ),
-        File(
-            results.output_dir
-            / attrs.evolve(shelly, file_orig=filename_2)._checksum
-            / "file_2_copy.txt"
-        ),
+    assert [f.name for f in results.outputs.file_copy] == [
+        "file_1_copy.txt",
+        "file_2_copy.txt",
     ]

pydra/utils/typing.py

@@ -1056,10 +1056,29 @@ def non_optional_type(type_: type) -> type:
     return type_
 
 
-def is_fileset_or_union(type_: type) -> bool:
-    """Check if the type is a FileSet or a Union containing a FileSet"""
+def is_fileset_or_union(type_: type, allow_none: bool | None = None) -> bool:
+    """Check if the type is a FileSet or a Union containing a FileSet
+
+    Parameters
+    ----------
+    type_ : type
+        the type to check
+    allow_none : bool, optional
+        whether to allow None as a valid type, by default None. If None, then None
+        is not allowed at the outer layer, but is allowed within a Union
+
+    Returns
+    -------
+    is_fileset : bool
+        whether the type is a FileSet or a Union containing a FileSet
+    """
+    if type_ is None and allow_none:
+        return True
     if is_union(type_):
-        return any(is_fileset_or_union(t) for t in ty.get_args(type_))
+        return any(
+            is_fileset_or_union(t, allow_none=allow_none or allow_none is None)
+            for t in ty.get_args(type_)
+        )
     elif not inspect.isclass(type_):
         return False
     return issubclass(type_, core.FileSet)