[Internal][Executor] Add a new error for invalid activate config (#1719)

# Description

Add a new error for invalid activate config.

# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [x] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [x] Title of the pull request is clear and informative.
- [x] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [x] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/contracts/_errors.py

@@ -3,3 +3,7 @@
 
 class FailedToImportModule(UserErrorException):
     pass
+
+
+class FlowDefinitionError(UserErrorException):
+    pass

src/promptflow/promptflow/contracts/flow.py

@@ -12,6 +12,7 @@
 
 import yaml
 
+from promptflow.contracts._errors import FlowDefinitionError
 from promptflow.exceptions import ErrorTarget
 
 from .._constants import LANGUAGE_KEY, FlowLanguage
@@ -213,19 +214,27 @@ class ActivateCondition:
     condition_value: Any
 
     @staticmethod
-    def deserialize(data: dict) -> "ActivateCondition":
+    def deserialize(data: dict, node_name: str = None) -> "ActivateCondition":
         """Deserialize the activate condition from a dict.
 
         :param data: The dict to be deserialized.
         :type data: dict
         :return: The activate condition constructed from the dict.
         :rtype: ~promptflow.contracts.flow.ActivateCondition
         """
-        result = ActivateCondition(
-            condition=InputAssignment.deserialize(data["when"]),
-            condition_value=data["is"],
-        )
-        return result
+        if "when" in data and "is" in data:
+            return ActivateCondition(
+                condition=InputAssignment.deserialize(data["when"]),
+                condition_value=data["is"],
+            )
+        else:
+            raise FlowDefinitionError(
+                message_format=(
+                    "The definition of activate config for node {node_name}"
+                    " is incorrect. Please check your flow yaml and resubmit."
+                ),
+                node_name=node_name if node_name else "",
+            )
 
 
 @dataclass
@@ -320,7 +329,7 @@ def deserialize(data: dict) -> "Node":
         if "type" in data:
             node.type = ToolType(data["type"])
         if "activate" in data:
-            node.activate = ActivateCondition.deserialize(data["activate"])
+            node.activate = ActivateCondition.deserialize(data["activate"], node.name)
         return node
 
 

src/promptflow/tests/executor/e2etests/test_activate.py

@@ -7,12 +7,14 @@
 from promptflow._utils.logger_utils import LogContext
 from promptflow.batch._batch_engine import OUTPUT_FILE_NAME, BatchEngine
 from promptflow.batch._result import BatchResult
+from promptflow.contracts._errors import FlowDefinitionError
 from promptflow.contracts.run_info import FlowRunInfo
 from promptflow.contracts.run_info import RunInfo as NodeRunInfo
 from promptflow.contracts.run_info import Status
 from promptflow.executor import FlowExecutor
 
 from ..utils import (
+    WRONG_FLOW_ROOT,
     MemoryRunStorage,
     get_flow_expected_result,
     get_flow_expected_status_summary,
@@ -75,6 +77,15 @@ def test_all_nodes_bypassed(self, dev_connections):
             content = fin.read()
             assert "The node referenced by output:'third_node' is bypassed, which is not recommended." in content
 
+    def test_invalid_activate_config(self):
+        flow_folder = "invalid_activate_config"
+        with pytest.raises(FlowDefinitionError) as ex:
+            FlowExecutor.create(get_yaml_file(flow_folder, root=WRONG_FLOW_ROOT), {})
+        assert ex.value.message == (
+            "The definition of activate config for node divide_num is incorrect. "
+            "Please check your flow yaml and resubmit."
+        )
+
     def test_aggregate_bypassed_nodes(self):
         flow_folder = "conditional_flow_with_aggregate_bypassed"
         mem_run_storage = MemoryRunStorage()

src/promptflow/tests/executor/unittests/executor/test_dag_manager.py

@@ -6,7 +6,7 @@
 
 def create_test_node(name, input, activate=None):
     input = InputAssignment.deserialize(input)
-    activate = ActivateCondition.deserialize(activate) if activate else None
+    activate = ActivateCondition.deserialize(activate, name) if activate else None
     return Node(
         name=name,
         tool="test_tool",

src/promptflow/tests/test_configs/wrong_flows/invalid_activate_config/flow.dag.yaml

@@ -0,0 +1,17 @@
+inputs:
+  num:
+    type: int
+outputs:
+  content:
+    type: string
+    reference: ${divide_num.output}
+nodes:
+- name: divide_num
+  type: python
+  source:
+    type: code
+    path: divide_num.py
+  inputs:
+    num: ${inputs.num}
+  activate:
+    when: ${inputs.num} > 0