debugging unittests

Author: tclose

pydra/design/tests/test_shell.py

@@ -309,7 +309,6 @@ def test_interface_template_with_overrides_and_optionals():
             type=tuple[int, str],
             sep=" ",
             position=5,
-            sep=" ",
         ),
     ] + outargs + [ShellDef.additional_args]
     assert sorted_fields(Cp.Outputs) == outargs + [

pydra/engine/helpers_file.py

@@ -12,6 +12,9 @@
 from fileformats.generic import FileSet
 from pydra.engine.helpers import is_lazy, attrs_values, list_fields
 
+if ty.TYPE_CHECKING:
+    from pydra.engine.specs import TaskDef
+    from pydra.design import shell
 
 logger = logging.getLogger("pydra")
 
@@ -235,18 +238,23 @@ def _template_formatting(field, definition, input_values):
     # as default, we assume that keep_extension is True
     if isinstance(template, (tuple, list)):
         formatted = [
-            _string_template_formatting(field, t, definition, input_values)
+            _single_template_formatting(field, t, definition, input_values)
             for t in template
         ]
     else:
         assert isinstance(template, str)
-        formatted = _string_template_formatting(
+        formatted = _single_template_formatting(
             field, template, definition, input_values
         )
-    return Path(formatted)
+    return formatted
 
 
-def _string_template_formatting(field, template, definition, input_values):
+def _single_template_formatting(
+    field: "shell.outarg",
+    template: str,
+    definition: "TaskDef",
+    input_values: dict[str, ty.Any],
+) -> Path | None:
     from pydra.utils.typing import MultiInputObj, MultiOutputFile
 
     inp_fields = re.findall(r"{\w+}", template)
@@ -328,7 +336,7 @@ def _string_template_formatting(field, template, definition, input_values):
         formatted_value = _element_formatting(
             template, val_dict, file_template, keep_extension=field.keep_extension
         )
-    return formatted_value
+    return Path(formatted_value) if formatted_value is not None else formatted_value
 
 
 def _element_formatting(

pydra/engine/tests/test_helpers_file.py

@@ -402,4 +402,4 @@ def test_template_formatting(tmp_path: Path):
         input_values=inputs_dict,
         output_dir=tmp_path,
         spec_type="input",
-    ) == [str(tmp_path / "file.bvec"), str(tmp_path / "file.bval")]
+    ) == [tmp_path / "file.bvec", tmp_path / "file.bval"]

pydra/engine/tests/test_nipype1_convert.py

@@ -77,7 +77,7 @@ def test_task_init_3(
     if input_type == "array":
         a_in = np.array(a_in)
 
-    nn = FunAddTwo(name="NA").split(splitter=splitter, a=a_in)
+    nn = FunAddTwo().split(splitter=splitter, a=a_in)
 
     assert np.allclose(nn.inputs.a, [3, 5])
     assert nn.state.splitter == state_splitter

pydra/engine/tests/test_node_task.py

@@ -16,44 +16,41 @@
 
 
 @python.define(outputs=["b"])
-def arrayout(val):
+def ArrayOut(val):
     return np.array([val, val])
 
 
 def test_multiout(tmpdir):
     """testing a simple function that returns a numpy array"""
-    wf = Workflow("wf", input_spec=["val"], val=2)
-    wf.add(arrayout(name="mo", val=wf.lzin.val))
 
-    wf.set_output([("array", wf.mo.lzout.b)])
-    wf.cache_dir = tmpdir
+    @workflow.define(outputs=["array"])
+    def Workflow(val):
+        mo = workflow.add(ArrayOut(val=val))
+        return mo.b
 
-    with Submitter(worker="cf", n_procs=2) as sub:
-        sub(runnable=wf)
+    wf = Workflow(val=2)
 
-    results = wf.result(return_inputs=True)
+    with Submitter(worker="cf", cache_dir=tmpdir, n_procs=2) as sub:
+        results = sub(wf)
 
-    assert results[0] == {"wf.val": 2}
-    assert np.array_equal(results[1].output.array, np.array([2, 2]))
+    assert np.array_equal(results.outputs.array, np.array([2, 2]))
 
 
 def test_multiout_st(tmpdir):
     """testing a simple function that returns a numpy array, adding splitter"""
-    wf = Workflow("wf", input_spec=["val"], val=[0, 1, 2])
-    wf.add(arrayout(name="mo"))
-    wf.mo.split("val", val=wf.lzin.val).combine("val")
 
-    wf.set_output([("array", wf.mo.lzout.b)])
-    wf.cache_dir = tmpdir
+    @workflow.define(outputs=["array"])
+    def Workflow(values):
+        mo = workflow.add(ArrayOut().split(val=values).combine("val"))
+        return mo.b
 
-    with Submitter(worker="cf", n_procs=2) as sub:
-        sub(runnable=wf)
+    wf = Workflow(values=[0, 1, 2])
 
-    results = wf.result(return_inputs=True)
+    with Submitter(worker="cf", cache_dir=tmpdir, n_procs=2) as sub:
+        results = sub(wf)
 
-    assert results[0] == {"wf.val": [0, 1, 2]}
     for el in range(3):
-        assert np.array_equal(results[1].output.array[el], np.array([el, el]))
+        assert np.array_equal(results.outputs.array[el], np.array([el, el]))
 
 
 def test_numpy_hash_1():
@@ -81,20 +78,19 @@ def test_numpy_hash_3():
 
 def test_task_numpyinput_1(tmp_path: Path):
     """task with numeric numpy array as an input"""
-    nn = Identity(name="NA")
-    nn.cache_dir = tmp_path
-    nn.split(x=[np.array([1, 2]), np.array([3, 4])])
+    nn = Identity().split(x=[np.array([1, 2]), np.array([3, 4])])
     # checking the results
-    results = nn()
-    assert (results[0].output.out == np.array([1, 2])).all()
-    assert (results[1].output.out == np.array([3, 4])).all()
+    outputs = nn(cache_dir=tmp_path)
+    assert (outputs.out[0] == np.array([1, 2])).all()
+    assert (outputs.out[1] == np.array([3, 4])).all()
 
 
 def test_task_numpyinput_2(tmp_path: Path):
     """task with numpy array of type object as an input"""
-    nn = Identity(name="NA")
-    nn.cache_dir = tmp_path
-    nn.split(x=[np.array(["VAL1"], dtype=object), np.array(["VAL2"], dtype=object)])
+    nn = Identity().split(
+        x=[np.array(["VAL1"], dtype=object), np.array(["VAL2"], dtype=object)]
+    )
     # checking the results
-    results = nn()
-    assert (results[0].output.out == np.array(["VAL1"], dtype=object)).all()
+    outputs = nn(cache_dir=tmp_path)
+    assert outputs.out[0] == np.array(["VAL1"], dtype=object)
+    assert outputs.out[1] == np.array(["VAL2"], dtype=object)

pydra/engine/tests/test_numpy_examples.py

@@ -1,64 +1,45 @@
 from ..helpers import load_task
-from pydra.design import python
+from pydra.design import python, workflow
+from pydra.engine.core import Task
+from pydra.engine.submitter import Submitter
 
 import numpy as np
 from pympler import asizeof
 from pytest import approx
 
 
-def generate_list(l):
-    return np.arange(l).tolist()
+def generate_list(n):
+    return np.arange(n).tolist()
 
 
 @python.define
-def show_var(a):
+def ShowVar(a):
     return a
 
 
 def create_wf(size):
-    wf = Workflow(name="wf", input_spec=["x"])
-    wf.split("x", x=generate_list(size))
-    wf.add(show_var(name="show", a=wf.lzin.x))
-    wf.set_output([("out", wf.show.lzout.out)])
-    wf.state.prepare_states(wf.inputs)
-    wf.state.prepare_inputs()
-    return wf
+    @workflow.define
+    def Workflow(x):
+        show = workflow.add(ShowVar(a=x))
+        return show.out
+
+    return Workflow().split(x=generate_list(size))
 
 
 def test_wf_memory():
     """creating two workflow with relatively big splitter: 1000, 2000 and 4000 elements
     testings if the size of workflow grows linearly
     """
 
-    wf_1000 = create_wf(size=1000)
-    wf_1000_mem = asizeof.asizeof(wf_1000)
+    wf_10000 = create_wf(size=10000)
+    wf_10000_mem = asizeof.asizeof(wf_10000)
 
-    wf_2000 = create_wf(size=2000)
-    wf_2000_mem = asizeof.asizeof(wf_2000)
+    wf_20000 = create_wf(size=20000)
+    wf_20000_mem = asizeof.asizeof(wf_20000)
 
-    wf_4000 = create_wf(size=4000)
-    wf_4000_mem = asizeof.asizeof(wf_4000)
+    wf_40000 = create_wf(size=40000)
+    wf_40000_mem = asizeof.asizeof(wf_40000)
     # checking if it's linear with the size of the splitter
     # check print(asizeof.asized(wf_4000, detail=2).format()) in case of problems
-    assert wf_4000_mem / wf_2000_mem == approx(2, 0.05)
-    assert wf_2000_mem / wf_1000_mem == approx(2, 0.05)
-
-
-def test_load_task_memory():
-    """creating two workflow with relatively big splitter: 1000 and 4000 elements
-    testings if load_task for a single element returns tasks of a similar size
-    """
-
-    wf_1000 = create_wf(size=1000)
-    wf_1000_pkl = wf_1000.pickle_task()
-    wf_1000_loaded = load_task(task_pkl=wf_1000_pkl, ind=1)
-    wf_1000_single_mem = asizeof.asizeof(wf_1000_loaded)
-
-    wf_4000 = create_wf(size=4000)
-    wf_4000_pkl = wf_4000.pickle_task()
-    wf_4000_loaded = load_task(task_pkl=wf_4000_pkl, ind=1)
-    wf_4000_single_mem = asizeof.asizeof(wf_4000_loaded)
-
-    # checking if it doesn't change with size of the splitter
-    # check print(asizeof.asized(wf_4000_loaded, detail=2).format()) in case of problems
-    assert wf_1000_single_mem / wf_4000_single_mem == approx(1, 0.05)
+    assert wf_40000_mem / wf_20000_mem == approx(2, 0.05)
+    assert wf_20000_mem / wf_10000_mem == approx(2, 0.05)