[Test] add stability test case for wan2.2, qwen-tts, qwen3-omni and qwen-image model and modified conftest.py in test/dfx/ (#2817)

Signed-off-by: zhumingjue <zhumingjue@huawei.com>
Signed-off-by: zhumingjue138 <zhumingjue@huawei.com>

docs/contributing/ci/CI_5levels.md

@@ -105,7 +105,8 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
       <td> Depends on reality </td>
       <td>
         <strong>Stability:</strong><br>
-        /tests/dfx/stability/tests/test.json<br>
+        /tests/dfx/stability/tests/test_qwen3_omni.json<br>
+        /tests/dfx/stability/tests/test_wan22.json<br>
         <strong>Reliability:</strong><br>
         tests/e2e/reliability/test_{model_name}.py
       </td>
@@ -574,12 +575,12 @@ L5 level testing focuses on the performance of model services under ***long-runn
 
 ### 4.2 Testing Content and Scope
 
--   ***Long-term Stability (Stability) Testing***: Uses the `tests/dfx/stability/tests/test.json` configuration to run the service under moderate load for an extended period (e.g., over 12 hours), monitoring whether metrics like memory/VRAM usage, response time, and throughput degrade over time, and whether the service process remains stable.
+-   ***Long-term Stability (Stability) Testing***: Uses JSON under `tests/dfx/stability/tests/` (for example `test_qwen3_omni.json` and `test_wan22.json`) to run the service under moderate load for an extended period (e.g., over 12 hours), monitoring whether metrics like memory/VRAM usage, response time, and throughput degrade over time, and whether the service process remains stable.
 -   ***Reliability Testing***: Uses `tests/e2e/reliability/test_{model_name}.py` to actively simulate various fault and abnormal scenarios, such as: dependent service interruption, abnormal input data, network flicker, hardware resource preemption, etc., to verify the system's fault tolerance, self-healing, and graceful degradation capabilities.
 
 ### 4.3 Test Directory and Execution Files
 
--   ***Stability Test Configuration***: `tests/dfx/stability/tests/test.json`
+-   ***Stability Test Configuration***: `tests/dfx/stability/tests/test_qwen3_omni.json`, `tests/dfx/stability/tests/test_wan22.json` (one JSON per model / runner family)
 -   ***Reliability Test Suite***: `tests/e2e/reliability/test_{model_name}.py`
 
 ### 4.4 Execution Method and Example
@@ -590,7 +591,7 @@ L5 level testing focuses on the performance of model services under ***long-runn
 <details>
 <summary> Test Examples</summary>
 
-When you want to add L5-level stability test cases, you can refer to the following format for case addition in `tests/dfx/stability/tests/test.json`:
+When you want to add L5-level stability test cases, add or extend the appropriate JSON file under `tests/dfx/stability/tests/` (for example `test_qwen3_omni.json` for Omni bench traffic, or `test_wan22.json` for diffusion `/v1/videos` workloads). The following illustrates the Qwen3-Omni shape:
 
 ```json
 {
@@ -661,7 +662,7 @@ All other optional parameters follow the same rules as the in Chapter 3.4.
 
 </details>
 
--   -   ***Stability***: `pytest -s -v tests/dfx/stability/scripts/test_{model_name}.py`
+-   -   ***Stability***: `pytest -s -v tests/dfx/stability/scripts/test_stability_qwen3_omni.py` or `pytest -s -v tests/dfx/stability/scripts/test_stability_wan22.py` (or add `test_stability_<model>.py` alongside a matching JSON config)
     -   ***Reliability***: `pytest -s -v tests/e2e/reliability/test_{model_name}.py`
 
 ## Summary

docs/contributing/ci/test_guide.md

@@ -83,6 +83,23 @@ Our test scripts use the pytest framework. First, please use `git clone https://
     pytest -s -v -m "full_model and omni and H100" --run-level=full_model
     ```
 
+=== "L5 level"
+
+    L5 includes stability and reliability testing. Typical commands:
+
+    ```bash
+    cd tests
+
+    # Stability: Qwen3-Omni
+    pytest -s -v dfx/stability/scripts/test_stability_qwen3_omni.py
+
+    # Stability: Wan2.2 (v1/videos diffusion benchmark loop)
+    pytest -s -v dfx/stability/scripts/test_stability_wan22.py
+
+    ```
+
+    The latest L5 commands for CI can be found in the [pipeline](https://github.com/vllm-project/vllm-omni/blob/main/.buildkite/test-ready.yml).
+
 You can find more information about markers in the documentation: [marker doc](./tests_markers.md)
 
 ## Adding New Test Cases

tests/dfx/conftest.py

@@ -1,4 +1,7 @@
 import json
+import os
+import subprocess
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -37,12 +40,45 @@ def modify_stage(default_path, updates, deletes):
     return path
 
 
+def _build_serve_args(serve_args: Any) -> list[str]:
+    """Convert server_params.serve_args to a flat CLI args list."""
+    if serve_args is None:
+        return []
+    if isinstance(serve_args, list):
+        return [str(item) for item in serve_args]
+    if not isinstance(serve_args, dict):
+        raise TypeError(f"serve_args must be dict/list/None, got {type(serve_args).__name__}")
+
+    args: list[str] = []
+    for key, value in serve_args.items():
+        flag = f"--{str(key).replace('_', '-')}"
+        if isinstance(value, bool):
+            if value:
+                args.append(flag)
+            continue
+        if value is None:
+            continue
+        if isinstance(value, (dict, list)):
+            args.extend([flag, json.dumps(value, ensure_ascii=False, separators=(",", ":"))])
+            continue
+        args.extend([flag, str(value)])
+    return args
+
+
 def create_unique_server_params(
     configs: list[dict[str, Any]],
     stage_configs_dir: Path,
 ) -> list[tuple[str, str, str | None, str | None, tuple[str, ...]]]:
-    unique_params = []
-    seen = set()
+    """Return one row per unique server configuration (same 5-tuple shape as upstream).
+
+    ``(test_name, model, deploy_yaml_path, stage_overrides_json, extra_cli_args)``.
+
+    JSON ``server_params.serve_args`` (dict/list) is expanded via ``_build_serve_args``
+    and **prepended** to ``extra_cli_args`` so perf / stability ``omni_server`` fixtures
+    stay identical to main while still honoring ``serve_args`` in benchmark JSON.
+    """
+    unique_params: list[tuple[str, str, str | None, str | None, tuple[str, ...]]] = []
+    seen: set[tuple[str, str, str | None, str | None, tuple[str, ...]]] = set()
     for config in configs:
         test_name = config["test_name"]
         server_params = config["server_params"]
@@ -63,7 +99,9 @@ def create_unique_server_params(
         # ``vllm_omni.entrypoints.cli.main serve`` — used for flags that
         # don't map to stage-level overrides, e.g. ``--async-chunk`` /
         # ``--no-async-chunk`` toggling the deploy-level async_chunk bool.
-        extra_cli_args = tuple(server_params.get("extra_cli_args") or ())
+        serve_flat = _build_serve_args(server_params.get("serve_args"))
+        raw_extra = tuple(server_params.get("extra_cli_args") or ())
+        extra_cli_args = tuple(serve_flat) + raw_extra
 
         server_param = (test_name, model, stage_config_path, stage_overrides_json, extra_cli_args)
         if server_param not in seen:
@@ -109,6 +147,139 @@ def create_benchmark_indices(
     return indices
 
 
+def _safe_filename_token(value: Any | None, *, default: str = "na") -> str:
+    """Make a single path segment safe for result filenames on common filesystems."""
+    if value is None:
+        return default
+    s = str(value).strip()
+    for bad in ("/", "\\", ":", "*", "?", '"', "<", ">", "|"):
+        s = s.replace(bad, "_")
+    return s if s else default
+
+
+def _resolve_baseline_value(
+    baseline_raw: Any,
+    *,
+    sweep_index: int | None,
+    max_concurrency: Any = None,
+    request_rate: Any = None,
+) -> Any:
+    """Pick the baseline threshold for this sweep step."""
+    if baseline_raw is None:
+        return 100000
+    if isinstance(baseline_raw, dict):
+        if max_concurrency is not None:
+            for key in (max_concurrency, str(max_concurrency)):
+                if key in baseline_raw:
+                    return baseline_raw[key]
+        if request_rate is not None:
+            for key in (request_rate, str(request_rate)):
+                if key in baseline_raw:
+                    return baseline_raw[key]
+        raise KeyError(
+            f"baseline dict has no key for max_concurrency={max_concurrency!r} "
+            f"or request_rate={request_rate!r}; keys={list(baseline_raw.keys())!r}"
+        )
+    if isinstance(baseline_raw, (list, tuple)):
+        if sweep_index is None:
+            raise ValueError("list baseline requires sweep_index")
+        if not (0 <= sweep_index < len(baseline_raw)):
+            raise IndexError(f"baseline list len={len(baseline_raw)} has no index {sweep_index}")
+        return baseline_raw[sweep_index]
+    return baseline_raw
+
+
+def _baseline_thresholds_for_step(
+    baseline_data: dict[str, Any],
+    *,
+    sweep_index: int | None = None,
+    max_concurrency: Any = None,
+    request_rate: Any = None,
+) -> dict[str, Any]:
+    """Resolve baseline config to one threshold per metric for this iteration."""
+    return {
+        metric_name: _resolve_baseline_value(
+            baseline_raw,
+            sweep_index=sweep_index,
+            max_concurrency=max_concurrency,
+            request_rate=request_rate,
+        )
+        for metric_name, baseline_raw in baseline_data.items()
+    }
+
+
+def run_benchmark(
+    args: list[str],
+    test_name: str,
+    flow: Any,
+    dataset_name: str,
+    num_prompt: int,
+    *,
+    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,
+) -> dict[str, Any]:
+    """Run one ``vllm bench serve --omni`` iteration and return parsed metrics."""
+    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
+        + [
+            "--num-warmups",
+            "2",
+            "--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)
+    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
+
+
 def pytest_addoption(parser: pytest.Parser) -> None:
     """Register shared CLI options for DFX benchmark suites."""
     parser.addoption(

tests/dfx/stability/conftest.py

@@ -1,4 +1,14 @@
-"""Stability pytest hooks and fixtures."""
+"""
+Stability-specific conftest: when pytest is executed under this directory,
+resource monitoring is started before each test and finalized after each test,
+so each stability test case gets its own HTML report (one report per case).
+No need to wrap pytest with `bash resource_monitor.sh run -- pytest ...`.
+
+Duration-based benchmark helper functions are hosted in ``helpers.py``,
+while this file focuses on pytest fixtures and setup/teardown.
+"""
+
+from __future__ import annotations
 
 import subprocess
 import sys
@@ -6,12 +16,66 @@ import threading
 
 import pytest
 
+from tests.dfx.conftest import get_benchmark_params_for_server
 from tests.dfx.stability.helpers import (
     finalize_resource_monitor,
     report_latest_gpu_samples,
     start_resource_monitor,
     wait_for_run_dir,
 )
+from tests.helpers.runtime import OmniServer
+
+DEFAULT_STABILITY_SERVER_TIMEOUT_ARGS = ["--stage-init-timeout", "600", "--init-timeout", "900"]
+
+_omni_server_lock = threading.Lock()
+
+
+@pytest.fixture(scope="module")
+def omni_server(request: pytest.FixtureRequest):
+    """Start OmniServer for stability tests, with per-module timeout override."""
+    timeout_args = getattr(request.module, "STABILITY_SERVER_TIMEOUT_ARGS", DEFAULT_STABILITY_SERVER_TIMEOUT_ARGS)
+    with _omni_server_lock:
+        # Same 5-tuple and CLI composition as ``tests/dfx/perf/scripts/run_benchmark.py`` on main;
+        # ``serve_args`` from JSON are folded into ``extra_cli_args`` inside
+        # ``create_unique_server_params``.
+        test_name, model, deploy_path, stage_overrides, extra_cli_args = request.param
+
+        print(f"Starting OmniServer with test: {test_name}, model: {model}")
+        server_args = list(timeout_args)
+        if deploy_path:
+            server_args = ["--deploy-config", deploy_path] + server_args
+        if stage_overrides:
+            server_args = ["--stage-overrides", stage_overrides] + server_args
+        if extra_cli_args:
+            server_args = list(extra_cli_args) + server_args
+        with OmniServer(model, server_args) as server:
+            server.test_name = test_name
+            print("OmniServer started successfully")
+            yield server
+            print("OmniServer stopping...")
+        print("OmniServer stopped")
+
+
+@pytest.fixture
+def stability_benchmark_params(request: pytest.FixtureRequest, omni_server):
+    test_name, param_index = request.param
+    if test_name != omni_server.test_name:
+        pytest.skip(f"Skipping parameter for {test_name} - current server is {omni_server.test_name}")
+
+    server_to_benchmark_mapping = getattr(request.module, "server_to_benchmark_mapping", None)
+    if server_to_benchmark_mapping is None:
+        raise ValueError("server_to_benchmark_mapping must be defined in the test module")
+
+    all_params = get_benchmark_params_for_server(test_name, server_to_benchmark_mapping)
+    if not all_params:
+        raise ValueError(f"No benchmark parameters found for test: {test_name}")
+    if param_index >= len(all_params):
+        raise ValueError(f"No benchmark parameters found for index {param_index} in test: {test_name}")
+
+    current = param_index + 1
+    total = len(all_params)
+    print(f"\n  Running benchmark {current}/{total} for {test_name}")
+    return {"test_name": test_name, "params": all_params[param_index]}
 
 
 @pytest.fixture(autouse=True)

tests/dfx/stability/helpers.py

@@ -1,17 +1,32 @@
-"""Stability resource monitor helpers."""
+"""Stability helpers for resource monitoring and benchmark execution."""
 
 from __future__ import annotations
 
+import json
 import os
+import random
+import re
+import shlex
 import subprocess
 import sys
+import tempfile
 import threading
 import time
+from collections.abc import Callable
 from pathlib import Path
+from typing import Any
+
+from tests.dfx.conftest import run_benchmark
 
 STABILITY_DIR = Path(__file__).resolve().parent
 RESOURCE_MONITOR_SCRIPT = STABILITY_DIR / "scripts" / "resource_monitor.sh"
 REPO_ROOT = STABILITY_DIR.parent.parent.parent
+_BUCKET_KEY_PATTERN = re.compile(r"^\(\s*([^,]+)\s*,\s*([^,]+)\s*,\s*([^,]+)\s*\)$")
+
+RunOneBatchFn = Callable[
+    [str, int, str, dict[str, Any], int, float | None, int | None, str, int],
+    dict[str, Any],
+]
 
 
 def start_resource_monitor():
@@ -115,3 +130,375 @@ def finalize_resource_monitor() -> str | None:
         return None
     except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
         return None
+
+
+def _normalize_bench_metrics(raw: dict[str, Any]) -> dict[str, Any]:
+    completed = int(raw.get("completed", raw.get("completed_requests", 0) or 0))
+    failed = int(raw.get("failed", raw.get("failed_requests", 0) or 0))
+    duration = float(raw.get("duration", 0.0) or 0.0)
+    errors = list(raw.get("errors") or [])
+    if failed and not errors:
+        errors = [f"{failed} benchmark request(s) failed"]
+    return {"completed": completed, "failed": failed, "duration": duration, "errors": errors}
+
+
+def _build_base_args(params: dict[str, Any], host: str, port: int) -> list[str]:
+    exclude = {
+        "request_rate",
+        "max_concurrency",
+        "num_prompts",
+        "baseline",
+        "duration_sec",
+        "num_prompts_per_batch",
+    }
+    args = ["--host", host, "--port", str(port)]
+    for key, value in params.items():
+        if key in exclude or value is None:
+            continue
+        arg_name = f"--{key.replace('_', '-')}"
+        if isinstance(value, bool) and value:
+            args.append(arg_name)
+        elif isinstance(value, dict):
+            args.extend([arg_name, json.dumps(value, ensure_ascii=False, separators=(",", ":"))])
+        elif not isinstance(value, bool):
+            args.extend([arg_name, str(value)])
+    return args
+
+
+def _build_diffusion_cmd(
+    host: str,
+    port: int,
+    model: str,
+    params: dict[str, Any],
+    num_prompts: int,
+    request_rate: float | None,
+    max_concurrency: int | None,
+    output_path: Path,
+    diffusion_benchmark_script: Path,
+) -> list[str]:
+    skip_keys = {
+        "request_rate",
+        "max_concurrency",
+        "num_prompts",
+        "baseline",
+        "duration_sec",
+        "num_prompts_per_batch",
+    }
+    cmd: list[str] = [
+        sys.executable,
+        "-u",
+        str(diffusion_benchmark_script),
+        "--host",
+        host,
+        "--port",
+        str(port),
+        "--model",
+        model,
+        "--output-file",
+        str(output_path),
+    ]
+    for key, value in params.items():
+        if key in skip_keys or value is None:
+            continue
+        flag = f"--{str(key).replace('_', '-')}"
+        if isinstance(value, bool) and value:
+            cmd.append(flag)
+        elif isinstance(value, bool):
+            continue
+        elif isinstance(value, (dict, list)):
+            cmd.extend([flag, json.dumps(value, ensure_ascii=False, separators=(",", ":"))])
+        else:
+            cmd.extend([flag, str(value)])
+
+    cmd.extend(["--num-prompts", str(num_prompts)])
+    if request_rate is not None:
+        cmd.extend(["--request-rate", str(request_rate)])
+    else:
+        cmd.extend(["--max-concurrency", str(max_concurrency), "--request-rate", "inf"])
+    return cmd
+
+
+def _sample_int_from_range_spec(value: Any, rng: random.Random) -> Any:
+    """Resolve one value that may be scalar or range spec into an int."""
+    if isinstance(value, int):
+        return value
+
+    if isinstance(value, (list, tuple)) and len(value) == 2 and all(isinstance(v, int) for v in value):
+        low, high = int(value[0]), int(value[1])
+        if low > high:
+            low, high = high, low
+        return rng.randint(low, high)
+
+    if isinstance(value, dict) and {"min", "max"} <= set(value):
+        low, high = int(value["min"]), int(value["max"])
+        if low > high:
+            low, high = high, low
+        return rng.randint(low, high)
+
+    if isinstance(value, str):
+        raw = value.strip()
+        if raw.isdigit():
+            return int(raw)
+        if "-" in raw:
+            parts = [p.strip() for p in raw.split("-", 1)]
+            if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
+                low, high = int(parts[0]), int(parts[1])
+                if low > high:
+                    low, high = high, low
+                return rng.randint(low, high)
+
+    return value
+
+
+def _sample_bucket_key(raw_key: str, rng: random.Random) -> str:
+    """Sample bucket tuple keys that use range syntax, e.g. ``(128-512, 128-512, 1)``."""
+    match = _BUCKET_KEY_PATTERN.match(raw_key.strip())
+    if not match:
+        return raw_key
+
+    sampled_parts: list[int] = []
+    for token in match.groups():
+        sampled = _sample_int_from_range_spec(token.strip(), rng)
+        if not isinstance(sampled, int):
+            return raw_key
+        sampled_parts.append(sampled)
+
+    # For video buckets (height>0 and num_frames>1), enforce even H/W to avoid
+    # ffmpeg yuv420p encoding/decoding failures ("Could not open video stream").
+    if sampled_parts[0] > 0 and sampled_parts[2] > 1:
+        sampled_parts[0] = max(2, sampled_parts[0] - (sampled_parts[0] % 2))
+        sampled_parts[1] = max(2, sampled_parts[1] - (sampled_parts[1] % 2))
+
+    return f"({sampled_parts[0]}, {sampled_parts[1]}, {sampled_parts[2]})"
+
+
+def _sample_stability_batch_params(params: dict[str, Any], batch_index: int) -> dict[str, Any]:
+    """Materialize per-batch random values for configured range fields."""
+    sampled = dict(params)
+    rng = random.Random(time.time_ns() + batch_index)
+
+    for field_name in (
+        "random_input_len",
+        "random_output_len",
+        "random_mm_base_items_per_request",
+        "width",
+        "height",
+    ):
+        if field_name in sampled:
+            sampled[field_name] = _sample_int_from_range_spec(sampled[field_name], rng)
+
+    bucket_config = sampled.get("random_mm_bucket_config")
+    if isinstance(bucket_config, dict):
+        sampled_bucket_config: dict[str, float] = {}
+        for raw_key, probability in bucket_config.items():
+            sampled_key = _sample_bucket_key(str(raw_key), rng)
+            sampled_bucket_config[sampled_key] = sampled_bucket_config.get(sampled_key, 0.0) + float(probability)
+        sampled["random_mm_bucket_config"] = sampled_bucket_config
+
+    return sampled
+
+
+def _run_one_vllm_bench_batch(
+    host: str,
+    port: int,
+    _model: str,
+    params: dict[str, Any],
+    num_prompts: int,
+    request_rate: float | None,
+    max_concurrency: int | None,
+    result_dir: str,
+    batch_index: int,
+) -> dict[str, Any]:
+    base = _build_base_args(params, host, port)
+    if request_rate is not None:
+        args = base + ["--request-rate", str(request_rate), "--num-prompts", str(num_prompts)]
+        flow = request_rate
+    else:
+        args = base + [
+            "--max-concurrency",
+            str(max_concurrency),
+            "--num-prompts",
+            str(num_prompts),
+            "--request-rate",
+            "inf",
+        ]
+        flow = max_concurrency
+
+    # Print the exact per-batch benchmark CLI (randomized params are already materialized).
+    preview_cmd = ["vllm", "bench", "serve", "--omni", *args]
+    print(f"\n[Stability][Batch {batch_index}] Benchmark command:")
+    print(shlex.join(preview_cmd))
+
+    dataset_name = params.get("dataset_name", "random")
+    old_benchmark_dir = os.environ.get("BENCHMARK_DIR")
+    try:
+        os.environ["BENCHMARK_DIR"] = result_dir
+        result = run_benchmark(
+            args=args,
+            test_name="stability",
+            flow=flow,
+            dataset_name=dataset_name,
+            num_prompt=num_prompts,
+            random_input_len=params.get("random_input_len"),
+            random_output_len=params.get("random_output_len"),
+        )
+        return _normalize_bench_metrics(result)
+    except (FileNotFoundError, OSError) as exc:
+        return {
+            "completed": 0,
+            "failed": 1,
+            "duration": 0.0,
+            "errors": [f"Benchmark batch failed: {type(exc).__name__}: {exc}"],
+        }
+    finally:
+        if old_benchmark_dir is not None:
+            os.environ["BENCHMARK_DIR"] = old_benchmark_dir
+        elif "BENCHMARK_DIR" in os.environ:
+            os.environ.pop("BENCHMARK_DIR")
+
+
+def _run_one_diffusion_batch(
+    host: str,
+    port: int,
+    model: str,
+    params: dict[str, Any],
+    num_prompts: int,
+    request_rate: float | None,
+    max_concurrency: int | None,
+    _result_dir: str,
+    _batch_index: int,
+) -> dict[str, Any]:
+    diffusion_benchmark_script = Path(REPO_ROOT / "benchmarks" / "diffusion" / "diffusion_benchmark_serving.py")
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", prefix="stability_diffusion_", delete=False) as tmp:
+        out_path = Path(tmp.name)
+    try:
+        cmd = _build_diffusion_cmd(
+            host,
+            port,
+            model,
+            params,
+            num_prompts,
+            request_rate,
+            max_concurrency,
+            out_path,
+            diffusion_benchmark_script,
+        )
+        proc = subprocess.run(
+            cmd,
+            cwd=str(REPO_ROOT),
+            capture_output=True,
+            text=True,
+        )
+        if proc.stdout:
+            print(proc.stdout, end="" if proc.stdout.endswith("\n") else "\n")
+        if proc.stderr:
+            print(proc.stderr, end="" if proc.stderr.endswith("\n") else "\n")
+        if proc.returncode != 0:
+            return {
+                "completed": 0,
+                "failed": 1,
+                "duration": 0.0,
+                "errors": [f"diffusion_benchmark_serving.py exited {proc.returncode}"],
+            }
+        if not out_path.is_file():
+            return {
+                "completed": 0,
+                "failed": 1,
+                "duration": 0.0,
+                "errors": [f"Missing benchmark output: {out_path}"],
+            }
+        with open(out_path, encoding="utf-8") as file:
+            metrics = json.load(file)
+        return _normalize_bench_metrics(metrics)
+    except (FileNotFoundError, OSError, json.JSONDecodeError) as exc:
+        return {
+            "completed": 0,
+            "failed": 1,
+            "duration": 0.0,
+            "errors": [f"Diffusion batch failed: {type(exc).__name__}: {exc}"],
+        }
+    finally:
+        out_path.unlink(missing_ok=True)
+
+
+def merge_batch_results(batch_results: list[dict[str, Any]], total_duration_sec: float) -> dict[str, Any]:
+    if not batch_results:
+        return {"completed": 0, "failed": 0, "duration": total_duration_sec, "errors": []}
+
+    completed = sum(result.get("completed", 0) for result in batch_results)
+    failed = sum(result.get("failed", 0) for result in batch_results)
+    merged: dict[str, Any] = {
+        "completed": completed,
+        "failed": failed,
+        "duration": total_duration_sec,
+        "errors": [],
+    }
+    for result in batch_results:
+        merged["errors"].extend(result.get("errors") or [])
+    return merged
+
+
+def print_merged_report(result: dict[str, Any]) -> None:
+    fmt = "{:<40} {:<10}"
+    fmt_float = "{:<40} {:<10.2f}"
+    completed = result.get("completed", 0)
+    failed = result.get("failed", 0)
+    duration = float(result.get("duration", 0.0) or 0.0)
+    print("\n============ Stability Benchmark Summary ============")
+    print(fmt.format("Successful requests:", completed))
+    print(fmt.format("Failed requests:", failed))
+    print(fmt_float.format("Total duration (s):", duration))
+    print("==================================================\n")
+
+
+def run_stability_benchmark_loop(
+    host: str,
+    port: int,
+    model: str,
+    duration_sec: int | float,
+    params: dict[str, Any],
+    *,
+    request_rate: float | None,
+    max_concurrency: int | None,
+    result_dir: str,
+    num_prompts_per_batch: int,
+    run_one_batch: RunOneBatchFn,
+    result_filename: str | None = None,
+) -> dict[str, Any]:
+    if (request_rate is None) == (max_concurrency is None):
+        raise ValueError("Exactly one of request_rate or max_concurrency must be specified")
+
+    start_time = time.perf_counter()
+    batch_results: list[dict[str, Any]] = []
+    batch_index = 0
+
+    while True:
+        if (time.perf_counter() - start_time) >= duration_sec:
+            break
+        sampled_params = _sample_stability_batch_params(params, batch_index)
+        result = run_one_batch(
+            host,
+            port,
+            model,
+            sampled_params,
+            num_prompts_per_batch,
+            request_rate,
+            max_concurrency,
+            result_dir,
+            batch_index,
+        )
+        batch_results.append(result)
+        batch_index += 1
+        if (time.perf_counter() - start_time) >= duration_sec:
+            break
+
+    total_duration = time.perf_counter() - start_time
+    merged = merge_batch_results(batch_results, total_duration)
+    print_merged_report(merged)
+
+    if result_filename and result_dir:
+        result_path = Path(result_dir) / result_filename
+        with open(result_path, "w", encoding="utf-8") as file:
+            json.dump(merged, file, indent=2, ensure_ascii=False)
+
+    return merged

tests/dfx/stability/scripts/test_benchmark_stability.py

@@ -1,297 +0,0 @@
-"""
-Stability test cases: start OmniServer first, then run benchmark traffic with either
-`request-rate` or `max-concurrency` for a fixed duration. No new requests are sent
-after the duration is reached, and the test asserts that there are no failed requests.
-
-The overall flow matches the perf logic: `load_configs`, `modify_stage`,
-`create_unique_server_params`, `create_test_parameter_mapping`,
-`get_benchmark_params_for_server`, `create_benchmark_indices`, and the
-`omni_server` fixture are aligned with perf. Only the benchmark execution
-(`run_stability_benchmark`, which is duration-based here) and the test cases differ.
-
-All test-specific parameters, such as `duration_sec`, `request_rate` /
-`max_concurrency`, and `num_prompts_per_batch`, are configured in
-`tests/dfx/stability/tests/test.json` and are no longer overridden
-through environment variables.
-"""
-
-import json
-import os
-import threading
-import time
-from pathlib import Path
-from typing import Any
-
-import pytest
-
-from tests.dfx.conftest import (
-    create_benchmark_indices,
-    create_test_parameter_mapping,
-    create_unique_server_params,
-    get_benchmark_params_for_server,
-    load_configs,
-)
-from tests.dfx.perf.scripts.run_benchmark import run_benchmark
-from tests.helpers.runtime import OmniServer
-
-STABILITY_DIR = Path(__file__).resolve().parent.parent
-DEPLOY_CONFIGS_DIR = STABILITY_DIR / "deploy"
-CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test.json")
-DEFAULT_NUM_PROMPTS_PER_BATCH = 20
-
-
-try:
-    BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
-except FileNotFoundError:
-    BENCHMARK_CONFIGS = []
-
-test_params = create_unique_server_params(BENCHMARK_CONFIGS, DEPLOY_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
-server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
-
-_omni_server_lock = threading.Lock()
-
-
-benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
-
-
-def _build_base_args(params: dict[str, Any], host: str, port: int) -> list[str]:
-    exclude = {
-        "request_rate",
-        "max_concurrency",
-        "num_prompts",
-        "baseline",
-        "duration_sec",
-        "num_prompts_per_batch",
-    }
-    args = ["--host", host, "--port", str(port)]
-    for key, value in params.items():
-        if key in exclude or value is None:
-            continue
-        arg_name = f"--{key.replace('_', '-')}"
-        if isinstance(value, bool) and value:
-            args.append(arg_name)
-        elif isinstance(value, dict):
-            args.extend([arg_name, json.dumps(value, ensure_ascii=False, separators=(",", ":"))])
-        elif not isinstance(value, bool):
-            args.extend([arg_name, str(value)])
-    return args
-
-
-def _run_one_benchmark_batch(
-    host: str,
-    port: int,
-    params: dict[str, Any],
-    num_prompts: int,
-    request_rate: float | None,
-    max_concurrency: int | None,
-    result_dir: str,
-    batch_index: int,
-) -> dict[str, Any]:
-    base = _build_base_args(params, host, port)
-    if request_rate is not None:
-        args = base + ["--request-rate", str(request_rate), "--num-prompts", str(num_prompts)]
-        flow = request_rate
-    else:
-        args = base + [
-            "--max-concurrency",
-            str(max_concurrency),
-            "--num-prompts",
-            str(num_prompts),
-            "--request-rate",
-            "inf",
-        ]
-        flow = max_concurrency
-
-    dataset_name = params.get("dataset_name", "random")
-    old_benchmark_dir = os.environ.get("BENCHMARK_DIR")
-    try:
-        os.environ["BENCHMARK_DIR"] = result_dir
-        result = run_benchmark(
-            args=args,
-            test_name="stability",
-            flow=flow,
-            dataset_name=dataset_name,
-            num_prompt=num_prompts,
-            random_input_len=params.get("random_input_len"),
-            random_output_len=params.get("random_output_len"),
-        )
-        return result
-    except (FileNotFoundError, OSError) as e:
-        # Surface batch failure so the stability test does not false-pass when
-        # run_benchmark fails before writing JSON (e.g. command not found).
-        return {
-            "completed": 0,
-            "failed": 1,
-            "duration": 0.0,
-            "errors": [f"Benchmark batch failed: {type(e).__name__}: {e}"],
-        }
-    finally:
-        if old_benchmark_dir is not None:
-            os.environ["BENCHMARK_DIR"] = old_benchmark_dir
-        elif "BENCHMARK_DIR" in os.environ:
-            os.environ.pop("BENCHMARK_DIR")
-
-
-def _merge_batch_results(batch_results: list[dict[str, Any]], total_duration_sec: float) -> dict[str, Any]:
-    if not batch_results:
-        return {"completed": 0, "failed": 0, "duration": total_duration_sec, "errors": []}
-
-    completed = sum(r.get("completed", 0) for r in batch_results)
-    failed = sum(r.get("failed", 0) for r in batch_results)
-    merged: dict[str, Any] = {
-        "completed": completed,
-        "failed": failed,
-        "duration": total_duration_sec,
-        "errors": [],
-    }
-    for r in batch_results:
-        merged["errors"].extend(r.get("errors") or [])
-    return merged
-
-
-def _print_merged_report(result: dict[str, Any]) -> None:
-    """Print the final summary: successful requests, failed requests, and total duration only."""
-    fmt = "{:<40} {:<10}"
-    fmt_float = "{:<40} {:<10.2f}"
-    completed = result.get("completed", 0)
-    failed = result.get("failed", 0)
-    duration = float(result.get("duration", 0.0) or 0.0)
-    print("\n============ Stability Benchmark Summary ============")
-    print(fmt.format("Successful requests:", completed))
-    print(fmt.format("Failed requests:", failed))
-    print(fmt_float.format("Total duration (s):", duration))
-    print("==================================================\n")
-
-
-def run_stability_benchmark(
-    host: str,
-    port: int,
-    duration_sec: int | float,
-    params: dict[str, Any],
-    *,
-    request_rate: float | None = None,
-    max_concurrency: int | None = None,
-    result_filename: str | None = None,
-    result_dir: str = "./",
-    num_prompts_per_batch: int = DEFAULT_NUM_PROMPTS_PER_BATCH,
-) -> dict[str, Any]:
-    if (request_rate is None) == (max_concurrency is None):
-        raise ValueError("Exactly one of request_rate or max_concurrency must be specified")
-
-    start_time = time.perf_counter()
-    batch_results: list[dict[str, Any]] = []
-    batch_index = 0
-
-    while True:
-        if (time.perf_counter() - start_time) >= duration_sec:
-            break
-        result = _run_one_benchmark_batch(
-            host=host,
-            port=port,
-            params=params,
-            num_prompts=num_prompts_per_batch,
-            request_rate=request_rate,
-            max_concurrency=max_concurrency,
-            result_dir=result_dir,
-            batch_index=batch_index,
-        )
-        batch_results.append(result)
-        batch_index += 1
-        if (time.perf_counter() - start_time) >= duration_sec:
-            break
-
-    total_duration = time.perf_counter() - start_time
-    merged = _merge_batch_results(batch_results, total_duration)
-    _print_merged_report(merged)
-
-    if result_filename and result_dir:
-        result_path = Path(result_dir) / result_filename
-        with open(result_path, "w", encoding="utf-8") as f:
-            json.dump(merged, f, indent=2, ensure_ascii=False)
-
-    return merged
-
-
-@pytest.fixture(scope="module")
-def omni_server(request):
-    """Start vLLM-Omni server as a subprocess with actual model weights.
-    Uses session scope so the server starts only once for the entire test session.
-    Multi-stage initialization can take 10-20+ minutes.
-    """
-    with _omni_server_lock:
-        test_name, model, stage_config_path, stage_overrides, extra_cli_args = request.param
-
-        print(f"Starting OmniServer with test: {test_name}, model: {model}")
-
-        server_args = ["--stage-init-timeout", "120"]
-        # --deploy-config and --stage-overrides compose at the CLI (see vllm_omni/entrypoints/utils.py):
-        # deploy-config sets the base; stage-overrides are applied on top. Both can be set.
-        if stage_config_path:
-            server_args = ["--deploy-config", stage_config_path] + server_args
-        if stage_overrides:
-            server_args = ["--stage-overrides", stage_overrides] + server_args
-        if extra_cli_args:
-            server_args = list(extra_cli_args) + server_args
-        with OmniServer(model, server_args) as server:
-            server.test_name = test_name
-            print("OmniServer started successfully")
-            yield server
-            print("OmniServer stopping...")
-
-        print("OmniServer stopped")
-
-
-@pytest.fixture(params=benchmark_indices)
-def stability_benchmark_params(request, omni_server):
-    """Benchmark parameters fixture with proper parametrization (same as perf)."""
-    test_name, param_index = request.param
-
-    if test_name != omni_server.test_name:
-        pytest.skip(f"Skipping parameter for {test_name} - current server is {omni_server.test_name}")
-
-    all_params = get_benchmark_params_for_server(test_name, server_to_benchmark_mapping)
-
-    if not all_params:
-        raise ValueError(f"No benchmark parameters found for test: {test_name}")
-
-    if param_index >= len(all_params):
-        raise ValueError(f"No benchmark parameters found for index {param_index} in test: {test_name}")
-
-    current = param_index + 1
-    total = len(all_params)
-    print(f"\n  Running benchmark {current}/{total} for {test_name}")
-
-    return {"test_name": test_name, "params": all_params[param_index]}
-
-
-@pytest.mark.slow
-@pytest.mark.parametrize("omni_server", test_params, indirect=True)
-@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
-def test_benchmark_stability(omni_server, stability_benchmark_params):
-    """Run the benchmark for a fixed duration using request-rate or max-concurrency and assert zero failed requests."""
-    test_name = stability_benchmark_params["test_name"]
-    params = stability_benchmark_params["params"]
-    duration_sec = params.get("duration_sec", 300)
-    num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
-    request_rate = params.get("request_rate")
-    max_concurrency = params.get("max_concurrency")
-
-    bench_params = {
-        k: v
-        for k, v in params.items()
-        if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
-    }
-
-    result = run_stability_benchmark(
-        host=omni_server.host,
-        port=omni_server.port,
-        duration_sec=duration_sec,
-        params=bench_params,
-        request_rate=request_rate,
-        max_concurrency=max_concurrency,
-        result_dir=str(STABILITY_DIR),
-        num_prompts_per_batch=num_prompts_per_batch,
-    )
-
-    assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
-    assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"

tests/dfx/stability/scripts/test_stability_qwen3_omni.py

@@ -0,0 +1,68 @@
+"""
+Qwen3-Omni stability: OmniServer + ``vllm bench serve --omni`` for a fixed duration.
+
+Configuration: ``tests/dfx/stability/tests/test_qwen3_omni.json``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.dfx.conftest import (
+    create_benchmark_indices,
+    create_test_parameter_mapping,
+    create_unique_server_params,
+    load_configs,
+)
+from tests.dfx.stability.helpers import _run_one_vllm_bench_batch, run_stability_benchmark_loop
+
+STABILITY_DIR = Path(__file__).resolve().parent.parent
+DEPLOY_CONFIGS_DIR = STABILITY_DIR / "deploy"
+CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test_qwen3_omni.json")
+DEFAULT_NUM_PROMPTS_PER_BATCH = 20
+STABILITY_SERVER_TIMEOUT_ARGS = ["--stage-init-timeout", "600"]
+
+try:
+    BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
+except FileNotFoundError:
+    BENCHMARK_CONFIGS = []
+
+test_params = create_unique_server_params(BENCHMARK_CONFIGS, DEPLOY_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
+server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
+benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("omni_server", test_params, indirect=True)
+@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
+def test_stability_qwen3_omni(omni_server, stability_benchmark_params):
+    test_name = stability_benchmark_params["test_name"]
+    params = stability_benchmark_params["params"]
+    duration_sec = params.get("duration_sec", 300)
+    num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
+    request_rate = params.get("request_rate")
+    max_concurrency = params.get("max_concurrency")
+
+    bench_params = {
+        k: v
+        for k, v in params.items()
+        if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
+    }
+
+    result = run_stability_benchmark_loop(
+        host=omni_server.host,
+        port=omni_server.port,
+        model=omni_server.model,
+        duration_sec=duration_sec,
+        params=bench_params,
+        request_rate=request_rate,
+        max_concurrency=max_concurrency,
+        result_dir=str(STABILITY_DIR),
+        num_prompts_per_batch=num_prompts_per_batch,
+        run_one_batch=_run_one_vllm_bench_batch,
+    )
+
+    assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
+    assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"

tests/dfx/stability/scripts/test_stability_qwen3_tts.py

@@ -0,0 +1,68 @@
+"""
+Qwen3-TTS stability: OmniServer + ``vllm bench serve --omni`` for a fixed duration.
+
+Configuration: ``tests/dfx/stability/tests/test_qwen3_tts.json``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.dfx.conftest import (
+    create_benchmark_indices,
+    create_test_parameter_mapping,
+    create_unique_server_params,
+    load_configs,
+)
+from tests.dfx.stability.helpers import _run_one_vllm_bench_batch, run_stability_benchmark_loop
+
+STABILITY_DIR = Path(__file__).resolve().parent.parent
+DEPLOY_CONFIGS_DIR = STABILITY_DIR / "deploy"
+CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test_qwen3_tts.json")
+DEFAULT_NUM_PROMPTS_PER_BATCH = 20
+STABILITY_SERVER_TIMEOUT_ARGS = ["--stage-init-timeout", "600"]
+
+try:
+    BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
+except FileNotFoundError:
+    BENCHMARK_CONFIGS = []
+
+test_params = create_unique_server_params(BENCHMARK_CONFIGS, DEPLOY_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
+server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
+benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("omni_server", test_params, indirect=True)
+@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
+def test_stability_qwen3_tts(omni_server, stability_benchmark_params):
+    test_name = stability_benchmark_params["test_name"]
+    params = stability_benchmark_params["params"]
+    duration_sec = params.get("duration_sec", 300)
+    num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
+    request_rate = params.get("request_rate")
+    max_concurrency = params.get("max_concurrency")
+
+    bench_params = {
+        k: v
+        for k, v in params.items()
+        if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
+    }
+
+    result = run_stability_benchmark_loop(
+        host=omni_server.host,
+        port=omni_server.port,
+        model=omni_server.model,
+        duration_sec=duration_sec,
+        params=bench_params,
+        request_rate=request_rate,
+        max_concurrency=max_concurrency,
+        result_dir=str(STABILITY_DIR),
+        num_prompts_per_batch=num_prompts_per_batch,
+        run_one_batch=_run_one_vllm_bench_batch,
+    )
+
+    assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
+    assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"

tests/dfx/stability/scripts/test_stability_qwen_image.py

@@ -0,0 +1,68 @@
+"""
+Qwen-Image stability: OmniServer (diffusion) + ``diffusion_benchmark_serving.py``.
+
+Configuration: ``tests/dfx/stability/tests/test_qwen_image.json``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.dfx.conftest import (
+    create_benchmark_indices,
+    create_test_parameter_mapping,
+    create_unique_server_params,
+    load_configs,
+)
+from tests.dfx.stability.helpers import _run_one_diffusion_batch, run_stability_benchmark_loop
+
+STABILITY_DIR = Path(__file__).resolve().parent.parent
+DEPLOY_CONFIGS_DIR = STABILITY_DIR / "deploy"
+CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test_qwen_image.json")
+DEFAULT_NUM_PROMPTS_PER_BATCH = 20
+STABILITY_SERVER_TIMEOUT_ARGS = ["--stage-init-timeout", "600", "--init-timeout", "900"]
+
+try:
+    BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
+except FileNotFoundError:
+    BENCHMARK_CONFIGS = []
+
+test_params = create_unique_server_params(BENCHMARK_CONFIGS, DEPLOY_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
+server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
+benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("omni_server", test_params, indirect=True)
+@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
+def test_stability_qwen_image(omni_server, stability_benchmark_params):
+    test_name = stability_benchmark_params["test_name"]
+    params = stability_benchmark_params["params"]
+    duration_sec = params.get("duration_sec", 300)
+    num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
+    request_rate = params.get("request_rate")
+    max_concurrency = params.get("max_concurrency")
+
+    bench_params = {
+        k: v
+        for k, v in params.items()
+        if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
+    }
+
+    result = run_stability_benchmark_loop(
+        host=omni_server.host,
+        port=omni_server.port,
+        model=omni_server.model,
+        duration_sec=duration_sec,
+        params=bench_params,
+        request_rate=request_rate,
+        max_concurrency=max_concurrency,
+        result_dir=str(STABILITY_DIR),
+        num_prompts_per_batch=num_prompts_per_batch,
+        run_one_batch=_run_one_diffusion_batch,
+    )
+
+    assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
+    assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"

tests/dfx/stability/scripts/test_stability_wan22.py

@@ -0,0 +1,68 @@
+"""
+Wan2.2 T2V stability: OmniServer (diffusion) + ``diffusion_benchmark_serving.py`` / ``v1/videos``.
+
+Configuration: ``tests/dfx/stability/tests/test_wan22.json``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.dfx.conftest import (
+    create_benchmark_indices,
+    create_test_parameter_mapping,
+    create_unique_server_params,
+    load_configs,
+)
+from tests.dfx.stability.helpers import _run_one_diffusion_batch, run_stability_benchmark_loop
+
+STABILITY_DIR = Path(__file__).resolve().parent.parent
+DEPLOY_CONFIGS_DIR = STABILITY_DIR / "deploy"
+CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test_wan22.json")
+DEFAULT_NUM_PROMPTS_PER_BATCH = 20
+STABILITY_SERVER_TIMEOUT_ARGS = ["--stage-init-timeout", "600", "--init-timeout", "900"]
+
+try:
+    BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
+except FileNotFoundError:
+    BENCHMARK_CONFIGS = []
+
+test_params = create_unique_server_params(BENCHMARK_CONFIGS, DEPLOY_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
+server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
+benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
+
+
+@pytest.mark.slow
+@pytest.mark.parametrize("omni_server", test_params, indirect=True)
+@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
+def test_stability_wan22(omni_server, stability_benchmark_params):
+    test_name = stability_benchmark_params["test_name"]
+    params = stability_benchmark_params["params"]
+    duration_sec = params.get("duration_sec", 300)
+    num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
+    request_rate = params.get("request_rate")
+    max_concurrency = params.get("max_concurrency")
+
+    bench_params = {
+        k: v
+        for k, v in params.items()
+        if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
+    }
+
+    result = run_stability_benchmark_loop(
+        host=omni_server.host,
+        port=omni_server.port,
+        model=omni_server.model,
+        duration_sec=duration_sec,
+        params=bench_params,
+        request_rate=request_rate,
+        max_concurrency=max_concurrency,
+        result_dir=str(STABILITY_DIR),
+        num_prompts_per_batch=num_prompts_per_batch,
+        run_one_batch=_run_one_diffusion_batch,
+    )
+
+    assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
+    assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"

tests/dfx/stability/tests/test.json

@@ -1,77 +0,0 @@
-[
-    {
-        "test_name": "test_qwen3_omni_stability",
-        "server_params": {
-            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
-            "stage_overrides": {
-                "2": {
-                    "max_num_batched_tokens": 1000000
-                }
-            }
-        },
-        "benchmark_params": [
-            {
-                "dataset_name": "random",
-                "backend": "openai-chat-omni",
-                "endpoint": "/v1/chat/completions",
-                "duration_sec": 300,
-                "request_rate": 0.2,
-                "num_prompts_per_batch": 10,
-                "random_input_len": 2500,
-                "random_output_len": 900,
-                "ignore_eos": true,
-                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
-            },
-            {
-                "dataset_name": "random",
-                "backend": "openai-chat-omni",
-                "endpoint": "/v1/chat/completions",
-                "duration_sec": 300,
-                "max_concurrency": 2,
-                "num_prompts_per_batch": 10,
-                "random_input_len": 2500,
-                "random_output_len": 900,
-                "ignore_eos": true,
-                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
-            }
-        ]
-    },
-    {
-        "test_name": "test_qwen3_omni_stability_async_chunk",
-        "server_params": {
-            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
-            "stage_overrides": {
-                "2": {
-                    "max_num_batched_tokens": 1000000
-                }
-            },
-            "extra_cli_args": ["--async-chunk"]
-        },
-        "benchmark_params": [
-            {
-                "dataset_name": "random",
-                "backend": "openai-chat-omni",
-                "endpoint": "/v1/chat/completions",
-                "duration_sec": 300,
-                "request_rate": 0.2,
-                "num_prompts_per_batch": 10,
-                "random_input_len": 2500,
-                "random_output_len": 900,
-                "ignore_eos": true,
-                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
-            },
-            {
-                "dataset_name": "random",
-                "backend": "openai-chat-omni",
-                "endpoint": "/v1/chat/completions",
-                "duration_sec": 300,
-                "max_concurrency": 2,
-                "num_prompts_per_batch": 10,
-                "random_input_len": 2500,
-                "random_output_len": 900,
-                "ignore_eos": true,
-                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
-            }
-        ]
-    }
-]

tests/dfx/stability/tests/test_qwen3_omni.json

@@ -0,0 +1,97 @@
+[
+    {
+        "test_name": "test_qwen3_omni_stability",
+        "server_params": {
+            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+            "stage_overrides": {
+                "2": {
+                    "max_num_batched_tokens": 1000000
+                }
+            }
+        },
+        "benchmark_params": [
+            {
+                "dataset_name": "random-mm",
+                "backend": "openai-chat-omni",
+                "endpoint": "/v1/chat/completions",
+                "duration_sec": 86400,
+                "request_rate": 0.3,
+                "num_prompts_per_batch": 10,
+                "random_input_len": {
+                    "min": 0,
+                    "max": 8000
+                },
+                "random_output_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_range_ratio": 0.0,
+                "random_mm_base_items_per_request": {
+                    "min": 0,
+                    "max": 6
+                },
+                "random_mm_num_mm_items_range_ratio": 0.0,
+                "random_mm_limit_mm_per_prompt": {
+                    "image": 2,
+                    "video": 2,
+                    "audio": 2
+                },
+                "random_mm_bucket_config": {
+                    "(128-1024, 128-1024, 1)": 0.34,
+                    "(256-1080, 256-1920, 2-16)": 0.33,
+                    "(0, 1-60, 1-3)": 0.33
+                },
+                "ignore_eos": true,
+                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
+            }
+        ]
+    },
+    {
+        "test_name": "test_qwen3_omni_stability_async_chunk",
+        "server_params": {
+            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+            "stage_overrides": {
+                "2": {
+                    "max_num_batched_tokens": 1000000
+                }
+            },
+            "extra_cli_args": ["--async-chunk"]
+        },
+        "benchmark_params": [
+            {
+                "dataset_name": "random-mm",
+                "backend": "openai-chat-omni",
+                "endpoint": "/v1/chat/completions",
+                "duration_sec": 86400,
+                "max_concurrency": 2,
+                "num_prompts_per_batch": 10,
+                "random_input_len": {
+                    "min": 0,
+                    "max": 8000
+                },
+                "random_output_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_range_ratio": 0.0,
+                "random_mm_base_items_per_request": {
+                    "min": 0,
+                    "max": 6
+                },
+                "random_mm_num_mm_items_range_ratio": 0.0,
+                "random_mm_limit_mm_per_prompt": {
+                    "image": 2,
+                    "video": 2,
+                    "audio": 2
+                },
+                "random_mm_bucket_config": {
+                    "(128-1024, 128-1024, 1)": 0.34,
+                    "(256-1080, 256-1920, 2-16)": 0.33,
+                    "(0, 1-60, 1-3)": 0.33
+                },
+                "ignore_eos": true,
+                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
+            }
+        ]
+    }
+]

tests/dfx/stability/tests/test_qwen3_tts.json

@@ -0,0 +1,56 @@
+[
+    {
+        "test_name": "test_qwen3_tts_stability",
+        "server_params": {
+            "model": "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
+        },
+        "benchmark_params": [
+            {
+                "dataset_name": "random",
+                "backend": "openai-audio-speech",
+                "endpoint": "/v1/audio/speech",
+                "duration_sec": 86400,
+                "request_rate": 0.3,
+                "num_prompts_per_batch": 10,
+                "random_input_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_output_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_range_ratio": 0.0,
+                "extra_body": {
+                    "voice": "Vivian",
+                    "language": "English"
+                },
+                "ignore_eos": true,
+                "percentile-metrics": "ttft,e2el,audio_rtf,audio_ttfp,audio_duration"
+            },
+            {
+                "dataset_name": "random",
+                "backend": "openai-audio-speech",
+                "endpoint": "/v1/audio/speech",
+                "duration_sec": 86400,
+                "max_concurrency": 2,
+                "num_prompts_per_batch": 10,
+                "random_input_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_output_len": {
+                    "min": 0,
+                    "max": 1000
+                },
+                "random_range_ratio": 0.0,
+                "extra_body": {
+                    "voice": "Vivian",
+                    "language": "English"
+                },
+                "ignore_eos": true,
+                "percentile-metrics": "ttft,e2el,audio_rtf,audio_ttfp,audio_duration"
+            }
+        ]
+    }
+]

tests/dfx/stability/tests/test_qwen_image.json

@@ -0,0 +1,28 @@
+[
+    {
+        "test_name": "test_qwen_image_stability",
+        "server_params": {
+            "model": "Qwen/Qwen-Image"
+        },
+        "benchmark_params": [
+            {
+                "dataset": "random",
+                "task": "t2i",
+                "backend": "vllm-omni",
+                "duration_sec": 86400,
+                "max_concurrency": 1,
+                "num_prompts_per_batch": 10,
+                "width": {
+                    "min": 512,
+                    "max": 2048
+                },
+                "height": {
+                    "min": 512,
+                    "max": 2048
+                },
+                "num_inference_steps": 50,
+                "enable_negative_prompt": true
+            }
+        ]
+    }
+]

tests/dfx/stability/tests/test_wan22.json

@@ -0,0 +1,31 @@
+[
+    {
+        "test_name": "test_wan22_i2v_stability_v1_videos",
+        "server_params": {
+            "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers",
+            "serve_args": {
+                "ulysses-degree": 2,
+                "vae-patch-parallel-size": 2,
+                "tensor-parallel-size": 1,
+                "use-hsdp": true,
+                "vae-use-slicing": true,
+                "vae-use-tiling": true
+            }
+        },
+        "benchmark_params": [
+            {
+                "dataset": "random",
+                "task": "i2v",
+                "backend": "v1/videos",
+                "duration_sec": 86400,
+                "max_concurrency": 1,
+                "num_prompts_per_batch": 20,
+                "enable_negative_prompt": true,
+                "random_request_config": [
+                    {"width": 832, "height": 480, "num_inference_steps": 4, "num_frames": 81, "fps": 16, "weight": 0.5},
+                    {"width": 1280, "height": 720, "num_inference_steps": 4, "num_frames": 121, "fps": 16, "weight": 0.5}
+                ]
+            }
+        ]
+    }
+]