update arg parser, implement validation checks, add tests

rasa/cli/data.py

@@ -144,6 +144,22 @@ def _add_data_validate_parsers(
     )
     arguments.set_validator_arguments(story_structure_parser)
 
+    flows_structure_parser = validate_subparsers.add_parser(
+        "flows",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        parents=parents,
+        help="Checks for inconsistencies in the flows files.",
+    )
+    flows_structure_parser.set_defaults(
+        func=lambda args: rasa.cli.utils.validate_files(
+            args.fail_on_warnings,
+            args.max_history,
+            _build_training_data_importer(args),
+            flows_only=True,
+        )
+    )
+    arguments.set_validator_arguments(flows_structure_parser)
+
 
 def _build_training_data_importer(args: argparse.Namespace) -> "TrainingDataImporter":
     config = rasa.cli.utils.get_validated_path(

rasa/cli/utils.py

@@ -217,6 +217,7 @@ def validate_files(
     max_history: Optional[int],
     importer: TrainingDataImporter,
     stories_only: bool = False,
+    flows_only: bool = False,
 ) -> None:
     """Validates either the story structure or the entire project.
 
@@ -225,13 +226,16 @@ def validate_files(
         max_history: The max history to use when validating the story structure.
         importer: The `TrainingDataImporter` to use to load the training data.
         stories_only: If `True`, only the story structure is validated.
+        flows_only: If `True`, only the flows are validated.
     """
     from rasa.validator import Validator
 
     validator = Validator.from_importer(importer)
 
     if stories_only:
         all_good = _validate_story_structure(validator, max_history, fail_on_warnings)
+    elif flows_only:
+        all_good = _validate_flows_structure(validator)
     else:
         if importer.get_domain().is_empty():
             rasa.shared.utils.cli.print_error_and_exit(
@@ -243,8 +247,9 @@ def validate_files(
         valid_stories = _validate_story_structure(
             validator, max_history, fail_on_warnings
         )
+        valid_flows = _validate_flows_structure(validator)
 
-        all_good = valid_domain and valid_nlu and valid_stories
+        all_good = valid_domain and valid_nlu and valid_stories and valid_flows
 
     validator.warn_if_config_mandatory_keys_are_not_set()
 
@@ -288,6 +293,10 @@ def _validate_story_structure(
     )
 
 
+def _validate_flows_structure(validator: "Validator") -> bool:
+    return validator.verify_flows_structure()
+
+
 def cancel_cause_not_found(
     current: Optional[Union["Path", Text]],
     parameter: Text,

rasa/shared/core/flows/flow.py

@@ -38,7 +38,7 @@ def __str__(self) -> Text:
 
 
 class UnresolvedFlowStepIdException(RasaException):
-    """Raised when a flow step is referenced but it's id can not be resolved."""
+    """Raised when a flow step is referenced, but its id can not be resolved."""
 
     def __init__(
         self, step_id: Text, flow: Flow, referenced_from: Optional[FlowStep]

rasa/validator.py

@@ -1,12 +1,18 @@
 import logging
+import string
 from collections import defaultdict
 from typing import Set, Text, Optional, Dict, Any, List
 
+from pypred import Predicate
+
 import rasa.core.training.story_conflict
 from rasa.shared.core.flows.flow import (
     ActionFlowStep,
+    BranchFlowStep,
     CollectInformationFlowStep,
     FlowsList,
+    IfFlowLink,
+    SetSlotsFlowStep,
 )
 import rasa.shared.nlu.constants
 from rasa.shared.constants import (
@@ -28,6 +34,7 @@
 from rasa.shared.core.generator import TrainingDataGenerator
 from rasa.shared.core.constants import SlotMappingType, MAPPING_TYPE
 from rasa.shared.core.training_data.structures import StoryGraph
+from rasa.shared.exceptions import RasaException
 from rasa.shared.importers.importer import TrainingDataImporter
 from rasa.shared.nlu.training_data.training_data import TrainingData
 import rasa.shared.utils.io
@@ -483,3 +490,132 @@ def warn_if_config_mandatory_keys_are_not_set(self) -> None:
                 f"'{ASSISTANT_ID_KEY}' mandatory key. Please replace the default "
                 f"placeholder value with a unique identifier."
             )
+
+    def verify_flows_steps_against_domain(self) -> bool:
+        """Checks flows steps' references against the domain file."""
+        all_good = True
+        domain_slot_names = [slot.name for slot in self.domain.slots]
+        for flow in self.flows.underlying_flows:
+            for step in flow.steps:
+                if isinstance(step, CollectInformationFlowStep):
+                    if step.collect_information not in domain_slot_names:
+                        raise RasaException(
+                            f"The slot '{step.collect_information}' is used in the "
+                            f"step '{step.id}' of flow '{flow.name}', but it "
+                            f"is not listed in the domain slots. "
+                            f"You should add it to your domain file!",
+                        )
+
+                elif isinstance(step, SetSlotsFlowStep):
+                    for slot in step.slots:
+                        slot_name = slot["key"]
+                        if slot_name not in domain_slot_names:
+                            raise RasaException(
+                                f"The slot '{slot_name}' is used in the step "
+                                f"'{step.id}' of flow '{flow.name}', but it "
+                                f"is not listed in the domain slots. "
+                                f"You should add it to your domain file!",
+                            )
+
+                elif isinstance(step, ActionFlowStep):
+                    if step.action not in self.domain.action_names_or_texts:
+                        raise RasaException(
+                            f"The action '{step.action}' is used in the step "
+                            f"'{step.id}' of flow '{flow.name}', but it "
+                            f"is not listed in the domain file. "
+                            f"You should add it to your domain file!",
+                        )
+        return all_good
+
+    def verify_unique_flows(self) -> bool:
+        """Checks if all flows have unique names and descriptions."""
+        all_good = True
+
+        flows_mapping: Dict[str, str] = {}
+        punctuation_table = str.maketrans({i: "" for i in string.punctuation})
+
+        for flow in self.flows.underlying_flows:
+            flow_description = flow.description
+            cleaned_description = flow_description.translate(punctuation_table)  # type: ignore[union-attr] # noqa: E501
+            if cleaned_description in flows_mapping.values():
+                raise RasaException(
+                    f"Detected duplicate flow description for flow '{flow.name}'. "
+                    f"Flow descriptions must be unique. "
+                    f"Please make sure that all flows have different descriptions."
+                )
+
+            if flow.name in flows_mapping:
+                raise RasaException(
+                    f"Detected duplicate flow name '{flow.name}'. "
+                    f"Flow names must be unique. "
+                    f"Please make sure that all flows have different names."
+                )
+
+            flows_mapping[flow.name] = cleaned_description
+
+        return all_good
+
+    def verify_predicates(self) -> bool:
+        """Checks that predicates used in branch flow steps or `collect_information` steps are valid."""  # noqa: E501
+        all_good = True
+        for flow in self.flows.underlying_flows:
+            for step in flow.steps:
+                if isinstance(step, BranchFlowStep):
+                    for link in step.next.links:
+                        if isinstance(link, IfFlowLink):
+                            try:
+                                predicate = Predicate(link.condition)
+                            except (TypeError, Exception) as exception:
+                                raise RasaException(
+                                    f"Could not initialize the predicate found "
+                                    f"under step '{step.id}'. Please make sure "
+                                    f"that all predicates are strings."
+                                ) from exception
+
+                            is_valid = predicate.is_valid()
+                            if not is_valid:
+                                raise RasaException(
+                                    f"Detected invalid condition '{link.condition}' "
+                                    f"at step '{step.id}' for flow '{flow.name}'. "
+                                    f"Please make sure that all conditions are valid."
+                                )
+                elif isinstance(step, CollectInformationFlowStep):
+                    predicates = [predicate.if_ for predicate in step.rejections]
+                    for predicate in predicates:
+                        try:
+                            pred = Predicate(predicate)
+                        except (TypeError, Exception) as exception:
+                            raise RasaException(
+                                f"Could not initialize the predicate found under step "
+                                f"'{step.id}'. Please make sure that all predicates "
+                                f"are strings."
+                            ) from exception
+
+                        is_valid = pred.is_valid()
+                        if not is_valid:
+                            raise RasaException(
+                                f"Detected invalid rejection '{predicate}' "
+                                f"at `collect_information` step '{step.id}' "
+                                f"for flow '{flow.name}'. "
+                                f"Please make sure that all conditions are valid."
+                            )
+        return all_good
+
+    def verify_flows_structure(self) -> bool:
+        """Checks if the flows structure is valid."""
+        if self.flows.is_empty():
+            rasa.shared.utils.io.raise_warning(
+                "No flows were found in the data files."
+                "Will not proceed with flow validation.",
+            )
+            return True
+
+        self.flows.validate()
+
+        all_good = (
+            self.verify_flows_steps_against_domain()
+            and self.verify_unique_flows()
+            and self.verify_predicates()
+        )
+
+        return all_good