diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml
index 00823951d..96aa9e8ba 100644
--- a/.buildkite/pipeline.yml
+++ b/.buildkite/pipeline.yml
@@ -1,8 +1,9 @@
# Document 1: Buildkite loads only this block on first parse. The next step resolves docs-only skip-ci
-# from git diff, then uploads document 2. When docs-only skip applies, image-build still runs if nightly-test
-# / main NIGHTLY so upload-nightly is not skipped together with test-ready/test-merge.
-#
-# Document 2: appended after `---`; same file, read by upload_pipeline_with_skip_ci.sh (not evaluated as a second pipeline by Buildkite).
+# from git diff, then uploads document 2. When docs-only skip applies, image-build still runs for the nightly
+# exception in upload_pipeline_with_skip_ci.sh (nightly-test label or main NIGHTLY=1); otherwise upload-nightly
+# can be skipped together with test-ready/test-merge. Weekly uploads also depend on image-build.
+
+# Document 2: appended after `---`; same file, read by upload_pipeline_with_skip_ci.sh (not evaluated as a second pipeline by Buildkite). Child uploads include test-ready.yml, test-merge.yml, test-nightly.yml, and test-weekly.yml.
steps:
- label: ":github: Resolve skip-ci & upload pipeline"
key: upload-ci-pipeline
@@ -61,3 +62,15 @@ steps:
- buildkite-agent pipeline upload .buildkite/test-nightly.yml
agents:
queue: "cpu_queue_premerge"
+
+ # L5 Test — main+WEEKLY=1 (scheduled), or PR with weekly-test label
+ - label: "Upload Weekly Pipeline"
+ depends_on: image-build
+ key: upload-weekly-pipeline
+ if: >-
+ (build.branch == "main" && build.env("WEEKLY") == "1") ||
+ (build.branch != "main" && build.pull_request.labels includes "weekly-test")
+ commands:
+ - buildkite-agent pipeline upload .buildkite/test-weekly.yml
+ agents:
+ queue: "cpu_queue_premerge"
diff --git a/.buildkite/test-weekly.yml b/.buildkite/test-weekly.yml
new file mode 100644
index 000000000..dba648996
--- /dev/null
+++ b/.buildkite/test-weekly.yml
@@ -0,0 +1,85 @@
+env:
+ VLLM_WORKER_MULTIPROC_METHOD: spawn
+ HF_HUB_DOWNLOAD_TIMEOUT: 300
+ HF_HUB_ETAG_TIMEOUT: 60
+
+steps:
+ - label: "Reliability Test - qwen3-omni"
+ timeout_in_minutes: 180
+ depends_on: upload-weekly-pipeline
+ if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test"
+ commands:
+ - pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m "slow"
+ agents:
+ queue: "mithril-h100-pool"
+ plugins:
+ - kubernetes:
+ podSpec:
+ containers:
+ - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+ resources:
+ limits:
+ nvidia.com/gpu: 2
+ volumeMounts:
+ - name: devshm
+ mountPath: /dev/shm
+ - name: hf-cache
+ mountPath: /root/.cache/huggingface
+ env:
+ - name: HF_HOME
+ value: /root/.cache/huggingface
+ - name: HF_TOKEN
+ valueFrom:
+ secretKeyRef:
+ name: hf-token-secret
+ key: token
+ nodeSelector:
+ node.kubernetes.io/instance-type: gpu-h100-sxm
+ volumes:
+ - name: devshm
+ emptyDir:
+ medium: Memory
+ - name: hf-cache
+ hostPath:
+ path: /mnt/hf-cache
+ type: DirectoryOrCreate
+
+ - label: "Reliability Test - wan22"
+ timeout_in_minutes: 180
+ depends_on: upload-weekly-pipeline
+ if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test"
+ commands:
+ - pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m "slow"
+ agents:
+ queue: "mithril-h100-pool"
+ plugins:
+ - kubernetes:
+ podSpec:
+ containers:
+ - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+ resources:
+ limits:
+ nvidia.com/gpu: 2
+ volumeMounts:
+ - name: devshm
+ mountPath: /dev/shm
+ - name: hf-cache
+ mountPath: /root/.cache/huggingface
+ env:
+ - name: HF_HOME
+ value: /root/.cache/huggingface
+ - name: HF_TOKEN
+ valueFrom:
+ secretKeyRef:
+ name: hf-token-secret
+ key: token
+ nodeSelector:
+ node.kubernetes.io/instance-type: gpu-h100-sxm
+ volumes:
+ - name: devshm
+ emptyDir:
+ medium: Memory
+ - name: hf-cache
+ hostPath:
+ path: /mnt/hf-cache
+ type: DirectoryOrCreate
diff --git a/docs/contributing/ci/CI_5levels.md b/docs/contributing/ci/CI_5levels.md
index 3baa7ff88..6c4519362 100644
--- a/docs/contributing/ci/CI_5levels.md
+++ b/docs/contributing/ci/CI_5levels.md
@@ -108,7 +108,8 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
/tests/dfx/stability/tests/test_qwen3_omni.json
/tests/dfx/stability/tests/test_wan22.json
Reliability:
- tests/e2e/reliability/test_{model_name}.py
+ tests/dfx/reliability/test_reliability_{model_key}.py
+ (e.g. test_reliability_qwen3_omni.py, test_reliability_wan22.py)
Chapter 4
@@ -576,12 +577,17 @@ 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 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.
+- ***Reliability Testing***: Uses pytest suites under `tests/dfx/reliability/` to inject controlled faults against a **live** `vllm_omni serve` instance (same **`omni_server` / `omni_server_function`** fixture style as E2E). Current suites emphasize **GPU memory pressure** (CUDA sidecar “memory hog”), **worker / runtime process kill** (`SIGKILL` on `VLLM::Worker` for Qwen3-Omni or `multiprocessing.spawn` for Wan2.2 video workers), **large multimodal chat** or **`/v1/videos`** jobs under OOM, **`/health` → 503** and **fast-fail / non-hanging concurrent** requests after kill, and **OpenAI-style 5xx error contracts** (e.g. text vs text+audio under OOM). **Post-fault recovery** checks exist where enabled (some cases may be `skip` while issues are tracked). See the Reliability `` block in Section 4.4 for file-level responsibilities and CI markers (`slow`, `hardware_test`, POSIX-only kill).
### 4.3 Test Directory and Execution Files
- ***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`
+- ***Reliability Test Suite*** (`tests/dfx/reliability/`):
+ - `test_reliability_qwen3_omni.py` — Qwen3-Omni chat / multimodal reliability (GPU OOM, process kill, recovery, error contract under `--async-chunk` vs default).
+ - `test_reliability_wan22.py` — Wan2.2 T2V video API reliability (`/v1/videos` under OOM and process kill, recovery).
+ - `helpers.py` — Shared primitives used by current suites: raw HTTP probes for `/v1/chat/completions` and `/health`, OpenAI-style error parsing, GPU OOM sidecar (`inject_gpu_oom` / `stop_gpu_oom_hogs`), and `pgrep`-based process-kill injector construction (`make_process_kill_fault_injector`).
+ - `conftest.py` — `fault_injector` and `omni_server_after_fault` / `omni_server_after_fault_function` fixtures to run a callable **after** the server is ready.
+ - `README.md` — Short local run commands for this directory.
### 4.4 Execution Method and Example
@@ -662,8 +668,57 @@ All other optional parameters follow the same rules as the in Chapter 3.4.
+
+ Reliability test suite (tests/dfx/reliability)
+
+#### Purpose and relationship to stability
+
+Reliability tests are **short fault-injection** integration runs (L5 **(b)** in `tests/dfx/reliability/README.md`). They complement **stability** JSON-driven long runs: instead of hours of steady traffic, they **perturb** the server (GPU OOM sidecar, fatal signals on selected processes) and check **failure mode** and **latency bounds** (e.g. chat or `/v1/videos` must not hang under concurrent fault-time load).
+
+#### Directory layout
+
+| Path | Responsibility |
+| ---- | -------------- |
+| `helpers.py` | Shared helpers used by current reliability suites: raw `POST`/`GET` probes (`/v1/chat/completions`, `/health`), OpenAI error parsing (`extract_openai_error_contract_from_bytes`), GPU OOM sidecar lifecycle (`inject_gpu_oom`, `stop_gpu_oom_hogs`), and process-kill injector builder (`make_process_kill_fault_injector`). |
+| `conftest.py` | Pytest fixtures: indirect `fault_injector`, `omni_server_after_fault` / `omni_server_after_fault_function` (run injector after server is ready, then yield server). |
+| `test_reliability_qwen3_omni.py` | Qwen3-Omni: OOM vs **text vs text+audio** error contract, large multimodal chat under OOM, concurrent pressure, **SIGKILL** on `VLLM::Worker`, `/health` → 503 + fast-fail + concurrent chat; optional OOM recovery scenario (may be skipped while tracked in issues). |
+| `test_reliability_wan22.py` | Wan2.2 T2V: large `/v1/videos` under OOM, **SIGKILL** on `multiprocessing.spawn` chain, health / fast-fail / concurrent video requests; optional recovery test (may be skipped). |
+| `README.md` | Minimal run / collect examples. |
+
+#### Parametrization and markers
+
+- Each test module defines a **`RELIABILITY_SCENARIOS`** list (`test_name`, `server_params`: model, `stage_config_name` or diffusion `server_args`, etc.). **`create_reliability_omni_server_params()`** in `tests/dfx/conftest.py` resolves stage paths (including XPU substitutions where applicable) and builds **`OmniServerParams`** lists consumed by **`@pytest.mark.parametrize(..., indirect=True)`** on `omni_server` or `omni_server_function`.
+- Cases are tagged **`@pytest.mark.slow`** for weekly / selective CI. GPU-heavy suites use **`@hardware_test(res={"cuda": "H100"}, num_cards=...)`** (Qwen3-Omni paths often require **2** cards; Wan2.2 video paths **1** card).
+- **Process-kill** tests use **`@pytest.mark.skipif(os.name == "nt", ...)`** because injection uses POSIX **`pgrep` / `kill`**.
+
+#### CI trigger
+
+Weekly Buildkite (`.buildkite/test-weekly.yml`) runs, for example:
+
+```bash
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m "slow"
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m "slow"
+```
+
+#### Local commands
+
+```bash
+pytest --collect-only tests/dfx/reliability
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow
+```
+
+#### Adding a new model suite
+
+1. Add `test_reliability_.py` under `tests/dfx/reliability/`.
+2. Define **`RELIABILITY_SCENARIOS`** and pass them through **`create_reliability_omni_server_params()`** with the correct deploy or e2e stage-config directory (same pattern as existing files).
+3. Reuse **`helpers`** for OOM / kill / raw HTTP; prefer **`assert_fault_exception()`** and **`resolve_oom_device_spec()`** from `tests/dfx/conftest.py` for consistent device selection vs stage YAML.
+4. Register **`slow`** (and **`hardware_test`** if needed); extend **`.buildkite/test-weekly.yml`** when the suite should run in weekly L5.
+
+
+
- - ***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`
+ - ***Reliability***: `pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow` and/or `pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow` (add `test_reliability_.py` for new models)
## Summary
diff --git a/tests/dfx/conftest.py b/tests/dfx/conftest.py
index 12eb8e6f1..29bb33302 100644
--- a/tests/dfx/conftest.py
+++ b/tests/dfx/conftest.py
@@ -1,13 +1,18 @@
import json
import os
+import re
import subprocess
+import time
from datetime import datetime
from pathlib import Path
from typing import Any
import pytest
+from tests.dfx.reliability.helpers import list_remote_process_pids_by_pattern, post_chat_completions_raw
+from tests.helpers.runtime import OmniServerParams
from tests.helpers.stage_config import modify_stage_config
+from vllm_omni.platforms import current_omni_platform
def load_configs(config_path: str) -> list[dict[str, Any]]:
@@ -26,18 +31,15 @@ def load_configs(config_path: str) -> list[dict[str, Any]]:
raise RuntimeError(f"Failed to load configuration file: {str(e)}")
-def modify_stage(default_path, updates, deletes):
- kwargs = {}
+def modify_stage(default_path: str, updates: dict[str, Any] | None, deletes: dict[str, Any] | None) -> str:
+ kwargs: dict[str, Any] = {}
if updates is not None:
kwargs["updates"] = updates
if deletes is not None:
kwargs["deletes"] = deletes
if kwargs:
- path = modify_stage_config(default_path, **kwargs)
- else:
- path = default_path
-
- return path
+ return modify_stage_config(default_path, **kwargs)
+ return default_path
def _build_serve_args(serve_args: Any) -> list[str]:
@@ -111,6 +113,122 @@ def create_unique_server_params(
return unique_params
+def configs_with_platform_stage_configs(configs: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ """Resolve stage config for XPU vs CUDA/ROCm."""
+ out: list[dict[str, Any]] = []
+ for config in configs:
+ config_copy = json.loads(json.dumps(config))
+ if current_omni_platform.is_xpu():
+ config_copy["server_params"]["stage_config_name"] = "xpu/qwen3_omni_ci.yaml"
+ out.append(config_copy)
+ return out
+
+
+def extract_server_args_by_test_name(configs: list[dict[str, Any]]) -> dict[str, list[str] | None]:
+ mapping: dict[str, list[str] | None] = {}
+ for cfg in configs:
+ test_name = str(cfg.get("test_name"))
+ server_params = cfg.get("server_params") or {}
+ raw_args = server_params.get("server_args")
+ mapping[test_name] = [str(item) for item in raw_args] if isinstance(raw_args, list) else None
+ return mapping
+
+
+def create_reliability_omni_server_params(
+ configs: list[dict[str, Any]], stage_configs_dir: Path
+) -> list[OmniServerParams]:
+ adjusted_configs = configs_with_platform_stage_configs(configs)
+ unique_params = create_unique_server_params(adjusted_configs, stage_configs_dir)
+ server_args_by_name = extract_server_args_by_test_name(adjusted_configs)
+ return [
+ OmniServerParams(
+ model=model,
+ stage_config_path=stage_config_path,
+ server_args=server_args_by_name.get(test_name),
+ )
+ for test_name, model, stage_config_path, _stage_overrides_json, _extra_cli_args in unique_params
+ ]
+
+
+def supports_video_generation(model_name: str) -> bool:
+ lower = model_name.lower()
+ return any(key in lower for key in ("wan", "video", "i2v", "t2v"))
+
+
+def supports_chat_generation(model_name: str) -> bool:
+ return not supports_video_generation(model_name)
+
+
+def parse_stage_devices(stage_config_path: str) -> str:
+ text = Path(stage_config_path).read_text(encoding="utf-8")
+ raw_devices: list[str] = re.findall(r"^\s*devices:\s*\"?([0-9,\s]+)\"?\s*$", text, flags=re.MULTILINE)
+ devices: set[int] = set()
+ for item in raw_devices:
+ for token in item.split(","):
+ token = token.strip()
+ if token:
+ devices.add(int(token))
+ if not devices:
+ raise ValueError(f"No runtime.devices found in stage config: {stage_config_path}")
+ return ",".join(str(x) for x in sorted(devices))
+
+
+def resolve_oom_device_spec(config: dict[str, Any], stage_config_path: str | None) -> str:
+ explicit = config.get("device")
+ if explicit is not None:
+ return str(explicit)
+ if not stage_config_path:
+ return "0"
+ return parse_stage_devices(stage_config_path)
+
+
+def assert_fault_exception(exc: Exception, error_keywords: tuple[str, ...]) -> None:
+ text = str(exc).lower()
+ assert any(key in text for key in error_keywords), f"unexpected error under fault injection: {exc}"
+
+
+def wait_chat_request_ready(host: str, port: int, model: str, timeout_sec: int = 180) -> None:
+ """Poll a minimal chat request until success."""
+ deadline = time.time() + timeout_sec
+ payload = json.dumps(
+ {
+ "model": model,
+ "messages": [{"role": "user", "content": "Say hello in one short sentence."}],
+ "stream": False,
+ "modalities": ["text"],
+ }
+ )
+ last_error: str | None = None
+ while time.time() < deadline:
+ try:
+ status, body = post_chat_completions_raw(host, port, payload)
+ if status == 200:
+ return
+ last_error = f"http={status} body={body[:200]!r}"
+ except Exception as exc: # noqa: BLE001
+ last_error = str(exc)
+ time.sleep(2)
+ raise TimeoutError(f"runtime-teardown warmup request did not succeed within {timeout_sec}s: {last_error}")
+
+
+def assert_no_extra_worker_processes(
+ baseline_pids: set[int],
+ worker_pattern: str,
+ timeout_sec: int = 60,
+) -> None:
+ """Ensure no extra worker PIDs remain after teardown."""
+ deadline = time.time() + timeout_sec
+ while time.time() < deadline:
+ current = set(list_remote_process_pids_by_pattern(worker_pattern))
+ extra = current - baseline_pids
+ if not extra:
+ return
+ time.sleep(2)
+ current = set(list_remote_process_pids_by_pattern(worker_pattern))
+ extra = sorted(current - baseline_pids)
+ assert not extra, f"orphan worker processes remain after container teardown: {extra}"
+
+
def create_test_parameter_mapping(configs: list[dict[str, Any]]) -> dict[str, dict]:
mapping = {}
for config in configs:
diff --git a/tests/dfx/reliability/README.md b/tests/dfx/reliability/README.md
new file mode 100644
index 000000000..7196101bc
--- /dev/null
+++ b/tests/dfx/reliability/README.md
@@ -0,0 +1,14 @@
+# L5(b) Reliability (`tests/dfx/reliability`)
+
+This directory contains reliability fault-injection tests for key models.
+The detailed RFC document is maintained outside this repository (local/internal only).
+
+- Fault helpers: `conftest.py`, `helpers.py`
+- Qwen3 tests: `test_reliability_qwen3_omni.py`
+- Wan2.2 tests: `test_reliability_wan22.py`
+
+```bash
+pytest --collect-only tests/dfx/reliability
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow
+```
diff --git a/tests/dfx/reliability/conftest.py b/tests/dfx/reliability/conftest.py
new file mode 100644
index 000000000..7cf79847f
--- /dev/null
+++ b/tests/dfx/reliability/conftest.py
@@ -0,0 +1,29 @@
+"""Pytest fixtures for reliability tests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.dfx.reliability.helpers import FaultInjector
+
+
+@pytest.fixture
+def fault_injector(request: pytest.FixtureRequest) -> FaultInjector:
+ """Indirect only: ``request.param`` must be a ``FaultInjector`` callable."""
+ return request.param
+
+
+@pytest.fixture
+def omni_server_after_fault(omni_server: Any, fault_injector: FaultInjector):
+ """After ``omni_server`` is up, run ``fault_injector(omni_server)``, then yield the server."""
+ fault_injector(omni_server)
+ yield omni_server
+
+
+@pytest.fixture
+def omni_server_after_fault_function(omni_server_function: Any, fault_injector: FaultInjector):
+ """Inject fault after function-scoped server startup, then yield server."""
+ fault_injector(omni_server_function)
+ yield omni_server_function
diff --git a/tests/dfx/reliability/helpers.py b/tests/dfx/reliability/helpers.py
new file mode 100644
index 000000000..9ebd0ce97
--- /dev/null
+++ b/tests/dfx/reliability/helpers.py
@@ -0,0 +1,567 @@
+"""Shared reliability fault-injection helpers.
+
+This module keeps fault injection callable from tests directly:
+- GPU OOM (CUDA sidecar memory hog)
+- process kill by pattern and signal
+- post-ready hooks via ``fault_injector`` / ``omni_server_after_fault`` fixtures
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import logging
+import os
+import select
+import shlex
+import signal
+import subprocess
+import sys
+import time
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+import psutil
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class OomHandle:
+ """Handle for a started CUDA memory hog subprocess."""
+
+ proc: subprocess.Popen | None
+ device: int
+ target_mem_ratio: float
+ start_ts: float
+
+
+def post_chat_completions_raw(
+ host: str,
+ port: int,
+ body: bytes | str,
+ *,
+ content_type: str = "application/json",
+ timeout_sec: int = 120,
+) -> tuple[int, bytes]:
+ """POST /v1/chat/completions with raw bytes; returns (status, response_body)."""
+ conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+ try:
+ headers = {"Content-Type": content_type}
+ payload = body.encode("utf-8") if isinstance(body, str) else body
+ conn.request("POST", "/v1/chat/completions", body=payload, headers=headers)
+ resp = conn.getresponse()
+ data = resp.read()
+ return resp.status, data
+ finally:
+ conn.close()
+
+
+def get_health_raw(host: str, port: int, *, timeout_sec: int = 20) -> tuple[int, bytes]:
+ """GET /health with stdlib HTTP client; returns (status, body)."""
+ conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+ try:
+ conn.request("GET", "/health")
+ resp = conn.getresponse()
+ return resp.status, resp.read()
+ finally:
+ conn.close()
+
+
+def post_json_raw_http_client(
+ host: str,
+ port: int,
+ path: str,
+ payload: dict[str, Any],
+ *,
+ timeout_sec: int = 30,
+) -> tuple[int, bytes]:
+ """POST JSON to one endpoint with stdlib HTTP client; returns (status, body)."""
+ conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+ try:
+ body = json.dumps(payload).encode("utf-8")
+ conn.request("POST", path, body=body, headers={"Content-Type": "application/json"})
+ resp = conn.getresponse()
+ return resp.status, resp.read()
+ finally:
+ conn.close()
+
+
+def post_json_raw(
+ host: str,
+ port: int,
+ path: str,
+ payload: dict[str, Any],
+ *,
+ timeout_sec: int = 30,
+) -> tuple[int, bytes]:
+ """POST JSON to one endpoint; returns (status, body)."""
+ return (
+ post_chat_completions_raw(
+ host,
+ port,
+ json.dumps(payload),
+ content_type="application/json",
+ timeout_sec=timeout_sec,
+ )
+ if path == "/v1/chat/completions"
+ else post_json_raw_http_client(
+ host,
+ port,
+ path,
+ payload,
+ timeout_sec=timeout_sec,
+ )
+ )
+
+
+def extract_openai_error_contract_from_bytes(response_body: bytes) -> dict[str, Any] | None:
+ """Best-effort parse OpenAI-style error object from raw response bytes."""
+ try:
+ payload = json.loads(response_body.decode("utf-8", errors="replace"))
+ except Exception: # noqa: BLE001
+ return None
+ return extract_openai_error_contract_from_payload(payload)
+
+
+def extract_openai_error_contract_from_payload(payload: Any) -> dict[str, Any] | None:
+ """Best-effort parse OpenAI-style error object from decoded JSON payload."""
+ if not isinstance(payload, dict):
+ return None
+ error_obj = payload.get("error")
+ if not isinstance(error_obj, dict):
+ return None
+ if not isinstance(error_obj.get("message"), str):
+ return None
+ return error_obj
+
+
+def _build_sidecar_cmd(device: int, target_mem_ratio: float, hold_seconds: int, strict: bool) -> list[str]:
+ sidecar = r"""
+import sys
+import time
+import torch
+
+device = int(sys.argv[1])
+target_ratio = float(sys.argv[2])
+hold_seconds = int(sys.argv[3])
+strict = sys.argv[4] == "1"
+
+torch.cuda.init()
+torch.cuda.set_device(device)
+props = torch.cuda.get_device_properties(device)
+free_before, total_bytes = torch.cuda.mem_get_info(device)
+target_bytes = int(free_before * target_ratio)
+chunk_bytes = 256 * 1024 * 1024
+chunks = []
+allocated = 0
+
+while allocated < target_bytes:
+ req_bytes = min(chunk_bytes, target_bytes - allocated)
+ req_elems = max(1, req_bytes // 2) # float16 -> 2 bytes
+ try:
+ chunk = torch.empty((req_elems,), dtype=torch.float16, device=f"cuda:{device}")
+ chunks.append(chunk)
+ allocated += chunk.numel() * 2
+ except RuntimeError:
+ break
+
+# In strict mode, keep filling with smaller chunks until allocator rejects.
+# This minimizes residual free memory and makes fault-path assertions steadier.
+if strict:
+ tail_chunk_bytes = [64 * 1024 * 1024, 16 * 1024 * 1024, 4 * 1024 * 1024, 1 * 1024 * 1024]
+ for tail_bytes in tail_chunk_bytes:
+ while True:
+ req_elems = max(1, tail_bytes // 2)
+ try:
+ chunk = torch.empty((req_elems,), dtype=torch.float16, device=f"cuda:{device}")
+ chunks.append(chunk)
+ allocated += chunk.numel() * 2
+ except RuntimeError:
+ break
+
+achieved_ratio = allocated / max(1, props.total_memory)
+achieved_free_ratio = allocated / max(1, int(free_before))
+free_after, _ = torch.cuda.mem_get_info(device)
+if strict and allocated < target_bytes:
+ print(
+ "ERROR:"
+ f"achieved_free_ratio={achieved_free_ratio:.4f};"
+ f"achieved_total_ratio={achieved_ratio:.4f};"
+ f"free_before={int(free_before)};"
+ f"free_after={int(free_after)};"
+ f"target_bytes={target_bytes};"
+ f"allocated={allocated}",
+ flush=True,
+ )
+ sys.exit(2)
+
+print(
+ "READY:"
+ f"achieved_free_ratio={achieved_free_ratio:.4f};"
+ f"achieved_total_ratio={achieved_ratio:.4f};"
+ f"free_before={int(free_before)};"
+ f"free_after={int(free_after)};"
+ f"target_bytes={target_bytes};"
+ f"allocated={allocated}",
+ flush=True,
+)
+if hold_seconds <= 0:
+ while True:
+ time.sleep(3600)
+time.sleep(hold_seconds)
+print("DONE", flush=True)
+"""
+ return [
+ sys.executable,
+ "-u",
+ "-c",
+ sidecar,
+ str(device),
+ str(target_mem_ratio),
+ str(hold_seconds),
+ "1" if strict else "0",
+ ]
+
+
+def start_gpu_oom_hog(
+ *,
+ device: int = 0,
+ target_mem_ratio: float = 0.95,
+ hold_seconds: int = 60,
+ startup_timeout_sec: int = 20,
+ strict: bool = True,
+ poll_interval_sec: float = 0.2,
+) -> OomHandle:
+ """Start a CUDA sidecar process that occupies GPU memory to trigger OOM.
+
+ Note:
+ ``target_mem_ratio`` is evaluated against free memory at injection start
+ (not total GPU memory), i.e. success gate is ``allocated / free_before``.
+ ``hold_seconds <= 0`` means keeping OOM pressure until the sidecar is
+ explicitly stopped via ``stop_gpu_oom_hog(s)``.
+ """
+ if os.name == "nt":
+ raise RuntimeError("CUDA OOM sidecar is intended for Linux CI/runtime.")
+ if not (0.0 <= target_mem_ratio < 1.0):
+ raise ValueError("target_mem_ratio should be in [0.0, 1.0).")
+
+ # Explicit opt-out for debugging: keep API shape stable while disabling injection.
+ if target_mem_ratio == 0.0:
+ print(f"[oom-sidecar][gpu={device}] DISABLED: target_mem_ratio=0.0 (no OOM injection)", flush=True)
+ return OomHandle(
+ proc=None,
+ device=device,
+ target_mem_ratio=target_mem_ratio,
+ start_ts=time.time(),
+ )
+
+ cmd = _build_sidecar_cmd(device, target_mem_ratio, hold_seconds, strict)
+ proc = subprocess.Popen(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ text=True,
+ bufsize=1,
+ )
+ assert proc.stdout is not None
+
+ deadline = time.time() + startup_timeout_sec
+ logs: list[str] = []
+ while time.time() < deadline:
+ ready, _, _ = select.select([proc.stdout], [], [], poll_interval_sec)
+ if ready:
+ line = proc.stdout.readline().strip()
+ if line:
+ logs.append(line)
+ print(f"[oom-sidecar][gpu={device}] {line}", flush=True)
+ if line.startswith("READY:"):
+ return OomHandle(
+ proc=proc,
+ device=device,
+ target_mem_ratio=target_mem_ratio,
+ start_ts=time.time(),
+ )
+ if line.startswith("ERROR:"):
+ proc.terminate()
+ raise RuntimeError(f"OOM sidecar failed to reach target: {line}")
+ if proc.poll() is not None:
+ break
+
+ proc.terminate()
+ if logs:
+ print(f"[oom-sidecar][gpu={device}] startup logs: {' | '.join(logs)}", flush=True)
+ raise TimeoutError(f"OOM sidecar startup timeout. logs={logs}")
+
+
+def stop_gpu_oom_hog(handle: OomHandle, *, timeout_sec: int = 5) -> None:
+ """Stop and cleanup CUDA OOM sidecar."""
+ proc = handle.proc
+ if proc is None:
+ return
+ if proc.poll() is not None:
+ return
+ proc.terminate()
+ try:
+ proc.wait(timeout=timeout_sec)
+ except subprocess.TimeoutExpired:
+ proc.kill()
+ proc.wait(timeout=timeout_sec)
+
+
+def inject_gpu_oom(
+ *,
+ device: int | str | list[int] = 0,
+ target_mem_ratio: float = 0.95,
+ hold_seconds: int = 60,
+ startup_timeout_sec: int = 20,
+ strict: bool = True,
+) -> OomHandle | list[OomHandle]:
+ """Convenience wrapper to start CUDA OOM sidecar(s).
+
+ Args:
+ device: One device id (``0``), comma-separated string (``"0,1,2"``),
+ or a list of device ids (``[0, 1, 2]``).
+ hold_seconds: OOM hold time in seconds; ``<=0`` keeps pressure until
+ ``stop_gpu_oom_hogs`` is called.
+ """
+ if isinstance(device, int):
+ devices = [device]
+ elif isinstance(device, str):
+ devices = [int(x.strip()) for x in device.split(",") if x.strip()]
+ else:
+ devices = [int(x) for x in device]
+ if not devices:
+ raise ValueError("device must not be empty.")
+
+ handles = [
+ start_gpu_oom_hog(
+ device=dev,
+ target_mem_ratio=target_mem_ratio,
+ hold_seconds=hold_seconds,
+ startup_timeout_sec=startup_timeout_sec,
+ strict=strict,
+ )
+ for dev in devices
+ ]
+ if len(handles) == 1:
+ return handles[0]
+ return handles
+
+
+def stop_gpu_oom_hogs(handles: OomHandle | list[OomHandle], *, timeout_sec: int = 5) -> None:
+ """Stop one or multiple OOM sidecars."""
+ if isinstance(handles, OomHandle):
+ stop_gpu_oom_hog(handles, timeout_sec=timeout_sec)
+ return
+ for handle in handles:
+ stop_gpu_oom_hog(handle, timeout_sec=timeout_sec)
+
+
+def _runtime_teardown_ssh_target() -> str:
+ target = os.getenv("RUNTIME_TEARDOWN_SSH_TARGET", "").strip()
+ # Default to root@127.0.0.1 for same-host SSH control path.
+ return target or "root@127.0.0.1"
+
+
+def _runtime_teardown_ssh_cmd(remote_cmd: str, *, step: str | None = None) -> subprocess.CompletedProcess[str]:
+ ssh_target = _runtime_teardown_ssh_target()
+ default_reuse_opts = "-o ControlMaster=auto -o ControlPersist=10m -o ControlPath=/tmp/vllm-rt-ssh-%r@%h:%p"
+ raw_opts = os.getenv("RUNTIME_TEARDOWN_SSH_OPTS", "").strip()
+ ssh_opts = shlex.split(raw_opts or default_reuse_opts)
+ timeout_sec = int(os.getenv("RUNTIME_TEARDOWN_SSH_TIMEOUT_SEC", "600"))
+ step_prefix = f"[runtime-teardown][ssh]{f'[{step}]' if step else ''}"
+ print(f"{step_prefix} target={ssh_target} running remote command...", flush=True)
+ # IMPORTANT: SSH joins remote argv into one shell command string. If we pass
+ # ["bash", "-c", remote_cmd] as separate argv items, remote shell parsing can
+ # make `-c` consume only the first word (e.g. "docker"), causing docker help.
+ # Wrap the whole command as one quoted string for remote bash -c.
+ remote_invocation = f"bash --noprofile --norc -c {shlex.quote(remote_cmd)}"
+ try:
+ out = subprocess.run(
+ ["ssh", *ssh_opts, ssh_target, remote_invocation],
+ check=False,
+ capture_output=True,
+ text=True,
+ timeout=timeout_sec,
+ )
+ except subprocess.TimeoutExpired as exc:
+ raise RuntimeError(
+ f"{step_prefix} timed out after {timeout_sec}s. Increase RUNTIME_TEARDOWN_SSH_TIMEOUT_SEC if needed."
+ ) from exc
+ print(f"{step_prefix} exit_code={out.returncode}", flush=True)
+ return out
+
+
+def list_remote_process_pids_by_pattern(pattern: str) -> list[int]:
+ """Return matched PIDs from remote host ``pgrep -f `` via SSH."""
+ cmd = f"pgrep -f {shlex.quote(pattern)} || true"
+ out = _runtime_teardown_ssh_cmd(cmd, step="pgrep")
+ if out.returncode not in (0, 1):
+ raise RuntimeError(f"remote pgrep failed for pattern={pattern!r}: {out.stderr.strip()}")
+ return [int(item) for item in out.stdout.split() if item.strip().isdigit()]
+
+
+def inject_process_kill(
+ *,
+ grep_pattern: str,
+ signal_name: str = "SIGTERM",
+ limit: int | None = None,
+ allow_zero_match: bool = False,
+ execute_kill: bool = True,
+) -> list[int]:
+ """Kill processes matching pattern with selected signal."""
+ if os.name == "nt":
+ raise RuntimeError("process-kill helper currently supports POSIX platforms only.")
+ if not grep_pattern.strip():
+ raise ValueError("grep_pattern must not be empty.")
+
+ sig = getattr(signal, signal_name, None)
+ if sig is None:
+ raise ValueError(f"Unsupported signal_name: {signal_name}")
+
+ out = subprocess.run(
+ ["pgrep", "-f", grep_pattern],
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+ pids = [int(item) for item in out.stdout.split() if item.strip().isdigit()]
+ if limit is not None:
+ pids = pids[:limit]
+
+ if not pids and not allow_zero_match:
+ raise RuntimeError(f"No process matched pattern: {grep_pattern}")
+
+ if execute_kill:
+ for pid in pids:
+ os.kill(pid, sig)
+ return pids
+
+
+def _safe_proc_info(pid: int) -> tuple[str, str]:
+ """Best-effort process name/cmdline lookup for debug logging."""
+ try:
+ proc = psutil.Process(pid)
+ name = proc.name()
+ cmdline = " ".join(proc.cmdline()) or ""
+ return name, cmdline
+ except Exception: # noqa: BLE001
+ return "", ""
+
+
+def _list_server_process_tree(server: Any) -> list[int]:
+ """Return [root, descendants...] PIDs for the current test server instance."""
+ root_proc = getattr(server, "proc", None)
+ if root_proc is None or getattr(root_proc, "pid", None) is None:
+ return []
+
+ root_pid = int(root_proc.pid)
+ try:
+ root = psutil.Process(root_pid)
+ except Exception: # noqa: BLE001
+ return [root_pid]
+
+ descendants = [child.pid for child in root.children(recursive=True)]
+ return [root_pid, *descendants]
+
+
+def _log_server_process_tree(server: Any) -> None:
+ """Print server process tree for debugging fault injection targets."""
+ pids = _list_server_process_tree(server)
+ if not pids:
+ logger.warning("[reliability][process-kill] current server has no visible process tree")
+ return
+ for pid in pids:
+ name, cmdline = _safe_proc_info(pid)
+ print(
+ f"[reliability][process-kill] current_server_proc pid={pid} name={name} cmdline={cmdline}",
+ flush=True,
+ )
+
+
+FaultInjector = Callable[[Any], None]
+"""Callable invoked with the live ``OmniServer`` after it is ready (see ``omni_server_after_fault``)."""
+
+
+def make_process_kill_fault_injector(
+ *,
+ grep_patterns: str | Sequence[str],
+ signal_name: str = "SIGKILL",
+ limit: int = 1,
+ post_kill_wait_seconds: float = 0.0,
+) -> FaultInjector:
+ """Build a post-ready injector that kills processes matched by ``pgrep -f``.
+
+ Tries each pattern in order until at least one PID is killed. If none match,
+ the returned callable issues ``pytest.skip`` (same behavior as the previous
+ inline reliability test).
+
+ Args:
+ grep_patterns: One pattern or an ordered list of patterns.
+ signal_name: Passed to :func:`inject_process_kill` (e.g. ``SIGKILL``).
+ limit: Maximum PIDs to kill per pattern (default ``1``).
+ post_kill_wait_seconds: Optional wait time after kill before test request starts.
+ """
+ patterns: tuple[str, ...] = (grep_patterns,) if isinstance(grep_patterns, str) else tuple(grep_patterns)
+
+ def _inject(server: Any) -> None:
+ _log_server_process_tree(server)
+ server_tree = set(_list_server_process_tree(server))
+ if not server_tree:
+ logger.warning(
+ "[reliability][process-kill] no server process tree found; fallback to global pgrep matching"
+ )
+ for pattern in patterns:
+ print(
+ f"[reliability][process-kill] trying pattern={pattern} signal={signal_name} limit={limit}",
+ flush=True,
+ )
+ pids = inject_process_kill(
+ grep_pattern=pattern,
+ signal_name=signal_name,
+ limit=limit,
+ allow_zero_match=True,
+ execute_kill=False,
+ )
+ filtered = [pid for pid in pids if not server_tree or pid in server_tree]
+ if pids and not filtered:
+ logger.warning(
+ "[reliability][process-kill] pattern=%s matched non-server pids=%s, skip them",
+ pattern,
+ pids,
+ )
+ continue
+ if filtered:
+ sig = getattr(signal, signal_name, None)
+ if sig is None:
+ raise ValueError(f"Unsupported signal_name: {signal_name}")
+ for pid in filtered:
+ name, cmdline = _safe_proc_info(pid)
+ print(
+ f"[reliability][process-kill] killing pid={pid} name={name} signal={signal_name} cmdline={cmdline}",
+ flush=True,
+ )
+ os.kill(pid, sig)
+ print(
+ f"[reliability][process-kill] matched pattern={pattern} killed_pids={filtered} killed_count={len(filtered)}",
+ flush=True,
+ )
+ if post_kill_wait_seconds > 0:
+ print(
+ f"[reliability][process-kill] waiting {post_kill_wait_seconds:.2f}s after kill",
+ flush=True,
+ )
+ time.sleep(post_kill_wait_seconds)
+ return
+ logger.warning(
+ "[reliability][process-kill] no process matched patterns=%s signal=%s limit=%s",
+ patterns,
+ signal_name,
+ limit,
+ )
+ pytest.skip("no matching runtime process found for kill injection")
+
+ return _inject
diff --git a/tests/dfx/reliability/test_reliability_qwen3_omni.py b/tests/dfx/reliability/test_reliability_qwen3_omni.py
new file mode 100644
index 000000000..18251946d
--- /dev/null
+++ b/tests/dfx/reliability/test_reliability_qwen3_omni.py
@@ -0,0 +1,682 @@
+"""
+Qwen3-Omni reliability integration tests.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import errno
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any, Protocol
+
+import pytest
+import torch
+
+from tests.dfx.conftest import (
+ assert_fault_exception,
+ create_reliability_omni_server_params,
+ resolve_oom_device_spec,
+)
+from tests.dfx.reliability.helpers import (
+ extract_openai_error_contract_from_bytes,
+ get_health_raw,
+ inject_gpu_oom,
+ make_process_kill_fault_injector,
+ post_chat_completions_raw,
+ post_json_raw,
+ stop_gpu_oom_hogs,
+)
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video
+from tests.helpers.runtime import dummy_messages_from_mix_data
+
+RELIABILITY_SCENARIOS: list[dict[str, Any]] = [
+ {
+ "test_name": "qwen3_omni_reliability_async_chunk",
+ "server_params": {
+ "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+ "stage_config_name": "qwen3_omni_moe.yaml",
+ "server_args": ["--async-chunk"],
+ },
+ },
+ {
+ "test_name": "qwen3_omni_reliability_default",
+ "server_params": {
+ "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+ "stage_config_name": "qwen3_omni_moe.yaml",
+ "server_args": ["--no-async-chunk"],
+ },
+ },
+]
+
+DEPLOY_CONFIGS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "vllm_omni" / "deploy"
+
+
+def _default_oom_device_spec() -> str:
+ """Use currently visible CUDA ordinals to avoid invalid device index in sidecar."""
+ count = torch.cuda.device_count()
+ if count <= 0:
+ return "0"
+ return ",".join(str(i) for i in range(count))
+
+
+OOM_INJECTION_CONFIG = {
+ "device": _default_oom_device_spec(),
+ "target_mem_ratio": 0.92,
+ "hold_seconds": 0,
+ "startup_timeout_sec": 20,
+ "strict": False,
+}
+# Post-fault recovery probe: keep the API process alive on the same GPU.
+# ``strict=True`` sidecars keep allocating until CUDA refuses, which often
+# evicts/kills the server; here we only take a fraction of *initial* free
+# memory and stop, then rely on a heavy multimodal forward to hit OOM.
+OOM_RECOVER_INJECTION_CONFIG = {
+ "device": _default_oom_device_spec(),
+ # ``strict=False`` stops at this fraction of *initial* free memory per GPU (see helpers sidecar).
+ # On multi-GPU hosts a mild ratio plus a small e2e-style mix can still leave several GB free
+ # on a data-parallel device and the request succeeds; 0.92+ pairs better with the heavy
+ # probe payload below. Raise toward 0.94 only if fault phase still never fails; avoid strict=True
+ # here so the API process is not squeezed to zero slack.
+ "target_mem_ratio": 0.8,
+ "hold_seconds": 0,
+ "startup_timeout_sec": 20,
+ "strict": False,
+}
+FAULT_ERROR_KEYWORDS = (
+ "the request failed",
+ "oom",
+ "out of memory",
+ "cuda",
+ "orchestrator",
+ "timeout",
+ "connection",
+ "500",
+ "503",
+)
+RUNTIME_WORKER_PATTERN = "VLLM::Worker"
+
+
+class _HasServeArgs(Protocol):
+ model: str
+ serve_args: list[str]
+
+
+def _get_system_prompt() -> dict:
+ return {
+ "role": "system",
+ "content": [
+ {
+ "type": "text",
+ "text": (
+ "You are Qwen, a virtual human developed by the Qwen Team, Alibaba Group, "
+ "capable of perceiving auditory and visual inputs, as well as generating text and speech."
+ ),
+ }
+ ],
+ }
+
+
+def _get_mix_prompt() -> str:
+ return "What is recited in the audio? What is in this image? Describe the video briefly."
+
+
+def _mix_chat_completions_probe_payload(omni_server: _HasServeArgs) -> dict[str, Any]:
+ """Heavy multimodal ``/v1/chat/completions`` body for OOM recover tests.
+
+ Uses the same synthetic sizes as ``test_reliability_fault_gpu_oom_chat_large_payload_failure``
+ (large vision tensors + long text), not the light expansion mix (224² / 5s audio): under
+ ``strict=False`` sidecars, a small mix often still completes on a GPU that retains multi-GB
+ free after the hog stops at ``target_mem_ratio``.
+ """
+ video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+ messages = dummy_messages_from_mix_data(
+ system_prompt=_get_system_prompt(),
+ video_data_url=video_data_url,
+ image_data_url=image_data_url,
+ audio_data_url=audio_data_url,
+ content_text="What is recited in the audio? What is in this image? What is in this video? " * 200,
+ )
+ return {
+ "model": omni_server.model,
+ "messages": messages,
+ "stream": True,
+ "modalities": ["text", "audio"],
+ }
+
+
+def _fault_keywords_match_response(*, status: int, body: bytes) -> bool:
+ """True when HTTP status or body text looks like an expected fault / OOM class outcome."""
+ if status >= 400:
+ return True
+ text = body.decode("utf-8", errors="replace").lower()
+ # Avoid matching bare ``"500"`` / ``"503"`` in successful streamed bodies (base64/SSE noise).
+ body_fault_hints = (
+ "the request failed",
+ "oom",
+ "out of memory",
+ "cuda",
+ "orchestrator",
+ "timeout",
+ "connection refused",
+ "connection reset",
+ )
+ return any(key in text for key in body_fault_hints)
+
+
+def _stage_config_path_from_omni_server(omni_server: _HasServeArgs) -> str | None:
+ args: list[str] = omni_server.serve_args
+ for i, arg in enumerate(args):
+ if arg == "--stage-configs-path" and i + 1 < len(args):
+ return args[i + 1]
+ if arg.startswith("--stage-configs-path="):
+ return arg.split("=", 1)[1]
+ return None
+
+
+def _looks_like_server_unreachable(exc: BaseException) -> bool:
+ """True when /health cannot be reached because nothing is listening (process exited)."""
+ if isinstance(exc, (ConnectionRefusedError, BrokenPipeError, ConnectionResetError)):
+ return True
+ errno_val = getattr(exc, "errno", None)
+ if isinstance(exc, OSError) and errno_val is not None:
+ return errno_val in (
+ errno.ECONNREFUSED,
+ errno.ECONNRESET,
+ errno.EPIPE,
+ )
+ msg = str(exc).lower()
+ return "connection refused" in msg or "actively refused" in msg
+
+
+QWEN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, DEPLOY_CONFIGS_DIR)
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_error_contract_consistent_chat_speech(
+ omni_server_function,
+) -> None:
+ """Black-box: text chat vs omni audio output should expose a consistent error contract under OOM.
+
+ Speech-style output is requested via ``/v1/chat/completions`` with ``modalities`` that include ``audio``.
+ """
+ device_spec = resolve_oom_device_spec(
+ OOM_INJECTION_CONFIG,
+ _stage_config_path_from_omni_server(omni_server_function),
+ )
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_INJECTION_CONFIG["strict"],
+ )
+ host = omni_server_function.host
+ port = omni_server_function.port
+ mix_base = _mix_chat_completions_probe_payload(omni_server_function)
+ oom_contract_timeout_sec = 120
+ try:
+ chat_status, chat_body = post_json_raw(
+ host,
+ port,
+ "/v1/chat/completions",
+ {
+ "model": mix_base["model"],
+ "messages": mix_base["messages"],
+ "stream": False,
+ "modalities": ["text"],
+ },
+ timeout_sec=oom_contract_timeout_sec,
+ )
+ speech_status, speech_body = post_json_raw(
+ host,
+ port,
+ "/v1/chat/completions",
+ {
+ "model": mix_base["model"],
+ "messages": mix_base["messages"],
+ "stream": False,
+ "modalities": ["text", "audio"],
+ },
+ timeout_sec=oom_contract_timeout_sec,
+ )
+ finally:
+ stop_gpu_oom_hogs(handle)
+
+ # Under OOM pressure both chat and speech should surface runtime-class (5xx)
+ # failures instead of request-validation-class (4xx) errors.
+ chat_error = extract_openai_error_contract_from_bytes(chat_body)
+ speech_error = extract_openai_error_contract_from_bytes(speech_body)
+ print(chat_status, chat_error, speech_status, speech_error)
+
+ assert chat_status >= 500, f"expected chat error under OOM, got status={chat_status}"
+ assert speech_status >= 500, f"expected speech runtime error under OOM, got status={speech_status}"
+
+ assert chat_error is not None, f"chat error payload not OpenAI-compatible: {chat_body[:300]!r}"
+ assert speech_error is not None, f"speech error payload not OpenAI-compatible: {speech_body[:300]!r}"
+ assert "code" in chat_error, f"chat error lacks code field: {chat_error!r}"
+ assert "code" in speech_error, f"speech error lacks code field: {speech_error!r}"
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_chat_large_payload_failure(omni_server_function, openai_client_function) -> None:
+ device_spec = resolve_oom_device_spec(
+ OOM_INJECTION_CONFIG,
+ _stage_config_path_from_omni_server(omni_server_function),
+ )
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_INJECTION_CONFIG["strict"],
+ )
+ try:
+ video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+ messages = dummy_messages_from_mix_data(
+ system_prompt=_get_system_prompt(),
+ video_data_url=video_data_url,
+ image_data_url=image_data_url,
+ audio_data_url=audio_data_url,
+ content_text=f"{_get_mix_prompt()} " * 200,
+ )
+ request_config = {
+ "model": omni_server_function.model,
+ "messages": messages,
+ "stream": True,
+ "key_words": {"audio": ["test"]},
+ }
+ try:
+ openai_client_function.send_omni_request(request_config, request_num=1)
+ except Exception as exc:
+ assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+ else:
+ pytest.fail("expected large chat payload request failure during GPU OOM injection")
+ finally:
+ stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_concurrent_pressure_failure(omni_server_function, openai_client_function) -> None:
+ device_spec = resolve_oom_device_spec(
+ OOM_INJECTION_CONFIG,
+ _stage_config_path_from_omni_server(omni_server_function),
+ )
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_INJECTION_CONFIG["strict"],
+ )
+ try:
+ video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+ messages = dummy_messages_from_mix_data(
+ system_prompt=_get_system_prompt(),
+ video_data_url=video_data_url,
+ image_data_url=image_data_url,
+ audio_data_url=audio_data_url,
+ content_text=f"{_get_mix_prompt()} " * 200,
+ )
+ request_config = {
+ "model": omni_server_function.model,
+ "messages": messages,
+ "stream": True,
+ "modalities": ["text", "audio"],
+ "key_words": {"audio": ["test"]},
+ }
+ try:
+ openai_client_function.send_omni_request(request_config, request_num=4)
+ except Exception as exc:
+ assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+ else:
+ pytest.fail("expected concurrent request failure under OOM injection")
+ finally:
+ stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+ "fault_injector",
+ [
+ pytest.param(
+ make_process_kill_fault_injector(
+ grep_patterns="VLLM::Worker",
+ signal_name="SIGKILL",
+ limit=1,
+ post_kill_wait_seconds=2.0,
+ ),
+ id="runtime_process_chain",
+ ),
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_request_failure(
+ omni_server_after_fault_function, openai_client_function
+) -> None:
+ messages = dummy_messages_from_mix_data(
+ system_prompt=_get_system_prompt(),
+ content_text="What is the capital of China? Answer in 20 words.",
+ )
+ request_config = {
+ "model": omni_server_after_fault_function.model,
+ "messages": messages,
+ "stream": False,
+ "modalities": ["text"],
+ "key_words": {"text": ["beijing"]},
+ }
+ try:
+ openai_client_function.send_omni_request(request_config, request_num=1)
+ except Exception as exc:
+ assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+ else:
+ pytest.fail("expected request failure after process-kill injection")
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+ "fault_injector",
+ [
+ pytest.param(
+ make_process_kill_fault_injector(
+ grep_patterns="VLLM::Worker",
+ signal_name="SIGKILL",
+ limit=1,
+ post_kill_wait_seconds=2.0,
+ ),
+ id="runtime_process_chain",
+ ),
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_health_fast_fail_and_concurrent(
+ omni_server_after_fault_function,
+) -> None:
+ """Black-box: after worker SIGKILL, /health→503, chat fails fast, concurrent chat does not hang."""
+ host = omni_server_after_fault_function.host
+ port = omni_server_after_fault_function.port
+ model = omni_server_after_fault_function.model
+
+ deadline = time.monotonic() + 20.0
+ last_observation = ""
+ saw_503 = False
+ health_final_status: int | None = None
+ health_final_body = b""
+ while time.monotonic() < deadline:
+ try:
+ status, body = get_health_raw(host, port, timeout_sec=5)
+ last_observation = f"http={status}, body={body[:200]!r}"
+ health_final_status, health_final_body = status, body
+ if status == 503:
+ saw_503 = True
+ break
+ except Exception as exc: # noqa: BLE001
+ last_observation = f"exception={exc!r}"
+ time.sleep(0.5)
+ assert saw_503, (
+ f"[process_kill health] expected /health to become 503 after fault injection, got {last_observation}"
+ )
+
+ payload = {
+ "model": model,
+ "messages": [{"role": "user", "content": "Say hello in one short sentence."}],
+ "stream": False,
+ "modalities": ["text"],
+ }
+ ff_status: int | None = None
+ ff_body = b""
+ ff_exc: BaseException | None = None
+ start = time.monotonic()
+ try:
+ ff_status, ff_body = post_json_raw(host, port, "/v1/chat/completions", payload, timeout_sec=20)
+ elapsed = time.monotonic() - start
+ assert elapsed < 15, f"[process_kill fast_fail] request did not fail fast after fault: {elapsed:.2f}s"
+ assert ff_status >= 500, (
+ f"[process_kill fast_fail] expected server-side failure after fault, "
+ f"got status={ff_status}, body={ff_body[:200]!r}"
+ )
+ except Exception as exc:
+ ff_exc = exc
+ elapsed = time.monotonic() - start
+ assert elapsed < 15, f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s"
+
+ payload_json = json.dumps(
+ {
+ "model": model,
+ "messages": [{"role": "user", "content": "What is the capital of China? Answer in one word."}],
+ "stream": False,
+ "modalities": ["text"],
+ }
+ )
+ start = time.monotonic()
+ with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+ futures = [
+ executor.submit(
+ post_chat_completions_raw,
+ host,
+ port,
+ payload_json,
+ timeout_sec=20,
+ )
+ for _ in range(4)
+ ]
+ done, pending = concurrent.futures.wait(
+ futures,
+ timeout=30,
+ return_when=concurrent.futures.ALL_COMPLETED,
+ )
+
+ elapsed = time.monotonic() - start
+ assert not pending, f"[process_kill concurrent] some fault-time requests hung: pending={len(pending)}"
+ assert elapsed < 30, f"[process_kill concurrent] fault-time request convergence is too slow: {elapsed:.2f}s"
+
+ fault_observed = False
+ conc_debug: list[Any] = []
+ for future in done:
+ try:
+ status, body = future.result()
+ conc_debug.append((status, body[:200]))
+ if status >= 500:
+ fault_observed = True
+ except Exception as exc:
+ conc_debug.append(repr(exc))
+ fault_observed = True
+ # DEBUG: remove before merge
+ print(
+ health_final_status,
+ health_final_body[:200],
+ ff_status,
+ ff_body[:200],
+ ff_exc,
+ conc_debug,
+ )
+ assert fault_observed, (
+ "[process_kill concurrent] expected at least one request to fail after process-kill fault injection"
+ )
+
+
+@pytest.mark.slow
+@pytest.mark.skip(reason="issue#2327")
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_state_converges_after_fault_removed(
+ omni_server_function,
+) -> None:
+ """Black-box: under bounded GPU pressure, a heavy mix chat fails; after hog stops, the same request succeeds.
+
+ Uses a milder sidecar than ``OOM_INJECTION_CONFIG`` (``strict=False`` and a capped
+ ``target_mem_ratio``) so the serving process stays alive: pressure should starve
+ the multimodal forward pass, not exhaust the whole device allocator.
+
+ Fault and recovery phases use raw ``/v1/chat/completions`` HTTP so tests can assert
+ on status codes and JSON error payloads (or fault keywords in streamed bodies).
+ """
+ device_spec = resolve_oom_device_spec(
+ OOM_RECOVER_INJECTION_CONFIG,
+ _stage_config_path_from_omni_server(omni_server_function),
+ )
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_RECOVER_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_RECOVER_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_RECOVER_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_RECOVER_INJECTION_CONFIG["strict"],
+ )
+ host = omni_server_function.host
+ port = omni_server_function.port
+ mix_payload = _mix_chat_completions_probe_payload(omni_server_function)
+ mix_chat_timeout_sec = 180
+ print(
+ "[oom-recover] injection ready "
+ f"device={device_spec} target_mem_ratio={OOM_RECOVER_INJECTION_CONFIG['target_mem_ratio']} "
+ f"strict={OOM_RECOVER_INJECTION_CONFIG['strict']} hold_seconds={OOM_RECOVER_INJECTION_CONFIG['hold_seconds']}"
+ )
+ print(f"[oom-recover] target endpoint http://{host}:{port}/v1/chat/completions")
+
+ failure_observed = False
+ fault_status: int | None = None
+ fault_body = b""
+ try:
+ try:
+ print(f"[oom-recover] fault-phase request start timeout={mix_chat_timeout_sec}s")
+ fault_status, fault_body = post_json_raw(
+ host,
+ port,
+ "/v1/chat/completions",
+ mix_payload,
+ timeout_sec=mix_chat_timeout_sec,
+ )
+ print(f"[oom-recover] fault-phase request done status={fault_status} body_prefix={fault_body[:200]!r}")
+ except Exception as exc:
+ failure_observed = True
+ print(f"[oom-recover] fault-phase request raised {type(exc).__name__}: {exc!r}")
+ assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+ else:
+ assert fault_status is not None
+ failure_observed = fault_status != 200 or _fault_keywords_match_response(
+ status=fault_status,
+ body=fault_body,
+ )
+ assert failure_observed, (
+ "expected mix multimodal request failure while OOM pressure is active; "
+ f"status={fault_status}, body_prefix={fault_body[:500]!r}"
+ )
+ if fault_status >= 400:
+ err = extract_openai_error_contract_from_bytes(fault_body)
+ if err is None:
+ assert _fault_keywords_match_response(status=fault_status, body=fault_body), (
+ "non-2xx without OpenAI-style error object should still mention fault hints in body; "
+ f"status={fault_status}, body_prefix={fault_body[:500]!r}"
+ )
+ else:
+ assert isinstance(err.get("message"), str) and err["message"].strip(), (
+ f"structured error should carry a non-empty message: {err!r}"
+ )
+ else:
+ assert _fault_keywords_match_response(status=fault_status, body=fault_body), (
+ "HTTP 200 under pressure should still surface fault hints in the response body; "
+ f"body_prefix={fault_body[:500]!r}"
+ )
+ finally:
+ print("[oom-recover] stopping OOM sidecar(s)")
+ stop_gpu_oom_hogs(handle)
+ print("[oom-recover] OOM sidecar(s) stopped")
+
+ recovery_deadline = time.monotonic() + 90.0
+ terminal_health: int | None = None
+ unreachable_streak = 0
+ last_health_exc: BaseException | None = None
+ while time.monotonic() < recovery_deadline:
+ try:
+ status, _ = get_health_raw(host, port, timeout_sec=5)
+ unreachable_streak = 0
+ if status in (200, 503):
+ terminal_health = status
+ print(f"[oom-recover] terminal health reached status={terminal_health}")
+ break
+ except Exception as exc:
+ last_health_exc = exc
+ if _looks_like_server_unreachable(exc):
+ unreachable_streak += 1
+ if unreachable_streak >= 5:
+ pytest.fail(
+ "after OOM sidecar stopped, /health is unreachable (connection refused / reset). "
+ "The APIServer process likely exited (e.g. orchestrator thread crash under OOM); "
+ "this test expects the server to stay up for post-fault health polling. "
+ f"last_exc={last_health_exc!r}"
+ )
+ else:
+ unreachable_streak = 0
+ time.sleep(1.0)
+ else:
+ pytest.fail(
+ "server did not converge to a terminal health state after OOM pressure was removed; "
+ f"last_health_exc={last_health_exc!r}"
+ )
+
+ probe_payload = {
+ "model": omni_server_function.model,
+ "messages": [{"role": "user", "content": "What is the capital of China? Answer in one word."}],
+ "stream": False,
+ "modalities": ["text"],
+ }
+ start = time.monotonic()
+ post_mix_status: int | None = None
+ post_mix_body = b""
+ request_status: int | None = None
+ try:
+ assert terminal_health is not None
+ if terminal_health == 200:
+ print(f"[oom-recover] recovery-phase mix request start timeout={mix_chat_timeout_sec}s")
+ post_mix_status, post_mix_body = post_json_raw(
+ host,
+ port,
+ "/v1/chat/completions",
+ mix_payload,
+ timeout_sec=mix_chat_timeout_sec,
+ )
+ print(
+ "[oom-recover] recovery-phase mix request done "
+ f"status={post_mix_status} body_prefix={post_mix_body[:200]!r}"
+ )
+ else:
+ print("[oom-recover] terminal health=503, send lightweight probe request")
+ request_status, _ = post_json_raw(host, port, "/v1/chat/completions", probe_payload, timeout_sec=20)
+ print(f"[oom-recover] lightweight probe done status={request_status}")
+ except Exception:
+ post_mix_status = None
+ request_status = None
+ elapsed = time.monotonic() - start
+ assert elapsed < 200, f"post-fault request should not hang after OOM removal: {elapsed:.2f}s"
+
+ if terminal_health == 200:
+ assert post_mix_status == 200, (
+ "health recovered but repeated mix multimodal request did not return HTTP 200; "
+ f"status={post_mix_status}, body_prefix={post_mix_body[:600]!r}"
+ )
+ recover_err = extract_openai_error_contract_from_bytes(post_mix_body)
+ assert recover_err is None, f"unexpected error object after recovery: {recover_err!r}"
+ else:
+ assert request_status is None or request_status >= 500, (
+ "unhealthy terminal state should fail fast on requests, "
+ f"got health={terminal_health}, request_status={request_status}"
+ )
diff --git a/tests/dfx/reliability/test_reliability_wan22.py b/tests/dfx/reliability/test_reliability_wan22.py
new file mode 100644
index 000000000..e27db616f
--- /dev/null
+++ b/tests/dfx/reliability/test_reliability_wan22.py
@@ -0,0 +1,411 @@
+"""
+Wan2.2 reliability integration tests.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+import pytest
+import requests
+
+from tests.dfx.conftest import (
+ assert_fault_exception,
+ create_reliability_omni_server_params,
+ resolve_oom_device_spec,
+ supports_video_generation,
+)
+from tests.dfx.reliability.helpers import (
+ get_health_raw,
+ inject_gpu_oom,
+ make_process_kill_fault_injector,
+ stop_gpu_oom_hogs,
+)
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import generate_synthetic_image
+
+RELIABILITY_SCENARIOS: list[dict[str, Any]] = [
+ {
+ "test_name": "wan22_t2v_reliability_default",
+ "server_params": {
+ "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers",
+ "server_args": [
+ "--num-gpus",
+ "1",
+ "--boundary-ratio",
+ "0.875",
+ "--flow-shift",
+ "5.0",
+ "--disable-log-stats",
+ ],
+ },
+ }
+]
+
+E2E_STAGE_CONFIGS_DIR = Path(__file__).resolve().parent.parent / "e2e" / "stage_configs"
+OOM_INJECTION_CONFIG = {
+ "target_mem_ratio": 0.95,
+ "hold_seconds": 0,
+ "startup_timeout_sec": 20,
+ "strict": True,
+}
+FAULT_ERROR_KEYWORDS = (
+ "oom",
+ "out of memory",
+ "cuda",
+ "job failed",
+ "unknown error",
+ "internal server error",
+ "500 server error",
+)
+PROCESS_KILL_ERROR_KEYWORDS = (
+ "timeout",
+ "did not complete within",
+ "connection",
+ "engine",
+ "orchestrator",
+ "dead",
+ "internal",
+ "500",
+ "503",
+)
+
+WAN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, E2E_STAGE_CONFIGS_DIR)
+DIFFUSION_VIDEO_PARAMS = [param for param in WAN_PARAMS if supports_video_generation(param.model)]
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_video_large_request_failure(omni_server_function, openai_client_function) -> None:
+ stage_config_path = getattr(omni_server_function, "stage_config_path", None)
+ device_spec = resolve_oom_device_spec(OOM_INJECTION_CONFIG, stage_config_path)
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_INJECTION_CONFIG["strict"],
+ )
+ try:
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ request_config = {
+ "form_data": {
+ "prompt": "Generate a realistic road-driving video with camera motion.",
+ "width": 512,
+ "height": 512,
+ "fps": 8,
+ "num_frames": 8,
+ "guidance_scale": 1.0,
+ "flow_shift": 5.0,
+ "num_inference_steps": 4,
+ "seed": 42,
+ },
+ "image_reference": image_data_url,
+ "stream": False,
+ }
+ try:
+ openai_client_function.send_video_diffusion_request(request_config, request_num=1)
+ except Exception as exc:
+ assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+ else:
+ pytest.fail("expected large /v1/videos request failure during GPU OOM injection")
+ finally:
+ stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+ "fault_injector",
+ [
+ pytest.param(
+ make_process_kill_fault_injector(
+ grep_patterns="multiprocessing.spawn",
+ signal_name="SIGKILL",
+ limit=1,
+ post_kill_wait_seconds=2.0,
+ ),
+ id="runtime_process_chain",
+ ),
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_video_request_failure(
+ omni_server_after_fault_function,
+ openai_client_function,
+) -> None:
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ request_config = {
+ "form_data": {
+ "prompt": "Generate a realistic road-driving video with camera motion.",
+ "width": 512,
+ "height": 512,
+ "fps": 8,
+ "num_frames": 8,
+ "guidance_scale": 1.0,
+ "flow_shift": 5.0,
+ "num_inference_steps": 4,
+ "seed": 42,
+ },
+ "image_reference": image_data_url,
+ "stream": False,
+ }
+ try:
+ openai_client_function.send_video_diffusion_request(request_config, request_num=1)
+ except Exception as exc:
+ assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
+ else:
+ pytest.fail("expected /v1/videos request failure after process-kill injection")
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+ "fault_injector",
+ [
+ pytest.param(
+ make_process_kill_fault_injector(
+ grep_patterns="multiprocessing.spawn",
+ signal_name="SIGKILL",
+ limit=1,
+ post_kill_wait_seconds=2.0,
+ ),
+ id="runtime_process_chain",
+ ),
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_video_health_fast_fail_and_concurrent(
+ omni_server_after_fault_function,
+ openai_client_function,
+) -> None:
+ """Black-box: after process kill, /v1/videos fails fast and concurrent calls don't hang.
+
+ /health→503 is checked last: optional ``pytest.skip`` then ``assert`` (drop the ``if`` to harden).
+ """
+ host = omni_server_after_fault_function.host
+ port = omni_server_after_fault_function.port
+ url = f"http://{host}:{port}/v1/videos"
+ image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+ request_config = {
+ "form_data": {
+ "prompt": "Generate a realistic road-driving video with camera motion.",
+ "width": 512,
+ "height": 512,
+ "fps": 8,
+ "num_frames": 8,
+ "guidance_scale": 1.0,
+ "flow_shift": 5.0,
+ "num_inference_steps": 4,
+ "seed": 42,
+ },
+ "image_reference": image_data_url,
+ "stream": False,
+ }
+
+ # Phase-1: one new /v1/videos request should fail fast.
+ payload = {
+ "prompt": "fast-fail check",
+ "width": "512",
+ "height": "512",
+ "fps": "8",
+ "num_frames": "8",
+ "num_inference_steps": "4",
+ }
+ start = time.monotonic()
+ try:
+ response = requests.post(
+ url,
+ data=payload,
+ headers={"Accept": "application/json"},
+ timeout=20,
+ )
+ elapsed = time.monotonic() - start
+ assert elapsed < 15, f"[process_kill fast_fail] /v1/videos did not fail fast after fault: {elapsed:.2f}s"
+ assert response.status_code >= 500, (
+ "[process_kill fast_fail] expected server-side error after fatal fault, "
+ f"got status={response.status_code}, body={response.text[:200]!r}"
+ )
+ except Exception:
+ elapsed = time.monotonic() - start
+ assert elapsed < 15, f"[process_kill fast_fail] /v1/videos exception was too slow after fault: {elapsed:.2f}s"
+
+ # Phase-2: concurrent requests should converge and at least one should fail.
+ start = time.monotonic()
+ with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
+ futures = [
+ executor.submit(openai_client_function.send_video_diffusion_request, request_config, 1) for _ in range(3)
+ ]
+ done, pending = concurrent.futures.wait(
+ futures,
+ timeout=40,
+ return_when=concurrent.futures.ALL_COMPLETED,
+ )
+
+ elapsed = time.monotonic() - start
+ assert not pending, f"[process_kill concurrent] some fault-time video requests hung: pending={len(pending)}"
+ assert elapsed < 40, f"[process_kill concurrent] fault-time video request convergence is too slow: {elapsed:.2f}s"
+
+ fault_observed = False
+ conc_debug: list[Any] = []
+ for future in done:
+ try:
+ future.result()
+ conc_debug.append("ok")
+ except Exception as exc:
+ fault_observed = True
+ conc_debug.append(repr(exc))
+ assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
+ assert fault_observed, (
+ "[process_kill concurrent] expected at least one /v1/videos request to fail after fault; "
+ f"conc_debug={conc_debug}"
+ )
+
+ # Phase-3: /health→503 after fault. Optional skip today; remove the ``if`` later to harden.
+ deadline = time.monotonic() + 20.0
+ last_observation = ""
+ final_health_status: int | None = None
+ while time.monotonic() < deadline:
+ try:
+ status, body = get_health_raw(host, port, timeout_sec=5)
+ last_observation = f"http={status}, body={body[:200]!r}"
+ final_health_status = status
+ if status == 503:
+ break
+ except Exception as exc: # noqa: BLE001
+ last_observation = f"exception={exc!r}"
+ time.sleep(0.5)
+ if final_health_status != 503:
+ pytest.skip("issue#3050")
+ assert final_health_status == 503, (
+ "[process_kill health] expected /health 503 after fatal fault, "
+ f"got status={final_health_status}, last_observation={last_observation}"
+ )
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.skip(reason="issue#2327")
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_video_oom_recovers_after_fault_removed(
+ omni_server_function,
+) -> None:
+ """Black-box: after removing OOM pressure, service converges to a terminal state and requests do not hang."""
+ stage_config_path = getattr(omni_server_function, "stage_config_path", None)
+ device_spec = resolve_oom_device_spec(OOM_INJECTION_CONFIG, stage_config_path)
+ handle = inject_gpu_oom(
+ device=device_spec,
+ target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+ hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+ startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+ strict=OOM_INJECTION_CONFIG["strict"],
+ )
+ host = omni_server_function.host
+ port = omni_server_function.port
+ create_url = f"http://{host}:{port}/v1/videos"
+
+ failure_observed = False
+ try:
+ for _ in range(3):
+ try:
+ response = requests.post(
+ create_url,
+ data={
+ "prompt": "oom recover probe",
+ "width": "512",
+ "height": "512",
+ "fps": "8",
+ "num_frames": "8",
+ "num_inference_steps": "4",
+ },
+ headers={"Accept": "application/json"},
+ timeout=25,
+ )
+ if response.status_code >= 500:
+ failure_observed = True
+ break
+ except Exception:
+ failure_observed = True
+ break
+ time.sleep(1.0)
+ finally:
+ stop_gpu_oom_hogs(handle)
+
+ assert failure_observed, "expected at least one video request failure while OOM pressure is active"
+
+ recovery_deadline = time.monotonic() + 90.0
+ terminal_health: int | None = None
+ unreachable_streak = 0
+ last_health_exc: BaseException | None = None
+ while time.monotonic() < recovery_deadline:
+ try:
+ status, _ = get_health_raw(host, port, timeout_sec=5)
+ unreachable_streak = 0
+ if status in (200, 503):
+ terminal_health = status
+ break
+ except Exception as exc: # noqa: BLE001
+ last_health_exc = exc
+ msg = str(exc).lower()
+ if "connection refused" in msg or "actively refused" in msg:
+ unreachable_streak += 1
+ if unreachable_streak >= 5:
+ pytest.fail(
+ "after OOM sidecar stopped, /health is unreachable repeatedly; "
+ "the APIServer process likely exited unexpectedly "
+ f"(last_exc={last_health_exc!r})"
+ )
+ else:
+ unreachable_streak = 0
+ time.sleep(1.0)
+ else:
+ pytest.fail(
+ "wan22 server did not converge to terminal health (200/503) after OOM pressure was removed; "
+ f"last_health_exc={last_health_exc!r}"
+ )
+
+ assert terminal_health is not None
+ start = time.monotonic()
+ recovery_resp: requests.Response | None = None
+ try:
+ recovery_resp = requests.post(
+ create_url,
+ data={
+ "prompt": "post-recovery admission check",
+ "width": "512",
+ "height": "512",
+ "fps": "8",
+ "num_frames": "8",
+ "num_inference_steps": "4",
+ },
+ headers={"Accept": "application/json"},
+ timeout=30,
+ )
+ recovery_status = recovery_resp.status_code
+ recovery_text_prefix = recovery_resp.text[:300]
+ except Exception as exc: # noqa: BLE001
+ recovery_status = None
+ recovery_text_prefix = repr(exc)
+ elapsed = time.monotonic() - start
+ assert elapsed < 30, f"post-fault /v1/videos should not hang after OOM removal: {elapsed:.2f}s"
+
+ if terminal_health == 200:
+ assert recovery_status == 200, (
+ "health recovered but /v1/videos admission did not succeed; "
+ f"status={recovery_status}, body={recovery_text_prefix!r}"
+ )
+ assert recovery_resp is not None
+ recovery_payload = recovery_resp.json()
+ assert "id" in recovery_payload, f"post-recovery create payload missing id: {recovery_payload!r}"
+ else:
+ assert recovery_status is None or recovery_status >= 500, (
+ "unhealthy terminal state should fail fast on /v1/videos, "
+ f"got health={terminal_health}, request_status={recovery_status}, body={recovery_text_prefix!r}"
+ )
diff --git a/tests/helpers/fixtures/runtime.py b/tests/helpers/fixtures/runtime.py
index 4cae13cd6..1fb5fd7a1 100644
--- a/tests/helpers/fixtures/runtime.py
+++ b/tests/helpers/fixtures/runtime.py
@@ -10,17 +10,27 @@ from __future__ import annotations
import threading
from collections.abc import Generator
-from typing import Any
+from typing import TYPE_CHECKING, Any
import pytest
-import yaml
-from tests.helpers.runtime import OmniServer
-from tests.helpers.stage_config import modify_stage_config
+if TYPE_CHECKING:
+ from tests.helpers.runtime import OmniServer
omni_fixture_lock = threading.Lock()
+@pytest.fixture(scope="function")
+def omni_server_function(
+ request: pytest.FixtureRequest,
+ run_level: str,
+ model_prefix: str,
+) -> Generator[OmniServer, Any, None]:
+ from tests.helpers.runtime import run_omni_server
+
+ yield from run_omni_server(request, run_level, model_prefix, omni_fixture_lock)
+
+
@pytest.fixture(scope="module")
def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: str) -> Generator[OmniServer, Any, None]:
"""Start vLLM-Omni through the standard or stage-CLI launcher.
@@ -29,81 +39,9 @@ def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: st
The ``use_stage_cli`` flag on ``OmniServerParams`` routes the setup through the
stage-CLI harness while still reusing the same fixture grouping semantics.
"""
- with omni_fixture_lock:
- from tests.helpers.runtime import OmniServer, OmniServerParams, OmniServerStageCli
+ from tests.helpers.runtime import run_omni_server
- params: OmniServerParams = request.param
- model = model_prefix + params.model
- port = params.port
- stage_config_path = params.stage_config_path
- if run_level in {"advanced_model", "full_model"} and stage_config_path is not None:
- with open(stage_config_path, encoding="utf-8") as f:
- cfg = yaml.safe_load(f) or {}
- # Strip ``load_format: dummy`` (CI overlay default) so advanced_model
- # tests use real weights. New schema (``stages:``) writes the field
- # flat at stage level; legacy schema (``stage_args:``) nests it as
- # ``engine_args.load_format``. Handle both.
- new_schema_stages = cfg.get("stages")
- stage_key = "stages" if new_schema_stages is not None else "stage_args"
- delete_path = "load_format" if new_schema_stages is not None else "engine_args.load_format"
- stage_entries = cfg.get(stage_key, [])
- stage_ids = [stage["stage_id"] for stage in stage_entries if "stage_id" in stage]
- stage_config_path = modify_stage_config(
- stage_config_path,
- deletes={stage_key: {stage_id: [delete_path] for stage_id in stage_ids}},
- )
-
- server_args = params.server_args or []
- if params.use_omni and params.stage_init_timeout is not None:
- server_args = [*server_args, "--stage-init-timeout", str(params.stage_init_timeout)]
- else:
- server_args = [*server_args, "--stage-init-timeout", "600"]
- if params.init_timeout is not None:
- server_args = [*server_args, "--init-timeout", str(params.init_timeout)]
- else:
- server_args = [*server_args, "--init-timeout", "900"]
- if params.use_stage_cli:
- if not params.use_omni:
- raise ValueError("omni_server with use_stage_cli=True requires use_omni=True")
- if stage_config_path is None:
- raise ValueError("omni_server with use_stage_cli=True requires a stage_config_path")
- server_args += ["--stage-configs-path", stage_config_path]
-
- with OmniServerStageCli(
- model,
- stage_config_path,
- server_args,
- port=port,
- env_dict=params.env_dict,
- ) as server:
- print("OmniServer started successfully")
- yield server
- print("OmniServer stopping...")
- else:
- if stage_config_path is not None:
- server_args += ["--stage-configs-path", stage_config_path]
-
- with (
- OmniServer(
- model,
- server_args,
- port=port,
- env_dict=params.env_dict,
- use_omni=params.use_omni,
- )
- if port
- else OmniServer(
- model,
- server_args,
- env_dict=params.env_dict,
- use_omni=params.use_omni,
- )
- ) as server:
- print("OmniServer started successfully")
- yield server
- print("OmniServer stopping...")
-
- print("OmniServer stopped")
+ yield from run_omni_server(request, run_level, model_prefix, omni_fixture_lock)
@pytest.fixture
@@ -115,6 +53,15 @@ def openai_client(request: pytest.FixtureRequest, run_level: str):
return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
+@pytest.fixture
+def openai_client_function(request: pytest.FixtureRequest, run_level: str):
+ """Resolve ``omni_server_function`` lazily for function-scoped reliability tests."""
+ from tests.helpers.runtime import OpenAIClientHandler
+
+ server = request.getfixturevalue("omni_server_function")
+ return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
+
+
@pytest.fixture(scope="module")
def omni_runner(request: pytest.FixtureRequest, model_prefix: str):
from tests.helpers.runtime import OmniRunner
diff --git a/tests/helpers/runtime.py b/tests/helpers/runtime.py
index 0cf0f9e48..0b4d730c5 100644
--- a/tests/helpers/runtime.py
+++ b/tests/helpers/runtime.py
@@ -9,7 +9,9 @@ import socket
import subprocess
import sys
import tempfile
+import threading
import time
+from collections.abc import Generator
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
@@ -120,6 +122,89 @@ class OmniServerParams(NamedTuple):
stage_init_timeout: int | None = None # None: fixture supplies default (600 s)
+def run_omni_server(
+ request: Any,
+ run_level: str,
+ model_prefix: str,
+ omni_fixture_lock: threading.Lock,
+) -> Generator["OmniServer", Any, None]:
+ from tests.helpers.stage_config import modify_stage_config
+
+ with omni_fixture_lock:
+ params: OmniServerParams = request.param
+ model = model_prefix + params.model
+ port = params.port
+ stage_config_path = params.stage_config_path
+ if run_level in {"advanced_model", "full_model"} and stage_config_path is not None:
+ with open(stage_config_path, encoding="utf-8") as f:
+ cfg = yaml.safe_load(f) or {}
+ # Strip ``load_format: dummy`` (CI overlay default) so advanced_model
+ # tests use real weights. New schema (``stages:``) writes the field
+ # flat at stage level; legacy schema (``stage_args:``) nests it as
+ # ``engine_args.load_format``. Handle both.
+ new_schema_stages = cfg.get("stages")
+ stage_key = "stages" if new_schema_stages is not None else "stage_args"
+ delete_path = "load_format" if new_schema_stages is not None else "engine_args.load_format"
+ stage_entries = cfg.get(stage_key, [])
+ stage_ids = [stage["stage_id"] for stage in stage_entries if "stage_id" in stage]
+ stage_config_path = modify_stage_config(
+ stage_config_path,
+ deletes={stage_key: {stage_id: [delete_path] for stage_id in stage_ids}},
+ )
+
+ server_args = params.server_args or []
+ if params.use_omni and params.stage_init_timeout is not None:
+ server_args = [*server_args, "--stage-init-timeout", str(params.stage_init_timeout)]
+ else:
+ server_args = [*server_args, "--stage-init-timeout", "600"]
+ if params.init_timeout is not None:
+ server_args = [*server_args, "--init-timeout", str(params.init_timeout)]
+ else:
+ server_args = [*server_args, "--init-timeout", "900"]
+ if params.use_stage_cli:
+ if not params.use_omni:
+ raise ValueError("omni_server with use_stage_cli=True requires use_omni=True")
+ if stage_config_path is None:
+ raise ValueError("omni_server with use_stage_cli=True requires a stage_config_path")
+ server_args += ["--stage-configs-path", stage_config_path]
+
+ with OmniServerStageCli(
+ model,
+ stage_config_path,
+ server_args,
+ port=port,
+ env_dict=params.env_dict,
+ ) as server:
+ print("OmniServer started successfully")
+ yield server
+ print("OmniServer stopping...")
+ else:
+ if stage_config_path is not None:
+ server_args += ["--stage-configs-path", stage_config_path]
+
+ with (
+ OmniServer(
+ model,
+ server_args,
+ port=port,
+ env_dict=params.env_dict,
+ use_omni=params.use_omni,
+ )
+ if port
+ else OmniServer(
+ model,
+ server_args,
+ env_dict=params.env_dict,
+ use_omni=params.use_omni,
+ )
+ ) as server:
+ print("OmniServer started successfully")
+ yield server
+ print("OmniServer stopping...")
+
+ print("OmniServer stopped")
+
+
class OmniServer:
"""Omniserver for vLLM-Omni tests."""
|