[flow test] Add flow chat/test/streamlit_chat in telemetry (#975)

# Description

Please add an informative description that covers that changes made by
the pull request and link all relevant issues.

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **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
- [ ] Title of the pull request is clear and informative.
- [ ] 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
- [ ] Pull request includes test coverage for the included changes.

Author: lalala123123

src/promptflow/promptflow/_cli/_pf/_flow.py

@@ -7,7 +7,6 @@
 import json
 import logging
 import os
-import sys
 import tempfile
 import webbrowser
 from pathlib import Path
@@ -39,8 +38,6 @@
 from promptflow._cli._utils import _copy_to_flow, activate_action, confirm, inject_sys_path, list_of_dict_to_dict
 from promptflow._sdk._constants import LOGGER_NAME, PROMPT_FLOW_DIR_NAME, ConnectionProvider
 from promptflow._sdk._pf_client import PFClient
-from promptflow._sdk._utils import dump_flow_result
-from promptflow.exceptions import UserErrorException
 
 DEFAULT_CONNECTION = "open_ai_connection"
 DEFAULT_DEPLOYMENT = "gpt-35-turbo"
@@ -364,8 +361,6 @@ def _init_flow_by_template(flow_name, flow_type, overwrite=False, connection=Non
 @exception_handler("Flow test")
 def test_flow(args):
     from promptflow._sdk._load_functions import load_flow
-    from promptflow._sdk._utils import parse_variant
-    from promptflow._sdk.operations._test_submitter import TestSubmitter
 
     config = list_of_dict_to_dict(args.config)
     pf_client = PFClient(config=config)
@@ -386,14 +381,6 @@ def test_flow(args):
 
     if args.multi_modal or args.ui:
         with tempfile.TemporaryDirectory() as temp_dir:
-            try:
-                import bs4  # noqa: F401
-                import streamlit_quill  # noqa: F401
-                from streamlit.web import cli as st_cli
-            except ImportError as ex:
-                raise UserErrorException(
-                    f"Please try 'pip install promptflow[executable]' to install dependency, {ex.msg}."
-                )
             flow = load_flow(args.flow)
 
             script_path = [
@@ -407,16 +394,8 @@ def test_flow(args):
                     flow_dag_path=flow.flow_dag_path,
                     connection_provider=pf_client._ensure_connection_provider(),
                 ).generate_to_file(script)
-
-            sys.argv = [
-                "streamlit",
-                "run",
-                os.path.join(temp_dir, "main.py"),
-                "--global.developmentMode=false",
-                "--client.toolbarMode=viewer",
-                "--browser.gatherUsageStats=false",
-            ]
-            st_cli.main()
+            main_script_path = os.path.join(temp_dir, "main.py")
+            pf_client.flows._chat_with_ui(script=main_script_path)
     else:
         if args.interactive:
             pf_client.flows._chat(
@@ -427,33 +406,21 @@ def test_flow(args):
                 show_step_output=args.verbose,
             )
         else:
-            result = pf_client.flows._test(
+            result = pf_client.flows.test(
                 flow=args.flow,
                 inputs=inputs,
                 environment_variables=environment_variables,
                 variant=args.variant,
                 node=args.node,
                 allow_generator_output=False,
                 stream_output=False,
+                dump_test_result=True,
             )
-            # Dump flow/node test info
-            flow = load_flow(args.flow)
-            if args.node:
-                dump_flow_result(flow_folder=flow.code, node_result=result, prefix=f"flow-{args.node}.node")
-            else:
-                if args.variant:
-                    tuning_node, node_variant = parse_variant(args.variant)
-                    prefix = f"flow-{tuning_node}-{node_variant}"
-                else:
-                    prefix = "flow"
-                dump_flow_result(flow_folder=flow.code, flow_result=result, prefix=prefix)
-
-            TestSubmitter._raise_error_when_test_failed(result, show_trace=args.node is not None)
             # Print flow/node test result
-            if isinstance(result.output, dict):
-                print(json.dumps(result.output, indent=4, ensure_ascii=False))
+            if isinstance(result, dict):
+                print(json.dumps(result, indent=4, ensure_ascii=False))
             else:
-                print(result.output)
+                print(result)
 
 
 def serve_flow(args):

src/promptflow/promptflow/_sdk/operations/_flow_operations.py

@@ -6,6 +6,7 @@
 import json
 import os
 import subprocess
+import sys
 from importlib.metadata import version
 from os import PathLike
 from pathlib import Path
@@ -23,6 +24,7 @@
     generate_flow_tools_json,
     generate_random_string,
     parse_variant,
+    dump_flow_result,
 )
 from promptflow._sdk.entities._validation import ValidationResult
 from promptflow._sdk.operations._run_submitter import remove_additional_includes, variant_overwrite_context
@@ -71,6 +73,21 @@ def test(
         result = self._test(
             flow=flow, inputs=inputs, variant=variant, node=node, environment_variables=environment_variables, **kwargs
         )
+
+        dump_test_result = kwargs.get("dump_test_result", False)
+        if dump_test_result:
+            # Dump flow/node test info
+            flow = load_flow(flow)
+            if node:
+                dump_flow_result(flow_folder=flow.code, node_result=result, prefix=f"flow-{node}.node")
+            else:
+                if variant:
+                    tuning_node, node_variant = parse_variant(variant)
+                    prefix = f"flow-{tuning_node}-{node_variant}"
+                else:
+                    prefix = "flow"
+                dump_flow_result(flow_folder=flow.code, flow_result=result, prefix=prefix)
+
         TestSubmitter._raise_error_when_test_failed(result, show_trace=node is not None)
         return result.output
 
@@ -162,6 +179,7 @@ def _is_chat_flow(flow):
             error_msg = "chat_history is required in the inputs of chat flow"
         return is_chat_flow, chat_history_input_name, error_msg
 
+    @monitor_operation(activity_name="pf.flows._chat", activity_type=ActivityType.INTERNALCALL)
     def _chat(
         self,
         flow,
@@ -202,6 +220,26 @@ def _chat(
                 show_step_output=kwargs.get("show_step_output", False),
             )
 
+    @monitor_operation(activity_name="pf.flows._chat_with_ui", activity_type=ActivityType.INTERNALCALL)
+    def _chat_with_ui(self, script):
+        try:
+            import bs4  # noqa: F401
+            import streamlit_quill  # noqa: F401
+            from streamlit.web import cli as st_cli
+        except ImportError as ex:
+            raise UserErrorException(
+                f"Please try 'pip install promptflow[executable]' to install dependency, {ex.msg}."
+            )
+        sys.argv = [
+            "streamlit",
+            "run",
+            script,
+            "--global.developmentMode=false",
+            "--client.toolbarMode=viewer",
+            "--browser.gatherUsageStats=false",
+        ]
+        st_cli.main()
+
     def _build_environment_config(self, flow_dag_path: Path):
         flow_info = yaml.safe_load(flow_dag_path.read_text())
         # standard env object: