[BufFix] Redact data when printing secret in the node

src/promptflow/promptflow/_sdk/operations/_test_submitter.py

@@ -10,6 +10,7 @@
 from types import GeneratorType
 from typing import Any, Mapping
 
+from promptflow._internal import ConnectionManager
 from promptflow._sdk._constants import LOGGER_NAME, PROMPT_FLOW_DIR_NAME
 from promptflow._sdk._utils import dump_flow_result, parse_variant
 from promptflow._sdk.entities._flow import Flow
@@ -149,12 +150,15 @@ def flow_test(
 
         if not connections:
             connections = SubmitterHelper.resolve_connections(flow=self.flow, client=self._client)
+        credential_list = ConnectionManager(connections).get_secret_list()
+
         # resolve environment variables
         SubmitterHelper.resolve_environment_variables(environment_variables=environment_variables, client=self._client)
         environment_variables = environment_variables if environment_variables else {}
         SubmitterHelper.init_env(environment_variables=environment_variables)
 
-        with LoggerOperations(file_path=self.flow.code / PROMPT_FLOW_DIR_NAME / "flow.log", stream=stream_log):
+        with LoggerOperations(file_path=self.flow.code / PROMPT_FLOW_DIR_NAME / "flow.log",
+                              stream=stream_log, credential_list=credential_list):
             storage = DefaultRunStorage(base_dir=self.flow.code, sub_dir=Path(".promptflow/intermediate"))
             flow_executor = FlowExecutor.create(
                 self.flow.path, connections, self.flow.code, storage=storage, raise_ex=False
@@ -190,11 +194,14 @@ def node_test(
         from promptflow.executor import FlowExecutor
 
         connections = SubmitterHelper.resolve_connections(flow=self.flow, client=self._client)
+        credential_list = ConnectionManager(connections).get_secret_list()
+
         # resolve environment variables
         SubmitterHelper.resolve_environment_variables(environment_variables=environment_variables, client=self._client)
         SubmitterHelper.init_env(environment_variables=environment_variables)
 
-        with LoggerOperations(file_path=self.flow.code / PROMPT_FLOW_DIR_NAME / f"{node_name}.node.log", stream=stream):
+        with LoggerOperations(file_path=self.flow.code / PROMPT_FLOW_DIR_NAME / f"{node_name}.node.log",
+                              stream=stream, credential_list=credential_list):
             result = FlowExecutor.load_and_exec_node(
                 self.flow.path,
                 node_name,

src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py

@@ -1407,3 +1407,45 @@ def test_flow_test_with_image_input_and_output(self):
         assert output_path.exists()
         image_path = Path(FLOWS_DIR) / "python_tool_with_image_input_and_output" / ".promptflow" / "intermediate"
         assert image_path.exists()
+
+    def test_data_scrubbing(self):
+        # Prepare connection
+        run_pf_command(
+            "connection",
+            "create",
+            "--file",
+            f"{CONNECTIONS_DIR}/custom_connection.yaml",
+            "--name",
+            "custom_connection")
+        # Test flow run
+        run_pf_command(
+            "flow",
+            "test",
+            "--flow",
+            f"{FLOWS_DIR}/print_secret_flow",
+        )
+        output_path = Path(FLOWS_DIR) / "print_secret_flow" / ".promptflow" / "flow.output.json"
+        assert output_path.exists()
+        log_path = Path(FLOWS_DIR) / "print_secret_flow" / ".promptflow" / "flow.log"
+        with open(log_path, "r") as f:
+            log_content = f.read()
+            assert "**data_scrubbed**" in log_content
+
+        # Test node run
+        run_pf_command(
+            "flow",
+            "test",
+            "--flow",
+            f"{FLOWS_DIR}/print_secret_flow",
+            "--node",
+            "print_secret",
+            "--inputs",
+            "conn=custom_connection",
+            "inputs.topic=atom"
+        )
+        output_path = Path(FLOWS_DIR) / "print_secret_flow" / ".promptflow" / "flow-print_secret.node.detail.json"
+        assert output_path.exists()
+        log_path = Path(FLOWS_DIR) / "print_secret_flow" / ".promptflow" / "print_secret.node.log"
+        with open(log_path, "r") as f:
+            log_content = f.read()
+        assert "**data_scrubbed**" in log_content

src/promptflow/tests/test_configs/flows/print_secret_flow/flow.dag.yaml

@@ -0,0 +1,17 @@
+inputs:
+  key:
+    type: string
+    default: text
+outputs:
+  output:
+    type: string
+    reference: ${print_secret.output}
+nodes:
+- name: print_secret
+  type: python
+  source:
+    type: code
+    path: print_secret.py
+  inputs:
+    connection: custom_connection
+    text: ${inputs.key}

src/promptflow/tests/test_configs/flows/print_secret_flow/print_secret.py

@@ -0,0 +1,11 @@
+import os
+
+from promptflow import tool
+from promptflow.connections import CustomConnection
+
+
+@tool
+def print_secret(text: str, connection: CustomConnection):
+    print(connection["key1"])
+    print(connection["key2"])
+    return text