diff --git a/docs/contributing/ci/CI_5levels.md b/docs/contributing/ci/CI_5levels.md
index 7bad897e7..3baa7ff88 100644
--- a/docs/contributing/ci/CI_5levels.md
+++ b/docs/contributing/ci/CI_5levels.md
@@ -105,7 +105,8 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
| Depends on reality |
Stability:
- /tests/dfx/stability/tests/test.json
+ /tests/dfx/stability/tests/test_qwen3_omni.json
+ /tests/dfx/stability/tests/test_wan22.json
Reliability:
tests/e2e/reliability/test_{model_name}.py
|
@@ -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
Test Examples
-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.
-- - ***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_.py` alongside a matching JSON config)
- ***Reliability***: `pytest -s -v tests/e2e/reliability/test_{model_name}.py`
## Summary
diff --git a/docs/contributing/ci/test_guide.md b/docs/contributing/ci/test_guide.md
index 0c9ec2fb3..018c47b05 100644
--- a/docs/contributing/ci/test_guide.md
+++ b/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
diff --git a/tests/dfx/conftest.py b/tests/dfx/conftest.py
index c3f6d0a15..607570443 100644
--- a/tests/dfx/conftest.py
+++ b/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(
diff --git a/tests/dfx/stability/conftest.py b/tests/dfx/stability/conftest.py
index e36c88b9a..30718d4bf 100644
--- a/tests/dfx/stability/conftest.py
+++ b/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)
diff --git a/tests/dfx/stability/helpers.py b/tests/dfx/stability/helpers.py
index 3956bd213..3a873f69c 100644
--- a/tests/dfx/stability/helpers.py
+++ b/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
diff --git a/tests/dfx/stability/scripts/test_benchmark_stability.py b/tests/dfx/stability/scripts/test_benchmark_stability.py
deleted file mode 100644
index 620241762..000000000
--- a/tests/dfx/stability/scripts/test_benchmark_stability.py
+++ /dev/null
@@ -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"
diff --git a/tests/dfx/stability/scripts/test_stability_qwen3_omni.py b/tests/dfx/stability/scripts/test_stability_qwen3_omni.py
new file mode 100644
index 000000000..d1c2af8cf
--- /dev/null
+++ b/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"
diff --git a/tests/dfx/stability/scripts/test_stability_qwen3_tts.py b/tests/dfx/stability/scripts/test_stability_qwen3_tts.py
new file mode 100644
index 000000000..beccd67d9
--- /dev/null
+++ b/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"
diff --git a/tests/dfx/stability/scripts/test_stability_qwen_image.py b/tests/dfx/stability/scripts/test_stability_qwen_image.py
new file mode 100644
index 000000000..a90e2092f
--- /dev/null
+++ b/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"
diff --git a/tests/dfx/stability/scripts/test_stability_wan22.py b/tests/dfx/stability/scripts/test_stability_wan22.py
new file mode 100644
index 000000000..afe9c4d0c
--- /dev/null
+++ b/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"
diff --git a/tests/dfx/stability/tests/test.json b/tests/dfx/stability/tests/test.json
deleted file mode 100644
index 255cd5b10..000000000
--- a/tests/dfx/stability/tests/test.json
+++ /dev/null
@@ -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"
- }
- ]
- }
-]
diff --git a/tests/dfx/stability/tests/test_qwen3_omni.json b/tests/dfx/stability/tests/test_qwen3_omni.json
new file mode 100644
index 000000000..a16ab805c
--- /dev/null
+++ b/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"
+ }
+ ]
+ }
+]
diff --git a/tests/dfx/stability/tests/test_qwen3_tts.json b/tests/dfx/stability/tests/test_qwen3_tts.json
new file mode 100644
index 000000000..fbf30d88a
--- /dev/null
+++ b/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"
+ }
+ ]
+ }
+]
diff --git a/tests/dfx/stability/tests/test_qwen_image.json b/tests/dfx/stability/tests/test_qwen_image.json
new file mode 100644
index 000000000..f3dd93f6f
--- /dev/null
+++ b/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
+ }
+ ]
+ }
+]
diff --git a/tests/dfx/stability/tests/test_wan22.json b/tests/dfx/stability/tests/test_wan22.json
new file mode 100644
index 000000000..c787ce96a
--- /dev/null
+++ b/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}
+ ]
+ }
+ ]
+ }
+]