Repository: nipype/pydra

Issue:
edges not created if input set after `wf.add`
see the xfailing `test_wf_2b`: https://github.com/nipype/pydra/blob/master/pydra/engine/tests/test_workflow.py#L95

compare with `test_wf_2` and `test_wf_2a`

Discussion:

Patch:
diff --git a/pydra/engine/core.py b/pydra/engine/core.py
index f5c8c931..c2f05697 100644
--- a/pydra/engine/core.py
+++ b/pydra/engine/core.py
@@ -462,7 +462,6 @@ class Workflow(TaskBase):
 
         # store output connections
         self._connections = None
-        self.node_names = []
 
     def __getattr__(self, name):
         if name == "lzin":
@@ -493,22 +492,37 @@ class Workflow(TaskBase):
         return self.graph.sorted_nodes
 
     def add(self, task):
+        """adding a task to the workflow"""
         if not is_task(task):
             raise ValueError("Unknown workflow element: {!r}".format(task))
         self.graph.add_nodes(task)
         self.name2obj[task.name] = task
         self._last_added = task
+        self.create_connections(task)
+        logger.debug(f"Added {task}")
+        self.inputs._graph = self.graph_sorted
+        return self
+
+    def create_connections(self, task):
+        """creating connections between tasks"""
         other_states = {}
         for field in dc.fields(task.inputs):
             val = getattr(task.inputs, field.name)
             if isinstance(val, LazyField):
                 # adding an edge to the graph if task id expecting output from a different task
                 if val.name != self.name:
+                    # checking if the connection is already in the graph
+                    if (getattr(self, val.name), task) in self.graph.edges:
+                        continue
                     self.graph.add_edges((getattr(self, val.name), task))
                     logger.debug("Connecting %s to %s", val.name, task.name)
-                if val.name in self.node_names and getattr(self, val.name).state:
-                    # adding a state from the previous task to other_states
-                    other_states[val.name] = (getattr(self, val.name).state, field.name)
+
+                    if getattr(self, val.name).state:
+                        # adding a state from the previous task to other_states
+                        other_states[val.name] = (
+                            getattr(self, val.name).state,
+                            field.name,
+                        )
         # if task has connections state has to be recalculated
         if other_states:
             if hasattr(task, "fut_combiner"):
@@ -517,10 +531,6 @@ class Workflow(TaskBase):
                 )
             else:
                 task.state = state.State(task.name, other_states=other_states)
-        self.node_names.append(task.name)
-        logger.debug(f"Added {task}")
-        self.inputs._graph = self.graph_sorted
-        return self
 
     async def _run(self, submitter=None, **kwargs):
         self.inputs = dc.replace(self.inputs, **kwargs)
@@ -530,7 +540,9 @@ class Workflow(TaskBase):
         result = self.result()
         if result is not None:
             return result
-
+        # creating connections that were defined after adding tasks to the wf
+        for task in self.graph.nodes:
+            self.create_connections(task)
         """
         Concurrent execution scenarios
 


Test patch:
diff --git a/pydra/engine/tests/test_workflow.py b/pydra/engine/tests/test_workflow.py
index 97823fdc..76eb4579 100644
--- a/pydra/engine/tests/test_workflow.py
+++ b/pydra/engine/tests/test_workflow.py
@@ -151,10 +151,6 @@ def test_wf_2a(plugin):
     assert wf.output_dir == odir
 
 
-@pytest.mark.xfail(
-    reason="x=lzout added after calling wf.add(add2_task):"
-    "edge is not created (fix: creating edges in run?)"
-)
 @pytest.mark.parametrize("plugin", Plugins)
 def test_wf_2b(plugin):
     """ workflow with 2 tasks, no splitter