[BugFix][CI]Fixing occasional CI failures (#3623)

Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/dfx/perf/scripts/run_benchmark.py

@@ -1,8 +1,6 @@
 import json
 import os
-import subprocess
 import threading
-from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -14,6 +12,7 @@ from tests.dfx.conftest import (
     create_unique_server_params,
     get_benchmark_params_for_server,
     load_configs,
+    run_benchmark,
 )
 from tests.helpers.runtime import OmniServer
 
@@ -98,91 +97,6 @@ def _safe_filename_token(value: Any | None, *, default: str = "na") -> str:
     return s if s else default
 
 
-def run_benchmark(
-    args: list,
-    test_name: str,
-    flow,
-    dataset_name: str,
-    num_prompt,
-    *,
-    baseline_config: dict[str, Any] | None = None,
-    sweep_index: int | None = None,
-    request_rate: Any | None = None,
-    max_concurrency: Any | None = None,
-    random_input_len: Any | None = None,
-    random_output_len: Any | None = None,
-) -> Any:
-    """Run a single benchmark iteration and return the parsed result JSON.
-
-    After ``vllm bench`` writes the JSON, ``result["baseline"]`` holds the same
-    per-metric resolved thresholds as ``assert_result`` (via ``_baseline_thresholds_for_step``).
-    When ``random_input_len`` / ``random_output_len`` are set, they are also written into the result JSON;
-    omitted keys when not configured.
-    """
-    current_dt = datetime.now().strftime("%Y%m%d-%H%M%S")
-    ri = _safe_filename_token(random_input_len)
-    ro = _safe_filename_token(random_output_len)
-    result_filename = f"result_{test_name}_{dataset_name}_{flow}_{num_prompt}_in{ri}_out{ro}_{current_dt}.json"
-    if "--result-filename" in args:
-        print(f"The result file will be overwritten by {result_filename}")
-    command = (
-        ["vllm", "bench", "serve", "--omni"]
-        + args
-        + [
-            "--save-result",
-            "--result-dir",
-            os.environ.get("BENCHMARK_DIR", "tests"),
-            "--result-filename",
-            result_filename,
-        ]
-    )
-    process = subprocess.Popen(
-        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True
-    )
-
-    for line in iter(process.stdout.readline, ""):
-        print(line, end=" ")
-
-    for line in iter(process.stderr.readline, ""):
-        print(line, end=" ")
-
-    if "--result-dir" in command:
-        index = command.index("--result-dir")
-        result_dir = command[index + 1]
-    else:
-        result_dir = "./"
-
-    result_path = os.path.join(result_dir, result_filename)
-    if not os.path.exists(result_path):
-        with open(OMNI_RESULT_TEMPLATE_PATH, encoding="utf-8") as f:
-            template_result: dict[str, Any] = json.load(f)
-        Path(result_path).parent.mkdir(parents=True, exist_ok=True)
-        with open(result_path, "w", encoding="utf-8") as f:
-            json.dump(template_result, f, ensure_ascii=False, indent=2)
-        print(f"Benchmark result file not generated, fallback to template: {result_path}")
-        result = template_result
-    else:
-        with open(result_path, encoding="utf-8") as f:
-            result = json.load(f)
-
-    if baseline_config:
-        result["baseline"] = _baseline_thresholds_for_step(
-            baseline_config,
-            sweep_index=sweep_index,
-            request_rate=request_rate,
-            max_concurrency=max_concurrency,
-        )
-    else:
-        result["baseline"] = {}
-    if random_input_len is not None:
-        result["random_input_len"] = random_input_len
-    if random_output_len is not None:
-        result["random_output_len"] = random_output_len
-    with open(result_path, "w", encoding="utf-8") as f:
-        json.dump(result, f, ensure_ascii=False, indent=2)
-    return result
-
-
 benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
 
 

tests/e2e/online_serving/test_qwen3_omni_expansion.py

@@ -49,34 +49,17 @@ def get_batch_token_config(default_path):
     )
 
 
-def get_default_config(default_path):
-    """Flip async_chunk on and bump stage 0 thinker output to 2048 tokens.
-
-    Pipeline registry (qwen3_omni/pipeline.py) already wires
-    thinker2talker_async_chunk / talker2code2wav_async_chunk on stages 0/1,
-    so no per-stage processor override is needed. Using only flat-schema
-    writes so _parse_stage_deploy stays in its flat branch (nested
-    ``engine_args:`` would drop other overlay fields).
-    """
-    return modify_stage_config(
-        default_path,
-        updates={
-            "stages": {0: {"default_sampling_params.max_tokens": 2048}},
-        },
-    )
-
-
 # CI deploy YAML (single file; xpu deltas applied via ``platforms:`` section).
 # The overlay explicitly sets ``async_chunk: False``, so ``default`` tests the
 # sync path and ``async_chunk`` tests the streaming path with a longer thinker
 # output — two distinct scenarios, kept as separate parametrizations.
-default_path = get_deploy_config_path("ci/qwen3_omni_moe.yaml")
+default_path = get_deploy_config_path("qwen3_omni_moe.yaml")
 
 test_params = [
     pytest.param(
         OmniServerParams(
             model=model,
-            stage_config_path=get_default_config(default_path),
+            stage_config_path=default_path,
             use_stage_cli=True,
             server_args=["--no-async-chunk"],
         ),
@@ -85,7 +68,7 @@ test_params = [
     pytest.param(
         OmniServerParams(
             model=model,
-            stage_config_path=get_default_config(default_path),
+            stage_config_path=default_path,
             use_stage_cli=True,
             server_args=["--async-chunk"],
         ),