Merge pull request #591

Add unit tests for runtime exception handling in flows

Author: drazvan

nemoguardrails/colang/v2_x/runtime/statemachine.py

@@ -951,11 +951,9 @@ def _advance_head_front(state: State, heads: List[FlowHead]) -> List[FlowHead]:
 
         if flow_finished:
             _finish_flow(state, flow_state, head.matching_scores)
-            flow_finished = True
             log.debug("Flow finished: %s with last element", head.flow_state_uid)
         elif flow_aborted:
             _abort_flow(state, flow_state, head.matching_scores)
-            flow_aborted = True
             log.debug("Flow aborted: %s by 'abort' statement", head.flow_state_uid)
 
     # Make sure that all actionable heads still exist in flows, otherwise remove them

tests/v2_x/test_event_mechanics.py

@@ -1452,5 +1452,86 @@ def test_event_number_parameter_comparison():
     )
 
 
+def test_runtime_exception_handling_1():
+    """Test to match events based on numeric parameters."""
+
+    content = """
+    flow a
+      $var = "test" + 3
+      match WaitEvent()
+
+    flow main
+      match StartEvent()
+      when a
+        pass
+      else
+        start UtteranceBotAction(script="success")
+      match Event()
+    """
+    state = run_to_completion(_init_state(content), start_main_flow_event)
+    assert is_data_in_events(
+        state.outgoing_events,
+        [],
+    )
+    state = run_to_completion(
+        state,
+        {
+            "type": "StartEvent",
+        },
+    )
+    assert is_data_in_events(
+        state.outgoing_events,
+        [
+            {
+                "type": "StartUtteranceBotAction",
+                "script": "success",
+            },
+        ],
+    )
+
+
+def test_runtime_exception_handling_2():
+    """Test to match events based on numeric parameters."""
+
+    content = """
+    flow a
+      start UtteranceBotAction(script="test")
+      match StartEvent()
+      $var = "test" + 3
+
+    flow main
+      activate a
+      match Event()
+    """
+    state = run_to_completion(_init_state(content), start_main_flow_event)
+    assert is_data_in_events(
+        state.outgoing_events,
+        [
+            {
+                "type": "StartUtteranceBotAction",
+                "script": "test",
+            },
+        ],
+    )
+    state = run_to_completion(
+        state,
+        {
+            "type": "StartEvent",
+        },
+    )
+    assert is_data_in_events(
+        state.outgoing_events,
+        [
+            {
+                "type": "StopUtteranceBotAction",
+            },
+            {
+                "type": "StartUtteranceBotAction",
+                "script": "test",
+            },
+        ],
+    )
+
+
 if __name__ == "__main__":
-    test_custom_regex_event_parameter_match()
+    test_runtime_exception_handling_2()