[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
@@ -150,12 +151,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
@@ -191,11 +195,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

@@ -6,8 +6,8 @@
 import logging
 import os
 import os.path
-import subprocess
 import shutil
+import subprocess
 import sys
 import tempfile
 import uuid
@@ -1141,16 +1141,18 @@ def test_flow_build_executable(self):
                 )
                 # Start the Python script as a subprocess
                 app_file = Path(temp_dir, "app.py").as_posix()
-                process = subprocess.Popen(['python', app_file], stderr=subprocess.PIPE)
+                process = subprocess.Popen(["python", app_file], stderr=subprocess.PIPE)
                 try:
                     # Wait for a specified time (in seconds)
                     wait_time = 5
                     process.wait(timeout=wait_time)
                     if process.returncode == 0:
                         pass
                     else:
-                        raise Exception(f"Process terminated with exit code {process.returncode}, "
-                                        f"{process.stderr.read().decode('utf-8')}")
+                        raise Exception(
+                            f"Process terminated with exit code {process.returncode}, "
+                            f"{process.stderr.read().decode('utf-8')}"
+                        )
                 except (subprocess.TimeoutExpired, KeyboardInterrupt):
                     pass
                 finally:
@@ -1539,3 +1541,42 @@ def test_run_file_with_set_priority(self, pf) -> None:
         except RunNotFoundError:
             pass
         pf.runs.get(name=name2)
+
+    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