debugging test_tasks

Author: tclose

new-docs/source/tutorial/2-advanced-execution.ipynb

@@ -393,6 +393,15 @@
     "Work in progress..."
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Hooks\n",
+    "\n",
+    "Work in progress..."
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},

new-docs/source/tutorial/6-workflow.ipynb

@@ -133,7 +133,7 @@
     "def SplitWorkflow(a: list[int], b: list[float]) -> list[float]:\n",
     "    # Multiply over all combinations of the elements of a and b, then combine the results\n",
     "    # for each a element into a list over each b element\n",
-    "    mul = workflow.add(Mul()).split(x=a, y=b).combine(\"x\")\n",
+    "    mul = workflow.add(Mul().split(x=a, y=b).combine(\"x\"))\n",
     "    # Sume the multiplications across all all b elements for each a element\n",
     "    sum = workflow.add(Sum(x=mul.out))\n",
     "    return sum.out"
@@ -154,8 +154,8 @@
    "source": [
     "@workflow.define\n",
     "def SplitThenCombineWorkflow(a: list[int], b: list[float], c: float) -> list[float]:\n",
-    "    mul = workflow.add(Mul()).split(x=a, y=b)\n",
-    "    add = workflow.add(Add(x=mul.out, y=c)).combine(\"Mul.x\")\n",
+    "    mul = workflow.add(Mul().split(x=a, y=b))\n",
+    "    add = workflow.add(Add(x=mul.out, y=c).combine(\"Mul.x\"))\n",
     "    sum = workflow.add(Sum(x=add.out))\n",
     "    return sum.out"
    ]

pydra/design/base.py

