[Fix] Recover disabled tests

src/promptflow-core/pyproject.toml

@@ -76,7 +76,7 @@ promptflow-recording = {path = "../promptflow-recording"}
 [tool.poetry.group.test.dependencies]
 pytest = "*"
 pytest-cov = "*"
-pytest-xdist = "*"
+pytest-xdist = {extras = ["psutil"], version = "*"}
 pytest-mock = "*"
 mock = "*"
 
@@ -95,7 +95,6 @@ markers = [
 # durations - list the slowest test durations
 addopts = """
 --junit-xml=test-results.xml \
---dist loadfile \
 --log-level=info \
 --log-format="%(asctime)s %(levelname)s %(message)s" \
 --log-date-format="[%Y-%m-%d %H:%M:%S]" \

src/promptflow-core/tests/conftest.py

@@ -104,7 +104,7 @@ def start_patches(patch_targets):
  from promptflow.recording.record_mode import check_pydantic_v2
 
  check_pydantic_v2()
- file_path = RECORDINGS_TEST_CONFIGS_ROOT / "node_cache.shelve"
+ file_path = RECORDINGS_TEST_CONFIGS_ROOT / "core_node_cache.shelve"
  RecordStorage.get_instance(file_path)
 
  from promptflow._core.tool import tool as original_tool

src/promptflow-devkit/pyproject.toml

@@ -103,7 +103,7 @@ promptflow-recording = {path = "../promptflow-recording"}
 [tool.poetry.group.test.dependencies]
 pytest = "*"
 pytest-cov = "*"
-pytest-xdist = "*"
+pytest-xdist = {extras = ["psutil"], version = "*"}
 pytest-mock = "*"
 pytest-asyncio = "*"
 mock = "*"

src/promptflow-devkit/tests/conftest.py

@@ -127,7 +127,11 @@ def prepare_symbolic_flow() -> str:
  for file_name in os.listdir(source_folder):
  if not Path(file_name).exists():
  os.symlink(source_folder / file_name, file_name)
- return target_folder
+ yield target_folder
+ with _change_working_dir(target_folder):
+ for file_name in os.listdir(target_folder):
+ if Path(file_name).exists():
+ os.remove(file_name)
 
 
 @pytest.fixture(scope="session")

src/promptflow-devkit/tests/sdk_cli_global_config_test/e2etests/test_connection_cli.py

@@ -1,32 +1,12 @@
-import os
-import sys
-
 import pytest
 from _constants import PROMPTFLOW_ROOT
+from sdk_cli_test.e2etests.test_cli import run_pf_command
 
-from promptflow._cli._pf.entry import main
 from promptflow._sdk._constants import SCRUBBED_VALUE
 
 CONNECTIONS_DIR = PROMPTFLOW_ROOT / "tests/test_configs/connections"
 
 
-def run_pf_command(*args, cwd=None):
- """Run a pf command with the given arguments and working directory.
-
- There have been some unknown issues in using subprocess on CI, so we use this function instead, which will also
- provide better debugging experience.
- """
- origin_argv, origin_cwd = sys.argv, os.path.abspath(os.curdir)
- try:
- sys.argv = ["pf"] + list(args)
- if cwd:
- os.chdir(cwd)
- main()
- finally:
- sys.argv = origin_argv
- os.chdir(origin_cwd)
-
-
 @pytest.mark.usefixtures("global_config")
 @pytest.mark.e2etest
 class TestConnectionCli:

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_chat_group.py

@@ -11,7 +11,6 @@
 @pytest.mark.e2etest
 @pytest.mark.usefixtures("use_secrets_config_file", "recording_injection", "setup_local_connection")
 class TestChatGroup:
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_chat_group_basic_invoke(self):
  question = "What's the most beautiful thing in the world?"
  ground_truth = "The world itself."

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

@@ -4,7 +4,9 @@
 import logging
 import os
 import os.path
+import random
 import shutil
+import string
 import subprocess
 import sys
 import tempfile
@@ -46,6 +48,28 @@
 TARGET_URL = "https://www.youtube.com/watch?v=o5ZQyXaAv1g"
 
 
+def randomword(length):
+ letters = string.ascii_lowercase
+ return "".join(random.choice(letters) for i in range(length))
+
+
+def lock_folder(folder_or_file: str):
+ """Lock the folder or file to prevent concurrent access.
+ Hash the folder or file and create a temp file in os temp dir.
+ """
+ folder_hash = str(hash(folder_or_file))
+ lock_file = os.path.join(tempfile.gettempdir(), f"lock_{folder_hash}")
+ loop_count = 0
+ while os.path.exists(lock_file) and loop_count < 600:
+ sleep(1)
+ loop_count += 1
+ if loop_count >= 600:
+ raise Exception(f"Cannot acquire lock for {folder_or_file}")
+ with open(lock_file, "w") as f:
+ f.write("lock")
+ return lock_file
+
+
 # TODO: move this to a shared utility module
 def run_pf_command(*args, cwd=None):
  """Run a pf command with the given arguments and working directory.
@@ -56,12 +80,23 @@ def run_pf_command(*args, cwd=None):
  origin_argv, origin_cwd = sys.argv, os.path.abspath(os.curdir)
  try:
  sys.argv = ["pf"] + list(args)
+ lock_file = randomword(10)
+ for index, item in enumerate(list(args)):
+ if item.startswith("--flow"):
+ try:
+ lock_file = list(args)[index + 1]
+ except IndexError:
+ pass
+ break
+ lock_file = lock_folder(lock_file)
  if cwd:
  os.chdir(cwd)
  main()
  finally:
  sys.argv = origin_argv
  os.chdir(origin_cwd)
+ if os.path.exists(lock_file) and os.path.isfile(lock_file):
+ os.remove(lock_file)
 
 
 @pytest.mark.usefixtures(
@@ -365,7 +400,6 @@ def test_flow_with_aad_connection(self):
  output = json.loads(open(output_path, "r", encoding="utf-8").read())
  assert output["result"] == "meid_token"
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_pf_flow_test_with_non_english_input_output(self, capsys):
  # disable trace to not invoke prompt flow service, which will print unexpected content to stdout
  with mock.patch("promptflow._sdk._tracing.is_trace_feature_disabled", return_value=True):
@@ -747,7 +781,6 @@ def test_init_eval_flow(self):
  run_pf_command("flow", "test", "--flow", flow_name, "--inputs", "groundtruth=App", "prediction=App")
  self._validate_requirement(Path(temp_dir) / flow_name / "flow.dag.yaml")
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_init_chat_flow(self):
  temp_dir = mkdtemp()
  with _change_working_dir(temp_dir):
@@ -969,7 +1002,6 @@ def check_connection_and_deployment(flow_folder, connection, deployment):
  assert not (flow_folder / "azure_openai.yaml").exists()
  assert not (flow_folder / "openai.yaml").exists()
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_flow_chat(self, monkeypatch, capsys):
  chat_list = ["hi", "what is chat gpt?"]
 
@@ -1018,7 +1050,6 @@ def mock_input(*args, **kwargs):
  assert "show_answer:" in outerr.out
  assert "[show_answer]: print:" in outerr.out
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_invalid_chat_flow(self, monkeypatch, capsys):
  def mock_input(*args, **kwargs):
  if chat_list:
@@ -1077,7 +1108,6 @@ def mock_input(*args, **kwargs):
  outerr = capsys.readouterr()
  assert "chat output is not configured" in outerr.out
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_chat_with_stream_output(self, monkeypatch, capsys):
  chat_list = ["hi", "what is chat gpt?"]
 
@@ -1124,7 +1154,6 @@ def mock_input(*args, **kwargs):
  )
  assert detail_path.exists()
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_flow_test_with_default_chat_history(self):
  run_pf_command(
  "flow",
@@ -1144,7 +1173,6 @@ def test_flow_test_with_default_chat_history(self):
  ]
  assert details["flow_runs"][0]["inputs"]["chat_history"] == expect_chat_history
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_flow_test_with_user_defined_chat_history(self, monkeypatch, capsys):
  chat_list = ["hi", "what is chat gpt?"]
 
@@ -2723,6 +2751,8 @@ def test_class_based_eager_flow_test_without_yaml(self, pf, capsys):
  stdout, _ = capsys.readouterr()
  assert "obj_input" in stdout
  assert "func_input" in stdout
+ # cleanup
+ os.remove(f"{EAGER_FLOWS_DIR}/basic_callable_class_without_yaml/flow.flex.yaml")
 
  def test_eager_flow_test_without_yaml_ui(self, pf, capsys):
  run_pf_command(

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_csharp_cli.py

@@ -1,34 +1,15 @@
 import json
 import os
 import os.path
-import sys
 from pathlib import Path
 from typing import TypedDict
 
 import pytest
+from sdk_cli_test.e2etests.test_cli import run_pf_command
 
-from promptflow._cli._pf.entry import main
 from promptflow._sdk._utilities.serve_utils import find_available_port
 
 
-# TODO: move this to a shared utility module
-def run_pf_command(*args, cwd=None):
- """Run a pf command with the given arguments and working directory.
-
- There have been some unknown issues in using subprocess on CI, so we use this function instead, which will also
- provide better debugging experience.
- """
- origin_argv, origin_cwd = sys.argv, os.path.abspath(os.curdir)
- try:
- sys.argv = ["pf"] + list(args)
- if cwd:
- os.chdir(cwd)
- main()
- finally:
- sys.argv = origin_argv
- os.chdir(origin_cwd)
-
-
 class CSharpProject(TypedDict):
  flow_dir: str
  data: str

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_experiment.py

@@ -399,7 +399,6 @@ def test_experiment_with_chat_group(self, pf: PFClient):
  exp = pf._experiments.get(exp.name)
  exp = ExperimentOrchestrator(pf, exp).start()
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  @pytest.mark.usefixtures("use_secrets_config_file", "recording_injection", "setup_local_connection")
  def test_experiment_test_chat_group_node(self, pf: PFClient):
  template_path = EXP_ROOT / "chat-group-node-exp-template" / "exp.yaml"

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_flow_run.py

@@ -1302,7 +1302,7 @@ def test_flow_with_nan_inf_metrics(self, pf: PFClient, monkeypatch) -> None:
  },
  # signature not in original YAML
  lambda: "inputs" not in load_yaml(f"{EAGER_FLOWS_DIR}/simple_with_yaml/flow.flex.yaml"),
- {"inputs.input_val": ["input1"], "inputs.line_number": [0], "outputs.output": ["Hello world! input1"]},
+ {"inputs.input_val": ["gpt"], "inputs.line_number": [0], "outputs.output": ["Hello world! gpt"]},
  id="with_yaml",
  ),
  pytest.param(
@@ -1427,6 +1427,7 @@ def test_flex_flow_run(
  def test_flex_flow_with_local_imported_func(self, pf):
  # run eager flow against a function from local file
  with inject_sys_path(f"{EAGER_FLOWS_DIR}/multiple_entries"):
+
  from entry2 import my_flow2
 
  run = pf.run(
@@ -1642,6 +1643,7 @@ def test_run_with_deployment_overwrite(self, pf):
  run_dict = run._to_dict()
  assert "error" not in run_dict, run_dict["error"]
 
+ @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603")
  def test_deployment_overwrite_failure(self, local_client, local_aoai_connection, pf):
  # deployment name not exist
  run = pf.run(

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_flow_serve.py

@@ -227,7 +227,6 @@ def test_unknown_api(flow_serving_client):
  assert response.status_code == 404
 
 
-@pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
 @pytest.mark.usefixtures("recording_injection", "setup_local_connection")
 @pytest.mark.e2etest
 @pytest.mark.parametrize(

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_flow_serve_fastapi.py

@@ -228,7 +228,6 @@ def test_unknown_api(fastapi_flow_serving_client):
  assert response.status_code == 405
 
 
-@pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
 @pytest.mark.usefixtures("recording_injection", "setup_local_connection")
 @pytest.mark.e2etest
 @pytest.mark.parametrize(

src/promptflow-devkit/tests/sdk_cli_test/e2etests/test_flow_test.py

@@ -1,4 +1,5 @@
 import logging
+import os
 import sys
 import tempfile
 from dataclasses import is_dataclass
@@ -110,7 +111,6 @@ def test_pf_test_flow_with_script_tool_with_custom_strong_type_connection(self):
  result = _client.test(flow=flow_path, inputs={"input_param": "Hello World!"}, node="my_script_tool")
  assert result == "connection_value is MyCustomConnection: True"
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_pf_test_with_streaming_output(self):
  flow_path = Path(f"{FLOWS_DIR}/chat_flow_with_stream_output")
  result = _client.test(flow=flow_path)
@@ -283,6 +283,8 @@ def test_class_based_eager_flow_test_without_yaml(self):
  flow="callable_without_yaml:MyFlow", inputs={"func_input": "input"}, init={"obj_input": "val"}
  )
  assert result["func_input"] == "input"
+ # Cleanup
+ os.remove("flow.flex.yaml")
 
  def test_eager_flow_test_with_yaml(self):
  clear_module_cache("entry")
@@ -430,7 +432,6 @@ def test_eager_flow_stream_output(self):
  # directly return the consumed generator to align with the behavior of DAG flow test
  assert result.output == "Hello world! "
 
- @pytest.mark.skipif(pytest.is_replay, reason="BUG 3178603, recording instable")
  def test_stream_output_with_builtin_llm(self):
  flow_path = Path(f"{EAGER_FLOWS_DIR}/builtin_llm/").absolute()
  # TODO(3171565): support default value for list & dict

src/promptflow-devkit/tests/sdk_cli_test/unittests/test_flow_serve_cli.py

@@ -1,34 +1,13 @@
-import os
-import sys
 from pathlib import Path
 
 import pytest
 from mock import mock
-
-from promptflow._cli._pf.entry import main
+from sdk_cli_test.e2etests.test_cli import run_pf_command
 
 FLOWS_DIR = Path("./tests/test_configs/flows")
 EAGER_FLOWS_DIR = Path("./tests/test_configs/eager_flows")
 
 
-# TODO: move this to a shared utility module
-def run_pf_command(*args, cwd=None):
- """Run a pf command with the given arguments and working directory.
-
- There have been some unknown issues in using subprocess on CI, so we use this function instead, which will also
- provide better debugging experience.
- """
- origin_argv, origin_cwd = sys.argv, os.path.abspath(os.curdir)
- try:
- sys.argv = ["pf"] + list(args)
- if cwd:
- os.chdir(cwd)
- main()
- finally:
- sys.argv = origin_argv
- os.chdir(origin_cwd)
-
-
 @pytest.mark.cli_test
 @pytest.mark.unittest
 class TestRun:

src/promptflow-recording/promptflow/recording/local/record_storage.py

@@ -342,6 +342,14 @@ def set_record(self, input_dict: Dict, output):
  try:
  line_record_pointer = self.file_records_pointer[hash_value]
  except KeyError:
+ if output_type == "Exception":
+ # https://help.openai.com/en/articles/6897213-openai-library-error-types-guidance
+ if (
+ output_value.type != "APIConnectionError"
+ and output_value.type != "InvalidRequestError"
+ and output_value.type != "DeploymentNotFound"
+ ):
+ raise output
  self.file_records_pointer[hash_value] = {
  "input": input_dict,
  "output": output_value,

src/promptflow-recording/recordings/local/core_node_cache.shelve.bak

@@ -0,0 +1,2 @@
+'ab3563e99d1ee052e067952ab332536a6bf5c025', (0, 1308)
+'67b119c190b574a9a275e8bdd93bcd06487d5318', (1536, 4997)

src/promptflow-recording/recordings/local/core_node_cache.shelve.dir

@@ -0,0 +1,2 @@
+'ab3563e99d1ee052e067952ab332536a6bf5c025', (0, 1308)
+'67b119c190b574a9a275e8bdd93bcd06487d5318', (1536, 4997)