Merge branch staging

Author: maxime-fender

tests/domain/execute_flow_from_position/test_go_to_flow.py

@@ -141,6 +141,7 @@ def test_flow_returns_all_go_to_flow_quick_replies_in_the_content_language_when_
     mock_repository.set_filter_use_case_uuids_without_production_version(
         mocker=mocker,
         return_value={
+            execution_environment.use_case_uuid,
             go_to_flow_nodes[0].target_use_case_uuid,
             go_to_flow_nodes[2].target_use_case_uuid,
         },
@@ -182,5 +183,8 @@ def test_flow_returns_all_go_to_flow_quick_replies_in_the_content_language_when_
         last_executed_node_uuid=flow.start_node_uuid,
     )
     mock_repository.assert_filter_use_case_uuids_without_production_version_called_once_with(
-        use_case_uuids={node.target_use_case_uuid for node in go_to_flow_nodes}
+        use_case_uuids={
+            execution_environment.use_case_uuid,
+            *(node.target_use_case_uuid for node in go_to_flow_nodes),
+        }
     )

use_case_executor/domain/flow_execution/quick_replies_rendering.py

@@ -18,6 +18,9 @@
 )
 from use_case_executor.domain.flow.node_data.go_to_use_case_data import GoToUseCaseData
 from use_case_executor.domain.flow.node_data.node_type import NodeType
+from use_case_executor.domain.flow.node_data.target_use_case_data import (
+    TargetUseCaseData,
+)
 from use_case_executor.domain.flow_execution import helpers
 from use_case_executor.domain.flow_execution.execution_environment import (
     ExecutionEnvironment,
@@ -36,7 +39,7 @@ def render(self) -> dict:
 
 
 @dataclasses.dataclass(kw_only=True)
-class QuickReply(BaseQuickReply):
+class GoToFlowQuickReply(BaseQuickReply):
     target_use_case_uuid: UUID
     target_node_uuid: Optional[UUID]
     target_edge_uuid: Optional[UUID]
@@ -62,11 +65,6 @@ def render(self) -> dict:
         }
 
 
-@dataclasses.dataclass(kw_only=True)
-class GoToFlowQuickReply(QuickReply):
-    pass
-
-
 @dataclasses.dataclass(kw_only=True)
 class GoToAdvancedQuickReply(BaseQuickReply):
     target_intent_record_id: int
@@ -85,9 +83,6 @@ def render(self) -> dict:
         }
 
 
