Raise useful error on workflow.connect_with for wrong labels (#2110)

* Typehints

* Fix typo in error

* Raise connection error on connect_with for wrong names

* Add Workflow.safe_connect_with

* Add test for Workflow.safe_connect_with

* Switch to adding parameter 'permissive' to Workflow.connect_with()

Author: PProfizi

src/ansys/dpf/core/workflow.py

@@ -22,6 +22,8 @@
 
 """Workflow."""
 
+from __future__ import annotations
+
 from enum import Enum
 import logging
 import os
@@ -714,7 +716,12 @@ def output_names(self):
         return out
 
     @version_requires("3.0")
-    def connect_with(self, left_workflow, output_input_names=None):
+    def connect_with(
+        self,
+        left_workflow: Workflow,
+        output_input_names: Union[tuple[str, str], dict[str, str]] = None,
+        permissive: bool = True,
+    ):
         """Prepend a given workflow to the current workflow.
 
         Updates the current workflow to include all the operators of the workflow given as argument.
@@ -724,15 +731,18 @@ def connect_with(self, left_workflow, output_input_names=None):
 
         Parameters
         ----------
-        left_workflow : core.Workflow
+        left_workflow:
             The given workflow's outputs are chained with the current workflow's inputs.
-        output_input_names : str tuple, str dict optional
+        output_input_names:
             Map used to connect the outputs of the given workflow to the inputs of the current
             workflow.
             Check the names of available inputs and outputs for each workflow using
             `Workflow.input_names` and `Workflow.output_names`.
             The default is ``None``, in which case it tries to connect each output of the
             left_workflow with an input of the current workflow with the same name.
+        permissive:
+            Whether to filter 'output_input_names' to only keep available connections.
+            Otherwise raise an error if 'output_input_names' contains unavailable inputs or outputs.
 
         Examples
         --------
@@ -791,24 +801,40 @@ def connect_with(self, left_workflow, output_input_names=None):
 
         """
         if output_input_names:
-            core_api = self._server.get_api_for_type(
-                capi=data_processing_capi.DataProcessingCAPI,
-                grpcapi=data_processing_grpcapi.DataProcessingGRPCAPI,
-            )
-            map = object_handler.ObjHandler(
-                data_processing_api=core_api,
-                internal_obj=self._api.workflow_create_connection_map_for_object(self),
-            )
             if isinstance(output_input_names, tuple):
-                self._api.workflow_add_entry_connection_map(
-                    map, output_input_names[0], output_input_names[1]
+                output_input_names = {output_input_names[0]: output_input_names[1]}
+            if isinstance(output_input_names, dict):
+                core_api = self._server.get_api_for_type(
+                    capi=data_processing_capi.DataProcessingCAPI,
+                    grpcapi=data_processing_grpcapi.DataProcessingGRPCAPI,
+                )
+                map = object_handler.ObjHandler(
+                    data_processing_api=core_api,
+                    internal_obj=self._api.workflow_create_connection_map_for_object(self),
                 )
-            elif isinstance(output_input_names, dict):
-                for key in output_input_names:
-                    self._api.workflow_add_entry_connection_map(map, key, output_input_names[key])
+                output_names = left_workflow.output_names
+                input_names = self.input_names
+                if permissive:
+                    output_input_names = dict(
+                        filter(
+                            lambda item: item[0] in left_workflow.output_names
+                            and item[1] in self.input_names,
+                            output_input_names.items(),
+                        )
+                    )
+                for output_name, input_name in output_input_names.items():
+                    if output_name not in output_names:
+                        raise ValueError(
+                            f"Cannot connect workflow output '{output_name}'. Exposed outputs are:\n{output_names}"
+                        )
+                    elif input_name not in input_names:
+                        raise ValueError(
+                            f"Cannot connect workflow input '{input_name}'. Exposed inputs are:\n{input_names}"
+                        )
+                    self._api.workflow_add_entry_connection_map(map, output_name, input_name)
             else:
                 raise TypeError(
-                    "output_input_names argument is expect" "to be either a str tuple or a str dict"
+                    "output_input_names argument is expected to be either a str tuple or a str dict"
                 )
             self._api.work_flow_connect_with_specified_names(self, left_workflow, map)
         else:

tests/test_workflow.py

@@ -695,6 +695,42 @@ def test_connect_with_dict_workflow(cyclic_lin_rst, cyclic_ds, server_type):
     fc = wf2.get_output("u", dpf.core.types.fields_container)
 
 
+def test_workflow_connect_raise_wrong_label(server_type):
+    workflow1 = dpf.core.Workflow()
+    forward_1 = dpf.core.operators.utility.forward()
+    workflow1.set_output_name("output", forward_1.outputs.any)
+
+    workflow2 = dpf.core.Workflow()
+    forward_2 = dpf.core.operators.utility.forward()
+    workflow2.set_input_name("input", forward_2.inputs.any)
+
+    with pytest.raises(
+        ValueError, match="Cannot connect workflow output 'out'. Exposed outputs are:\n"
+    ):
+        workflow2.connect_with(workflow1, output_input_names={"out": "input"}, permissive=False)
+    with pytest.raises(
+        ValueError, match="Cannot connect workflow input 'in'. Exposed inputs are:\n"
+    ):
+        workflow2.connect_with(workflow1, output_input_names={"output": "in"}, permissive=False)
+    workflow2.connect_with(workflow1, output_input_names={"output": "input"}, permissive=False)
+
+
+def test_workflow_connect_with_permissive(server_type):
+    workflow1 = dpf.core.Workflow()
+    forward_1 = dpf.core.operators.utility.forward()
+    workflow1.set_output_name("output", forward_1.outputs.any)
+
+    workflow2 = dpf.core.Workflow()
+    forward_2 = dpf.core.operators.utility.forward()
+    workflow2.set_input_name("input", forward_2.inputs.any)
+
+    workflow2.connect_with(workflow1, output_input_names={"out": "input"})
+
+    workflow2.connect_with(workflow1, output_input_names={"output": "in"})
+
+    workflow2.connect_with(workflow1, output_input_names=("output", "input"))
+
+
 @pytest.mark.xfail(raises=dpf.core.errors.ServerTypeError)
 def test_info_workflow(allkindofcomplexity, server_type):
     model = dpf.core.Model(allkindofcomplexity, server=server_type)