adding a test that propagates splitter two multiple tasks - small temporary fix to the state class

Author: djarecka

pydra/engine/state.py

@@ -136,6 +136,7 @@ def splitter_final(self):
 
     @property
     def splitter_rpn(self):
+        # TODO NOW: this also changes other_states (properly, but should be explicite)
         _splitter_rpn = hlpst.splitter2rpn(
             deepcopy(self.splitter), other_states=self.other_states
         )
@@ -273,6 +274,7 @@ def update_connections(self, new_other_states=None, new_combiner=None):
         if new_other_states:
             self.other_states = new_other_states
         self._connect_splitters()
+        self.splitter_rpn  # TODO: temporary fix
         if new_combiner:
             self.combiner = new_combiner
         self.set_input_groups()

pydra/engine/tests/test_workflow.py

@@ -829,6 +829,33 @@ def test_wf_ndst_9(plugin):
     assert wf.output_dir.exists()
 
 
+# workflows with structures A ->  B -> C
+
+
+@pytest.mark.parametrize("plugin", Plugins)
+def test_wf_3sernd_ndst_1(plugin):
+    """ workflow with three "serial" tasks, checking if the splitter is propagating"""
+    wf = Workflow(name="wf_3sernd_ndst_1", input_spec=["x", "y"])
+    wf.add(multiply(name="mult", x=wf.lzin.x, y=wf.lzin.y).split(["x", "y"]))
+    wf.add(add2(name="add2_1st", x=wf.mult.lzout.out))
+    wf.add(add2(name="add2_2nd", x=wf.add2_1st.lzout.out))
+    wf.inputs.x = [1, 2]
+    wf.inputs.y = [11, 12]
+    wf.set_output([("out", wf.add2_2nd.lzout.out)])
+    wf.plugin = plugin
+
+    with Submitter(plugin=plugin) as sub:
+        sub(wf)
+
+    results = wf.result()
+    assert results.output.out[0] == 13
+    assert results.output.out[1] == 14
+    assert results.output.out[2] == 24
+    assert results.output.out[3] == 26
+    # checking the output directory
+    assert wf.output_dir.exists()
+
+
 # workflows with structures A -> C, B -> C