@@ -830,10 +830,16 @@ def extract_function_inputs_and_outputs(
         inputs = input_types
     for inpt_name, default in input_defaults.items():
         inpt = inputs[inpt_name]
-        if isinstance(inpt, arg_type) and inpt.default is EMPTY:
-            inpt.default = default
-        else:
+        if isinstance(inpt, arg_type):
+            if inpt.default is EMPTY:
+                inpt.default = default
+        elif inspect.isclass(inpt):
             inputs[inpt_name] = arg_type(type=inpt, default=default)
+        else:
+            raise ValueError(
+                f"Unrecognised input type ({inpt}) for input {inpt_name} with default "
+                f"value {default}"
+            )
     return_type = type_hints.get("return", ty.Any)
     if outputs and len(outputs) > 1:
         if return_type is not ty.Any:

pydra/engine/audit.py

@@ -1,6 +1,7 @@
 """Module to keep track of provenance information."""
 
 import os
+import typing as ty
 import json
 from pydra.utils.messenger import send_message, make_message, gen_uuid, now, AuditFlag
 from pydra.engine.helpers import attrs_values
@@ -12,6 +13,9 @@
 except ImportError:
     import importlib.resources as importlib_resources  # type: ignore
 
+if ty.TYPE_CHECKING:
+    from pydra.engine.task import Task
+
 
 class Audit:
     """Handle provenance tracking and resource utilization."""
@@ -178,17 +182,19 @@ def audit_check(self, flag):
         """
         return self.audit_flags & flag
 
-    def audit_task(self, task):
+    def audit_task(self, task: "Task"):
         import subprocess as sp
-        from .helpers import attrs_fields
+        from .helpers import list_fields
 
         label = task.name
 
-        command = task.cmdline if hasattr(task.inputs, "executable") else None
-        attr_list = attrs_fields(task.inputs)
+        command = (
+            task.definition.cmdline if hasattr(task.definition, "executable") else None
+        )
+        attr_list = list_fields(task.definition)
         for attrs in attr_list:
             input_name = attrs.name
-            value = getattr(task.inputs, input_name)
+            value = task.inputs[input_name]
             if isinstance(value, FileSet):
                 input_path = os.path.abspath(value)
                 file_hash = hash_function(value)

pydra/engine/core.py

@@ -32,7 +32,6 @@
     TaskHook,
 )
 from .helpers import (
-    create_checksum,
     attrs_fields,
     attrs_values,
     load_result,
@@ -203,8 +202,7 @@ def checksum(self):
         """
         if self._checksum is not None:
             return self._checksum
-        input_hash = self.definition._hash
-        self._checksum = create_checksum(self.definition._task_type, input_hash)
+        self._checksum = self.definition._checksum
         return self._checksum
 
     @property

pydra/engine/specs.py

@@ -159,6 +159,7 @@ def __call__(
         audit_flags: AuditFlag = AuditFlag.NONE,
         messengers: ty.Iterable[Messenger] | None = None,
         messenger_args: dict[str, ty.Any] | None = None,
+        name: str | None = None,
         **kwargs: ty.Any,
     ) -> OutputsType:
         """Create a task from this definition and execute it to produce a result.
@@ -183,6 +184,8 @@ def __call__(
             Messengers, by default None
         messenger_args : dict, optional
             Messenger arguments, by default None
+        name : str
+            The name of the task, by default None
         **kwargs : dict
             Keyword arguments to pass on to the worker initialisation
 
@@ -209,7 +212,7 @@ def __call__(
                 worker=worker,
                 **kwargs,
             ) as sub:
-                result = sub(self)
+                result = sub(self, name=name)
         except TypeError as e:
             if hasattr(e, "__notes__") and WORKER_KWARG_FAIL_NOTE in e.__notes__:
                 if match := re.match(
@@ -412,6 +415,10 @@ def _hash(self):
         hsh, self._hashes = self._compute_hashes()
         return hsh
 
+    @property
+    def _checksum(self):
+        return f"{self._task_type}-{self._hash}"
+
     def _hash_changes(self):
         """Detects any changes in the hashed values between the current inputs and the
         previously calculated values"""

pydra/engine/submitter.py

@@ -168,9 +168,13 @@ def worker(self):
     def __call__(
         self,
         task_def: "TaskDef",
+        name: str | None = "task",
     ):
         """Submitter run function."""
 
+        if name is None:
+            name = "task"
+
         task_def._check_rules()
         # If the outer task is split, create an implicit workflow to hold the split nodes
         if task_def._splitter:
@@ -190,10 +194,10 @@ def Split(defn: TaskDef, output_types: dict):
                 f"Task {self} is marked for combining, but not splitting. "
                 "Use the `split` method to split the task before combining."
             )
-        task = Task(task_def, submitter=self, name="task", environment=self.environment)
+        task = Task(task_def, submitter=self, name=name, environment=self.environment)
         try:
             self.run_start_time = datetime.now()
-            if task.is_async:  # Only workflow tasks can be async
+            if self.worker.is_async:  # Only workflow tasks can be async
                 self.loop.run_until_complete(
                     self.worker.run_async(task, rerun=self.rerun)
                 )

pydra/engine/task.py

@@ -149,7 +149,7 @@ def __init__(
         self.inputs_mod_root = {}
 
 
-class BoshTask(ShellTask):
+class BoshTask(ShellDef):
 
     def _command_args_single(self, state_ind=None, index=None):
         """Get command line arguments for a single state"""

pydra/engine/tests/test_dockertask.py

@@ -1,9 +1,10 @@
 import pytest
-from ..task import ShellTask
-from ..submitter import Submitter
+from pydra.engine.specs import ShellDef
+from pydra.engine.submitter import Submitter
 from fileformats.generic import File
-from ..environments import Docker
+from pydra.engine.environments import Docker
 from pydra.design import shell, workflow
+from pydra.engine.core import Task
 from .utils import no_win, need_docker, result_submitter, result_no_submitter
 
 
@@ -14,13 +15,19 @@ def test_docker_1_nosubm():
     no submitter
     """
     cmd = "whoami"
-    docky = shell.define(cmd)(environment=Docker(image="busybox"))
-    assert docky.environment.image == "busybox"
-    assert docky.environment.tag == "latest"
-    assert isinstance(docky.environment, Docker)
-    assert docky.cmdline == cmd
+    Docky = shell.define(cmd)
+    docky = Docky()
+    docky_task = Task(
+        definition=docky,
+        name="docky",
+        submitter=Submitter(environment=Docker(image="busybox")),
+    )
+    assert docky_task.environment.image == "busybox"
+    assert docky_task.environment.tag == "latest"
+    assert isinstance(docky_task.environment, Docker)
+    assert docky_task.cmdline == cmd
 
-    res = docky()
+    res = docky_task()
     assert res.output.stdout == "root\n"
     assert res.output.return_code == 0
 
@@ -32,14 +39,14 @@ def test_docker_1(plugin):
     using submitter
     """
     cmd = "whoami"
-    docky = shell.define(cmd)(environment=Docker(image="busybox"))
+    Docky = shell.define(cmd)
+    docky = Docky()
 
-    with Submitter(worker=plugin) as sub:
-        docky(submitter=sub)
+    with Submitter(environment=Docker(image="busybox")) as sub:
+        res = sub(docky)
 
-    res = docky.result()
-    assert res.output.stdout == "root\n"
-    assert res.output.return_code == 0
+    assert res.outputs.stdout == "root\n"
+    assert res.outputs.return_code == 0
 
 
 @no_win
@@ -50,7 +57,8 @@ def test_docker_2(results_function, plugin):
     with and without submitter
     """
     cmdline = "echo hail pydra"
-    docky = shell.define(cmdline)(environment=Docker(image="busybox"))
+    Docky = shell.define(cmdline)
+    docky = Docky()
     # cmdline doesn't know anything about docker
     assert docky.cmdline == cmdline
     res = results_function(docky, plugin)
@@ -68,13 +76,9 @@ def test_docker_2a(results_function, plugin):
     cmd_exec = "echo"
     cmd_args = ["hail", "pydra"]
     # separate command into exec + args
-    docky = ShellTask(
-        name="docky",
-        executable=cmd_exec,
-        args=cmd_args,
-        environment=Docker(image="busybox"),
-    )
-    assert docky.definition.executable == "echo"
+    Docky = shell.define(" ".join([cmd_exec] + cmd_args))
+    docky = Docky()
+    assert docky.executable == "echo"
     assert docky.cmdline == f"{cmd_exec} {' '.join(cmd_args)}"
 
     res = results_function(docky, plugin)
@@ -93,9 +97,9 @@ def test_docker_st_1(results_function, plugin):
     splitter = executable
     """
     cmd = ["pwd", "whoami"]
-    docky = ShellTask(name="docky", environment=Docker(image="busybox")).split(
-        "executable", executable=cmd
-    )
+    Docky = shell.define("placeholder")
+    docky = Docky().split(executable=cmd)
+
     assert docky.state.splitter == "docky.executable"
 
     res = results_function(docky, plugin)