-QuickReplyTypes = QuickReply | GoToFlowQuickReply | GoToAdvancedQuickReply
-
-
 def render_answer_node_quick_replies(
     answer_node: Node,
     execution_environment: ExecutionEnvironment,
@@ -128,8 +123,8 @@ def _convert_and_order_answer_node_quick_replies(
     language: str,
     use_case_uuid: UUID,
     has_go_back_quick_reply: bool,
-) -> list[QuickReplyTypes]:
-    quick_replies: list[QuickReplyTypes] = []
+) -> list[BaseQuickReply]:
+    quick_replies: list[BaseQuickReply] = []
     for quick_reply in sorted(
         [
             edge
@@ -146,7 +141,7 @@ def _convert_and_order_answer_node_quick_replies(
         )
         if quick_reply.target_node.type != NodeType.go_to_use_case:
             quick_replies.append(
-                QuickReply(
+                GoToFlowQuickReply(
                     title=title,
                     target_use_case_uuid=use_case_uuid,
                     target_node_uuid=quick_reply.target_node.uuid,
@@ -157,30 +152,18 @@ def _convert_and_order_answer_node_quick_replies(
 
         go_to_use_case_data = quick_reply.target_node.data
         assert isinstance(go_to_use_case_data, GoToUseCaseData)
-        if go_to_use_case_data.target_use_case_uuid is not None:
-            assert go_to_use_case_data.target_intent_record_id is None
-            quick_replies.append(
-                GoToFlowQuickReply(
-                    title=title,
-                    target_use_case_uuid=go_to_use_case_data.target_use_case_uuid,
-                    target_node_uuid=None,
-                    target_edge_uuid=None,
-                )
-            )
-        elif go_to_use_case_data.target_intent_record_id is not None:
-            quick_replies.append(
-                GoToAdvancedQuickReply(
-                    title=title,
-                    target_intent_record_id=go_to_use_case_data.target_intent_record_id,
-                )
+        quick_replies.append(
+            _convert_target_use_case_data(
+                title=title,
+                target_use_case_data=go_to_use_case_data,
+                error_message="the target use case is not filled",
             )
-        else:
-            raise InvalidFlowConfiguration("the target use case is not filled")
+        )
 
     if has_go_back_quick_reply:
         assert answer_node.parent_node_uuid is not None
         quick_replies.append(
-            QuickReply(
+            GoToFlowQuickReply(
                 title=constants.GO_BACK_QUICK_REPLY_TITLE[language],
                 target_use_case_uuid=use_case_uuid,
                 target_node_uuid=answer_node.parent_node_uuid,
@@ -194,54 +177,62 @@ def _convert_and_order_answer_node_quick_replies(
 def _convert_handover_failure_quick_replies(
     handover_failure_quick_replies: list[HandoverFailureQuickReply],
     language: str,
-) -> list[QuickReplyTypes]:
-    quick_replies: list[QuickReplyTypes] = []
+) -> list[BaseQuickReply]:
+    quick_replies: list[BaseQuickReply] = []
     for quick_reply in handover_failure_quick_replies:
         title = helpers.get_translation_value(
             translation_dict=quick_reply.title_translations,
             language=language,
             translation_field="handover failure quick reply title",
         )
-        if quick_reply.target_use_case_uuid is not None:
-            assert quick_reply.target_intent_record_id is None
-            quick_replies.append(
-                GoToFlowQuickReply(
-                    title=title,
-                    target_use_case_uuid=quick_reply.target_use_case_uuid,
-                    target_node_uuid=None,
-                    target_edge_uuid=None,
-                )
-            )
-        elif quick_reply.target_intent_record_id is not None:
-            quick_replies.append(
-                GoToAdvancedQuickReply(
-                    title=title,
-                    target_intent_record_id=quick_reply.target_intent_record_id,
-                )
-            )
-        else:
-            raise InvalidFlowConfiguration(
-                "the use case to send the customer to is not filled for an agent handover quick reply"
+        quick_replies.append(
+            _convert_target_use_case_data(
+                title=title,
+                target_use_case_data=quick_reply,
+                error_message=(
+                    "the use case to send the customer to is not filled for an agent"
+                    " handover quick reply"
+                ),
             )
+        )
 
     return quick_replies
 
 
+def _convert_target_use_case_data(
+    title: str, target_use_case_data: TargetUseCaseData, error_message: str
+) -> BaseQuickReply:
+    if target_use_case_data.target_use_case_uuid is not None:
+        assert target_use_case_data.target_intent_record_id is None
+        return GoToFlowQuickReply(
+            title=title,
+            target_use_case_uuid=target_use_case_data.target_use_case_uuid,
+            target_node_uuid=None,
+            target_edge_uuid=None,
+        )
+    if target_use_case_data.target_intent_record_id is not None:
+        return GoToAdvancedQuickReply(
+            title=title,
+            target_intent_record_id=target_use_case_data.target_intent_record_id,
+        )
+    raise InvalidFlowConfiguration(error_message)
+
+
 @inject
 def _get_target_use_case_uuids_to_keep(
     target_use_case_uuids: set[UUID],
     use_latest: bool,
     repository: Repository = Provide["repository"],
 ) -> set[UUID]:
-    if use_latest is True or not target_use_case_uuids:
+    if use_latest or not target_use_case_uuids:
         return target_use_case_uuids
     return repository.filter_use_case_uuids_without_production_version(
         use_case_uuids=target_use_case_uuids
     )
 
 
 def _render_quick_replies(
-    quick_replies: list[QuickReplyTypes],
+    quick_replies: list[BaseQuickReply],
     use_latest: bool,
 ) -> list[dict]:
     target_use_case_uuids_to_keep = _get_target_use_case_uuids_to_keep(
@@ -258,6 +249,6 @@ def _render_quick_replies(
             if quick_reply.target_use_case_uuid in target_use_case_uuids_to_keep:
                 rendered_quick_replies.append(quick_reply.render())
         else:
-            assert isinstance(quick_reply, (QuickReply, GoToAdvancedQuickReply))
+            assert isinstance(quick_reply, GoToAdvancedQuickReply)
             rendered_quick_replies.append(quick_reply.render())
     return rendered_quick_replies