# Vllm Omni Test

> Generate and run tests for vllm-project/vllm-omni with CI-aligned levels and markers; wire new tests into Buildkite (test-ready.yml for L1/L2, test-merge.yml for L3, test-nightly.yml for L4). On completion, always provide copy-paste local and CI-like pytest commands plus prerequisites. Use when creating regression tests, adding L1-L4 coverage, selecting pytest markers, or validating fixes from issues/PRs.

- Skill: `vllm-project-vllm-omni/vllm-omni-test` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add vllm-project-vllm-omni/vllm-omni-test`
- Raw SKILL.md: https://api.skillmd.com/api/skills/vllm-project-vllm-omni/vllm-omni-test/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: vllm-project (https://skillmd.com/u/vllm-project-vllm-omni)
- Updated: 2026-09-10
- Page: https://skillmd.com/skills/vllm-project-vllm-omni/vllm-omni-test

---


# vLLM-Omni Test Generator & Runner

## Purpose

Use this skill to generate minimal, stable test cases and run them with the correct marker/level strategy for [vllm-project/vllm-omni](https://github.com/vllm-project/vllm-omni).

**Link convention:** Paths such as `.buildkite/` and `docs/contributing/` live at the **vllm-omni repo root**. Markdown links use repo-relative paths from this skill file (e.g. `../../../.buildkite/cuda/test-ready.yml`, `../../../docs/contributing/ci/test_system_overview.md`).

Default priorities:

1. Reproducible regression coverage for bug fixes
2. Correct test level and marker selection
3. Low flake, low dependency tests first
4. CI-compatible run commands
5. **Actionable run commands for the human**: whenever you add or change tests, always finish with **copy-paste-ready** `pytest` lines (local: single file and/or single test; CI-like: markers + `--run-level`), plus short **prerequisites** (GPU tier, HF cache, optional `model_prefix`). Do not assume the reader will infer commands from `test-routing.md` alone.

## Inputs

- Issue/PR link and summary
- Changed files or suspected code path
- Whether the user wants local quick validation or CI-equivalent validation
- Hardware constraints (CPU only / CUDA / ROCm / NPU)

## Workflow

### Step 1: Classify Test Goal

- **Bugfix regression**: start from a minimal failing scenario and add assertions that prevent recurrence. Before writing tests, output **`required` / `recommended` / `not_needed`**: **`required`** — stable logic/contract bug that should have been caught; **`recommended`** — environment-sensitive but a small regression still helps; **`not_needed`** — one-off external/config failure or existing tests already cover the path. Prefer the narrowest stable L1 (CPU) case; escalate to L2/L3 only when the bug needs real weights or serving.
- **Feature coverage**: verify new behavior and one negative/boundary case.
- **Perf/benchmark claim**: require benchmark-oriented tests and explicit metrics.

### Step 2: Select Test Level

- **L1**: unit/logic, deterministic, CPU-friendly, fastest feedback.
- **L2**: basic e2e/integration and platform-dependent checks.
- **L3/L4**: advanced model/integration/perf validation.

Use [references/test-routing.md](references/test-routing.md) for level-to-marker and command mapping.

### Step 3: Pick Markers

Always attach markers deliberately:

- **Level**: `core_model` (L1/L2) and/or `advanced_model` (L3) and/or `full_model` (L4 nightly)
- **Model type** (required on model-centric e2e — pick exactly one):
  - `omni` — end-to-end multimodal LLM pipelines (thinker/talker/stages; Qwen-Omni family)
  - `tts` — speech synthesis / TTS-only models (`/v1/audio/speech`, voice clone, etc.)
  - `diffusion` — generative diffusion models (image / audio / text / video from noise)
- **Cross-cutting area** (when relevant): `parallel`, `cache`, `example`, `benchmark`
- **Hardware**: `cpu`, `gpu`, `cuda`, `rocm`, `npu`, …; SKU (`H100`, `L4`, …) and `cards_{n}` via `@hardware_test(...)` / `hardware_marks(...)` only — never `pytest.mark.H100` (`check-mark` rejects it)
- Optional: `slow`. Multi-card: `num_cards=N` on the helper (filter with `cards_N` / `not cards_1`)

**Baseline smoke (L2 + L3):** The simplest e2e case per model — default deploy, minimal request — should usually carry **both** `@pytest.mark.core_model` and `@pytest.mark.advanced_model` on the **same** test function so `test-ready.yml` and `test-merge.yml` share one test. `send_*_request` picks validation depth from `--run-level`. References: `test_voxcpm2_tts.py::test_text_to_audio_001`, `test_qwen3_tts_customvoice.py::test_text_to_audio_001`. Heavier scenarios use **`advanced_model` only**; L4 expansion uses **`full_model`**.

For hardware-aware tests, prefer `@hardware_test(...)` or `hardware_marks(...)` in `tests/helpers/mark.py`.

**Diffusion nightly split** (`test-nightly.yml`): all diffusion tests use `pytest.mark.diffusion`, but CI groups them by **output modality**, not by separate pytest markers:

| Nightly group | Scope | Typical models / paths |
|---------------|--------|-------------------------|
| **Diffusion X2I(&A&T) Model Test** | x2**i** (image), x2**a** (audio), x2**t** (text) and other **non-video** diffusion | Qwen-Image*, BAGEL, FLUX, SD3, Z-Image, LongCat, DreamZero, … — `test_*_expansion.py` under X2I shards; L4 sweep uses `-k "not test_wan and not test_bagel_expansion and not hunyuan"` for L4 |
| **Diffusion X2V Model Test** | x2**v** (video) only | Wan2.2, HunyuanVideo 1.5, Wan VACE, … — e.g. `test_wan22_expansion.py`, `test_hunyuan_video_15_expansion.py` |

Wire new **video** diffusion expansion tests into the **X2V** group; wire **image/audio/text** diffusion expansion tests into **X2I(&A&T)**. Do not place x2v modules in X2I shards (see comment in `test-nightly.yml` above the X2I group).

### Naming: generated test module files (L2–L4, model-centric e2e)

When adding **new** pytest modules whose primary scope is a **specific model** (typical under `tests/e2e/offline_inference/` or `tests/e2e/online_serving/`), use this filename pattern:

| Level | Filename pattern | Example (model `Qwen/Qwen2.5-Omni-7B`) |
|-------|------------------|----------------------------------------|
| **L2**, **L3** | `test_{lowercase_model_slug}.py` | `test_qwen2_5_omni.py` |
| **L4** | `test_{lowercase_model_slug}_expansion.py` | `test_qwen2_5_omni_expansion.py` |

**Slug rules for `{lowercase_model_slug}`:**

1. Start from the HuggingFace-style id (e.g. `Qwen/Qwen2.5-Omni-7B`), but **do not** put the org into the filename: use the **repo segment only** (`Qwen2.5-Omni-7B`), not `Qwen_...` / `qwen_qwen2_5_...`.
2. **Lowercase**; replace `.`, `-`, and whitespace with a single `_` (e.g. `Qwen2.5-Omni-7B` → `qwen2_5_omni`). **Omit** trailing size tokens such as `7b` / `30b` in the basename when a single file covers that model line in the directory (matches `test_qwen2_5_omni.py` in-tree).
3. If two checkpoints in the same folder need separate modules, add a **minimal** disambiguator (e.g. `_7b` vs `_3b`) only then.
4. **L1** unit tests are **not** bound to this pattern; use `tests/<area>/test_<feature>.py` as today.

Routing tables and commands: [references/test-routing.md](references/test-routing.md) § *Model-centric e2e filename convention*.

Existing references: `tests/e2e/offline_inference/test_qwen2_5_omni.py` (L2-style omni), `tests/e2e/offline_inference/test_qwen3_5_9b.py` (L2-style omni, single-stage VL), `tests/e2e/online_serving/test_qwen3_omni_expansion.py` (L4-style omni), `tests/e2e/online_serving/test_qwen_image_edit_expansion.py` / `test_qwen_image_expansion.py` (L4-style diffusion).

### Step 4: Generate Test Case Skeleton

**1. Pick the functional scenario** (then choose directory, fixtures, and markers):

| Scenario | Typical location | Fixtures / runner pattern | Baseline markers & level |
|----------|------------------|---------------------------|---------------------------|
| **Offline inference e2e** | `tests/e2e/offline_inference/` | **Module (default):** `omni_runner` + `offline_client`. **Function (isolation only):** `omni_runner_function` + `offline_client_function`. Diffusion/TTS may use `Omni(...).generate` directly | L2: `core_model` + **one of** `omni` / `tts` / `diffusion`; `@hardware_test(...)` when GPU/NPU is required |
| **Online serving e2e** | `tests/e2e/online_serving/` | **Module (default):** `omni_server` + `online_client`. **Function (isolation only):** `omni_server_function` + `online_client_function`. Clients: `send_omni_request` (omni), `send_audio_speech_request` (tts), `send_diffusion_request` / `send_video_diffusion_request` / `send_images_generations_request` (diffusion) | Baseline smoke: **`core_model` + `advanced_model`**; heavier paths: `advanced_model` only; L4 expansion: `full_model` |
| **Documentation / runnable examples** | `tests/examples/offline_inference/`, `tests/examples/online_serving/` | **Offline docs (preferred):** extract Python/Bash blocks from the doc README (e.g. `ReadmeSnippet.extract_readme_snippets`), `pytest.mark.parametrize` each snippet, run via `example_runner.run` with a stable `output_subfolder`. **Online docs:** copy client/request scripts into dedicated tests and keep them in sync with the doc page. | Usually **L4**: `advanced_model`, often `example` plus hardware marks matching the nightly docs-example job (see `.buildkite/cuda/test-nightly.yml`). Full conventions: [docs/contributing/ci/test_examples/l4_doc_example_tests.inc.md](../../../docs/contributing/ci/test_examples/l4_doc_example_tests.inc.md) (introduced in [PR #1910](https://github.com/vllm-project/vllm-omni/pull/1910): naming, output directory layout, skip rules, avoid trimming `num_inference_steps` without a strong CI reason). |
| **Performance / benchmark** | `tests/dfx/perf/tests/*.json` + `run_*_benchmark.py` | JSON or script-driven server + load config; assert explicit metrics / baselines | L4 Perf: JSON `mark` with `full_model` + `omni`/`tts`/`diffusion`; wire `test-nightly.yml` Perf steps |
| **Invalid parameter / negative HTTP validation** | `tests/dfx/reliability/invalid_param_test/` | Live `omni_server` + low-level `send_*_http_request` with `err_code` / `err_message` | `pytest.mark.slow` + `omni` / `tts` / `diffusion` + `@hardware_marks` (`H100` or `L4`); CI in **`test-weekly.yml`** (not ready/merge/nightly) |

**If the user’s test plan includes invalid parameter validation / invalid params / negative HTTP / 400 validation:** do **not** add those `test_*` functions to `tests/e2e/online_serving/test_*.py` or `*_expansion.py`. **Move or author them** under `tests/dfx/reliability/invalid_param_test/` in the **endpoint-matching script** (see **Invalid parameter validation** below). Success-path e2e and invalid-param dfx tests must stay in separate modules.

**1b. Model type — `omni` vs `tts` vs `diffusion`**

After choosing offline/online/docs/perf, classify the **product under test** and attach **exactly one** model-type marker. All three can live under the same `tests/e2e/` trees; conventions diverge:

| Dimension | **Omni** (`pytest.mark.omni`) | **TTS** (`pytest.mark.tts`) | **Diffusion** (`pytest.mark.diffusion`) |
|-----------|------------------------------|----------------------------|----------------------------------------|
| **What it is** | Multimodal LLM pipeline (thinker/talker/stages; text + vision + audio I/O) | Speech synthesis / voice models | Generative diffusion (noise → image, audio, text, or video) |
| **Examples** | Qwen2.5-Omni, Qwen3-Omni | Qwen3-TTS, VoxCPM2, Higgs-Audio, Step-Audio2 | Qwen-Image, BAGEL, Wan2.2, HunyuanVideo |
| **Offline runner** | `omni_runner` + `offline_client`, `generate_multimodal` | `Omni(...).generate` with deploy YAML + TTS params | `Omni(...).generate` + `OmniDiffusionSamplingParams` |
| **Online client** | `online_client.send_omni_request` (chat completions, modalities) | `online_client.send_audio_speech_request` (`/v1/audio/speech`) | `send_diffusion_request` (chat/T2I), `send_video_diffusion_request` (`/v1/videos`, X2V), or `send_images_generations_http_request` / `send_images_edits_http_request` (DALL-E routes) |
| **Typical assertions** | Stage outputs, text/audio keywords via `OfflineOmniClient` / response handler | WAV bytes, stream chunks, speech endpoint contract | Image/video dimensions, `final_output_type`, `assert_diffusion_response` |
| **Deploy YAML** | Per-model deploy configs (`ci/qwen3_omni_moe.yaml`, …) | `qwen3_tts.yaml`, `voxcpm2.yaml`, … | Often default serve; parallel/offload YAML for heavy DiT |
| **Nightly group** (`test-nightly.yml`) | **Omni Model Test** — `-m "full_model and omni"` | **TTS Model Test** — `-m "full_model and tts"` | **Diffusion X2I(&A&T)** *or* **Diffusion X2V** (see Step 3 table; same `diffusion` marker, different YAML group / file shard) |
| **L4 pressure** | Expansion per modality/model as needed | Expansion + accuracy/perf in TTS group | X2I: merge feature combos per [#1832](https://github.com/vllm-project/vllm-omni/issues/1832); X2V: separate nightly group |

Do **not** mix fixtures across types (e.g. do not use `omni_runner` layout for a pure diffusion or TTS model without mirroring an in-tree test in that family).

**Diffusion only — X2I(&A&T) vs X2V (nightly routing, not extra markers):**

- **X2I(&A&T)**: image / audio / text generation — Qwen-Image*, FLUX, SD3, Z-Image, BAGEL (expansion), LongCat, audio diffusion, etc.
- **X2V**: video generation only — Wan2.2, HunyuanVideo 1.5, Wan VACE, LTX video similarity paths.

When adding a new `test_*_expansion.py`, place it in the matching **nightly group** step (explicit file list in `test-nightly.yml`), not only by marker expression.

**2. Use the narrowest deterministic skeleton for the scenario**

*L1 unit / logic (CPU-first):*

**Mocking rule (L1 only):** use **pytest** integration — **`mocker`** (`pytest-mock`) or **`monkeypatch`** (built-in). **Do not** import or call **`unittest.mock`** (`patch`, `MagicMock`, `@patch`, `with patch(...)`, etc.) in L1 tests; patches must auto-revert with the test lifecycle.

```python
import pytest

pytestmark = [pytest.mark.core_model, pytest.mark.cpu]


def test_<scenario_name>(mocker):
    # Prefer mocker.patch / mocker.spy / mocker.Mock — not unittest.mock.patch
    fake_fn = mocker.patch("vllm_omni.some.module.expensive_call", return_value=...)
    # Act
    # Assert
    fake_fn.assert_called_once()


def test_<env_or_attr>(monkeypatch):
    # For simple env / attribute substitution without a Mock object
    monkeypatch.setenv("SOME_FLAG", "1")
    monkeypatch.setattr("vllm_omni.some.module.CONST", 42)
    ...
```

**Avoid in L1:**

```python
# BAD — do not use in L1 unit tests
from unittest.mock import patch, MagicMock

@patch("vllm_omni.some.module.fn")
def test_bad(mock_fn): ...

def test_also_bad():
    with patch("...") as m: ...
```

See **L1 unit test constraints (mocking)** below for the full do/don't list.

*Offline multimodal e2e — **Omni** (representative):*

```python
@pytest.mark.core_model
@pytest.mark.omni
@hardware_test(...)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_<scenario>(omni_runner, offline_client) -> None:
    request_config = {"prompts": ..., "modalities": [...]}  # optional: images, videos, audios
    offline_client.send_omni_request(request_config)
```

*Offline generative e2e — **Diffusion** (representative):*

```python
@pytest.mark.core_model
@pytest.mark.diffusion
@hardware_test(...)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_text_to_image_001(offline_client) -> None:
    offline_client.send_diffusion_request({"prompt": "...", "extra_body": {"num_inference_steps": 4, ...}})
```

*Offline TTS e2e — **Qwen3-TTS** (two-stage; representative):*

```python
@pytest.mark.advanced_model
@pytest.mark.tts
@hardware_test(...)
@pytest.mark.parametrize("omni_runner", tts_server_params, indirect=True)
def test_text_to_audio_001(omni_runner, offline_client) -> None:
    offline_client.send_audio_speech_request({
        "input": "...",
        "task_type": "Base",
        "ref_audio": REF_AUDIO_URL,
        "ref_text": REF_TEXT,
    })
```

*Offline TTS e2e — **single-stage** (Coqui XTTS, MOSS-TTS-Nano; representative):*

```python
@pytest.mark.advanced_model
@pytest.mark.tts
@hardware_test(...)
@pytest.mark.parametrize("omni_runner", tts_server_params, indirect=True)
def test_voice_clone_001(offline_client) -> None:
    offline_client.send_single_stage_tts_request({
        "input": "...",
        "language": "en",
        "prompt_audio_path": REF_AUDIO_PATH,
        "response_format": "wav",
        "run_level": "advanced_model",
    })
```

*Online serving e2e — **Omni** (representative):*

```python
@pytest.mark.core_model
@pytest.mark.omni
@hardware_test(...)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_text_to_text_001(omni_server, online_client) -> None:
    request_config = {"model": omni_server.model, "messages": ..., "modalities": ["text"]}
    online_client.send_omni_request(request_config)
```

*Online serving e2e — **TTS** (representative):*

```python
@pytest.mark.core_model
@pytest.mark.advanced_model
@pytest.mark.tts
@hardware_test(...)
@pytest.mark.parametrize("omni_server", tts_server_params, indirect=True)
def test_text_to_audio_001(omni_server, online_client) -> None:
    request_config = {"model": omni_server.model, "input": "...", "response_format": "wav", ...}
    online_client.send_audio_speech_request(request_config)
```

*Online serving e2e — **Diffusion** X2I (representative):*

```python
@pytest.mark.core_model
@pytest.mark.diffusion
@pytest.mark.parametrize("omni_server", _get_default_case(MODEL), indirect=True)
def test_text_to_image_001(omni_server, online_client) -> None:
    online_client.send_diffusion_request({...})  # chat completions + extra_body
```

*Online serving e2e — **Diffusion** X2V (representative):*

```python
@pytest.mark.core_model
@pytest.mark.diffusion
def test_text_to_video_001(omni_server, online_client) -> None:
    online_client.send_video_diffusion_request({"model": ..., "form_data": {...}})  # /v1/videos
```

*Documentation example tests:* follow the **Preferred Test Strategy** in [l4_doc_example_tests.inc.md](../../../docs/contributing/ci/test_examples/l4_doc_example_tests.inc.md): dynamic extraction for offline READMEs; explicit copied client code for online pages until extraction is justified; use the documented **naming**, **output directory** (page folder + case id), and **skipping** rules (e.g. Gradio-only scripts).

*Performance tests:* add or extend entries under `tests/dfx/perf/tests/` (and JSON configs where the project uses them), with **explicit baselines**, **`mark`** on each case (`hardware_marks` + `full_model` + type marker), and the same nightly Perf step pattern as in-tree configs.

**3. Cross-cutting rules**

- New `.py` files need the Omni SPDX header (`Copyright contributors to the vLLM-Omni project`, not `vLLM project`). `tests/` may keep stdlib `re`/`base64`; pickle is still banned unless the file is already on the pickle allowlist. Every collected `tests/**/test_*.py` needs a CI **level** mark and a **hardware** platform mark/helper (`check-mark` runs locally; GHA skips it). Do not write `pytest.mark.H100` / `pytest.mark.L4`; use the helpers so `cards_{n}` is attached. Do not grow `allowed_files` to land a test. Policy: [docs/contributing/README.md](../../../docs/contributing/README.md#linting).
- Reuse existing fixtures for the chosen scenario; do not mix “online client” assumptions into offline `OmniRunner` tests without a clear reason.
- Avoid external network dependency in assertions unless the scenario is explicitly “online serving” or doc examples that require a model hub (then align with CI secrets/cache).
- Keep **one test function = one intent** (one modality combo, one endpoint contract, or one acceleration combo).
- **E2E test function layout**: **one case → one `test_<scenario>` function** with a name that states what is validated (endpoint, size/`n`, server flag, or route). **Do not** merge multiple cases into a single test that branches on `request.node.callspec.id`, `param.id`, or `if case_id == ...`. Use `@pytest.mark.parametrize("omni_server", [...], indirect=True)` per function (usually **one** `OmniServerParams` per test). A loop inside one test is OK only when it serves **that function’s single intent** (e.g. three standard sizes in `test_*_sizes_256_512_1024`).
- **Runtime fixture scope** (`tests/helpers/fixtures/runtime.py`): default **`omni_server` / `omni_runner`** (module) + matching client/handler; use **`omni_server_function` / `omni_runner_function`** only when each `test_*` must start a fresh instance (see below).
- **L1 mocks**: never `unittest.mock`; use `mocker` or `monkeypatch` only (see below).
- **API calls (L2+ e2e, online and offline)**: **reuse** `send_*_request` in `tests/helpers/runtime.py` when it exists; **otherwise add** the helper in `runtime.py` first, then call it from the test. General `assert_*` inside `send_*_request`; special `assert_*` only in the test. See **Runtime send helpers** below — do not call `omni.generate`, raw HTTP, or SDK clients from `test_*.py`.
- **Response assertions**: reusable checks on API bodies / decoded media belong in **`tests/helpers/assertions.py`** — not as private `_assert_*` helpers inside `test_*.py` (see below).
- **Model-specific payloads stay in test modules** — per-model `MODEL`, deploy path, `REF_AUDIO_URL`, `get_prompt()`, `_build_request()`, and inline `request_config` dicts live in `test_{slug}.py` / `test_{slug}_expansion.py` (and offline/L1 siblings). **Do not** create `tests/helpers/{slug}.py` to deduplicate them; a little copy across files is preferred (see `test_glm_tts.py`, `test_cosyvoice3_tts_expansion.py`). `tests/helpers/` is repo-wide harness only (`mark`, `media`, `runtime`, `stage_config`, `assertions`, `fixtures/`).

#### L1 unit test constraints (mocking)

L1 tests (`core_model and cpu`, under `tests/diffusion/`, `tests/engine/`, `tests/model_executor/`, etc.) must follow the repo’s **pytest-mock** convention (`pytest-mock>=3.10.0` in `pyproject.toml` `[project.optional-dependencies] dev`).

| Do | Don't |
|----|-------|
| `def test_foo(mocker):` + `mocker.patch(...)`, `mocker.spy(...)`, `mocker.Mock()`, `mocker.MagicMock()`, `mocker.AsyncMock()` | `from unittest.mock import patch, MagicMock, Mock, AsyncMock` |
| `def test_bar(monkeypatch):` + `monkeypatch.setattr` / `setenv` / `delenv` / `setitem` | `@patch(...)` decorator |
| Let `mocker` auto-stop patches after the test | `with patch(...):` / `patch.object(...)` context managers |
| Mirror neighboring L1 tests in the same directory | `unittest.mock.create_autospec` unless an existing file already documents an exception |

**Rationale:** `mocker` ties patch lifecycle to pytest fixtures (no leaked patches across tests). `unittest.mock` decorators/context managers are easy to compose incorrectly with parametrized or async tests and are inconsistent with in-tree L1 style.

**Minimal patterns:**

```python
def test_returns_cached_config(mocker):
    loader = mocker.patch(
        "vllm_omni.foo.load_yaml",
        return_value={"stages": []},
    )
    result = get_deploy_config("ci/foo.yaml")
    assert result["stages"] == []
    loader.assert_called_once_with("ci/foo.yaml")


def test_skips_when_env_unset(monkeypatch):
    monkeypatch.delenv("VLLM_OMNI_FEATURE", raising=False)
    assert should_enable_feature() is False
```

E2E levels (L2+) generally **avoid mocks**; if a rare L2 stub is unavoidable, still prefer `mocker` over `unittest.mock` for consistency.

#### Runtime fixtures — scope (`tests/helpers/fixtures/runtime.py`)

vLLM-Omni e2e tests start a real **OmniServer** (online) or **OmniRunner** (offline). Pick scope by **how often the process must be recreated**, not by test level alone.

| Scope | Online fixtures | Offline fixtures | When to use |
|-------|-----------------|------------------|-------------|
| **Module** (default) | `omni_server` → `online_client` | `omni_runner` → `offline_client` | **Default for L2/L3/L4 expansion** — amortize model/server init across `test_*` in the same module. Same `OmniServerParams` / runner config can be reused by multiple tests. |
| **Function** | `omni_server_function` → `online_client_function` | `omni_runner_function` → `offline_client_function` | **Only when required** — each `test_*` must get a **clean** server/runner (no shared engine/GPU state). Typical: `tests/dfx/reliability/`, sleep/wakeup, crash/restart, tests that mutate global server state. |

**Rules:**

1. **Default to module scope** (`omni_server` / `omni_runner`) unless the scenario **explicitly** needs a fresh instance per test function.
2. **Indirect parametrize name must match the fixture name**: `@pytest.mark.parametrize("omni_server", ...)` with `omni_server` + `online_client`; `@pytest.mark.parametrize("omni_server_function", ...)` with `omni_server_function` + `online_client_function`. Do not mix module fixture with function client (or vice versa).
3. Different `OmniServerParams` per `test_*` (e.g. default vs `--enable-cpu-offload`) is still OK with **module** `omni_server` — pytest parametrizes per test node; only switch to `_function` when isolation between tests matters, not merely because `server_args` differ.
4. One `test_*` with **many** server configs (expansion matrix) → single function + `@pytest.mark.parametrize("omni_server", [...], indirect=True)` + module `omni_server` (see in-tree `test_qwen_image_expansion.py`).

```python
# Default — module-scoped server (L4 expansion)
@pytest.mark.parametrize("omni_server", [pytest.param(OmniServerParams(model=MODEL), marks=H100)], indirect=True)
def test_foo_images_generations_default_1024(omni_server, online_client) -> None:
    online_client.send_images_generations_request({...})

# Function-scoped — reliability / per-test clean state only
@pytest.mark.parametrize(
    "omni_server_function",
    [pytest.param(OmniServerParams(model=MODEL), marks=H100)],
    indirect=True,
)
def test_foo_sleep_wakeup_cycle(omni_server_function, online_client_function) -> None:
    online_client_function.send_omni_sleep_http_request({...})
    online_client_function.send_omni_wakeup_http_request({...})
```

#### Runtime send helpers — online **and** offline (`tests/helpers/runtime.py`)

**L2+ e2e (online serving and offline inference) must call APIs through `tests/helpers/runtime.py`.** Fixtures live in `tests/helpers/fixtures/runtime.py`; **send/assert implementation** lives in `tests/helpers/runtime.py` + `tests/helpers/assertions.py`.

| Principle | Action |
|-----------|--------|
| **Reuse first** | Grep `runtime.py` for an existing **`send_*_request`** on `OnlineOmniClient` (online) or `OfflineOmniClient` (offline) that matches the endpoint / pipeline shape. |
| **Extend when close** | If an existing `send_*_request` almost fits (e.g. missing one optional field), extend it in `runtime.py` — do not fork logic in the test file. |
| **Add when missing** | No suitable helper → add **`send_<feature>_request`** (high-level: call + general `assert_*`) or **`send_<route>_<verb>_http_request`** (low-level HTTP for negative/dfx) in **`runtime.py` first**, then call it from tests. |
| **Test module owns payload only** | In `test_*.py`: `MODEL`, deploy path, vendored media, `get_prompt()`, and inline **`request_config` dicts** only. **No** `omni.generate(...)`, raw `requests.post`, OpenAI SDK calls, or `_collect_audio()` / `_process_output()` in e2e tests. |

**Online** (`online_client` from `omni_server`): `OnlineOmniClient.send_*_request`.

**Offline** (`offline_client` from `omni_runner`): `OfflineOmniClient.send_*_request`.

| Do | Don't |
|----|-------|
| Online: `online_client.send_omni_request`, `send_diffusion_request`, `send_audio_speech_request`, `send_video_diffusion_request`, `send_images_generations_request`, … | `requests.post(f"{base_url}/v1/...", json=…)` or `client.chat.completions.create(...)` inside a test |
| Offline: `offline_client.send_omni_request`, `send_diffusion_request`, `send_audio_speech_request`, `send_single_stage_tts_request`, `send_single_stage_tts_batch_request`, … | `omni_runner.omni.generate(...)` + hand-rolled tensor/WAV extraction in `test_*.py` |
| Add missing `send_*` to **`runtime.py`** first; bundle general `assert_*` inside the send helper | A one-off `def _post_*` / `def _collect_audio` at the bottom of a test module |
| Mirror naming/style of neighboring `send_*` (docstring, `request_config` dict, optional `run_level`, `err_code` / `err_message` for negative cases) | Different parameter shapes per test file for the same endpoint |

**Workflow when generating online/offline e2e tests:**

1. Decide whether the needed check is **general** (every success call of this `send_*`) or **special** (one case / one parameter combo only). See **General vs special assert placement** below.
2. **Search `runtime.py`** for a matching **`send_*_request`**; if missing, add it (with general `assert_*` inside) before writing the test body.
3. In the test module: build `request_config` → call **`send_*_request` only** for the general contract; call **`assert_*` in the test file only** for special, case-specific checks (import from `assertions.py`).
4. Reserve **low-level** `send_*_http_request` for negative/dfx tests (`err_code` / `err_message`) — not for ordinary L2+ success-path e2e.
5. When wiring Buildkite `source_file_dependencies`, include **`tests/helpers/runtime.py`** and/or **`tests/helpers/assertions.py`** when new helpers were added.

**Exceptions (document in the test docstring why):** models whose offline prompt path cannot go through existing handlers yet (e.g. `test_higgs_audio_v2.py`, `test_voxtral_tts.py` with custom tokenizer compose) may call `omni.generate` directly until a `send_*_request` is added to `runtime.py` — treat as **debt**, not the default for new TTS/diffusion/omni e2e.

**General vs special assert placement:**

| Kind | Definition | Where it lives | Test file calls |
|------|------------|----------------|-----------------|
| **General** | Default success contract for this endpoint/client method — e.g. HTTP 200, `data[]` shape, `n` count, `size` dimensions, decodable image/audio, bundled omni/diffusion fields | Implement in `assertions.py`, invoke **inside** `send_*_request` in `runtime.py` (after low-level send / SDK call, when `err_code` is not set) | **`send_*_request` only** — do **not** repeat the same `assert_*` |
| **Special** | Extra check tied to **this test case only** — e.g. seed byte-identical replay, accuracy/CLIP threshold, perf ceiling, model-specific optional field | Add or reuse `assert_*` in `assertions.py` (never inline in test) | `send_*_request` **then** `assert_<special>(...)` **once** for that case |

Changing `request_config` fields (`size`, `n`, `seed`, server flags) is **not** special validation — the general `assert_*` should read those from `request_config` inside `send_*`.

**`send_*` ↔ `assert_*` pairing (online `OnlineOmniClient`):**

| High-level `send_*` (prefer in L2+ e2e) | Assert already invoked inside `runtime.py` | Low-level HTTP-only sibling |
|----------------------------------------|--------------------------------------------|-----------------------------|
| `send_omni_request` | `assert_omni_response` | `send_chat_completions_http_request` → `assert_http_error` only |
| `send_diffusion_request` | `assert_diffusion_response` | — |
| `send_audio_speech_request` | `assert_audio_speech_response` | `send_audio_speech_http_request` → `assert_http_error` only |
| `send_video_diffusion_request` | `assert_diffusion_response` | `send_videos_*_http_request` → `assert_http_error` only |
| `send_images_generations_request` *(add when needed)* | **`assert_images_generations_response`** *(general — inside send)* | `send_images_generations_http_request` → `assert_http_error` only |
| `send_images_edits_request` *(add when needed)* | **`assert_images_edits_response`** *(general — inside send)* | `send_images_edits_http_request` → `assert_http_error` only |

**Rule:** Tests call high-level `send_*_request` for the general contract. **Never** call the same bundled `assert_*` again. Call an extra `assert_*` in the test **only** for special, case-specific validation.

**Common `OnlineOmniClient` entry points** (non-exhaustive — grep `runtime.py` before adding):

| Area | High-level (SDK + assert) | Low-level HTTP (`*_http_request`) |
|------|---------------------------|-----------------------------------|
| Omni chat | `send_omni_request` | `send_chat_completions_http_request` |
| Diffusion T2I (chat route) | `send_diffusion_request` | — |
| Diffusion X2V | `send_video_diffusion_request` | `send_videos_create_http_request`, `send_video_content_http_request`, … |
| DALL-E T2I / edit | `send_images_generations_request`, `send_images_edits_request` | `send_images_generations_http_request`, `send_images_edits_http_request` |
| TTS (online) | `send_audio_speech_request` | `send_audio_speech_http_request`, `send_audio_generate_http_request`, … |
| Ops / meta | — | `send_health_http_request`, `send_models_http_request`, `send_omni_sleep_http_request`, … |

**Common `OfflineOmniClient` entry points** (offline — grep `runtime.py` before adding):

| Area | High-level (`send_*_request` + assert) | Notes |
|------|----------------------------------------|-------|
| Omni multimodal | `send_omni_request` | `generate_multimodal` path |
| Diffusion offline | `send_diffusion_request` | chat-route / `OmniTextPrompt` offline |
| Qwen-style TTS | `send_audio_speech_request` | two-stage, `generate_multimodal` + `mm_processor_kwargs` |
| Single-stage TTS (Coqui XTTS, MOSS-TTS-Nano, …) | `send_single_stage_tts_request`, `send_single_stage_tts_batch_request` | `prompt` + `additional_information` + `omni.generate` — **do not** duplicate in tests |
| New model family | **Add** `send_<family>_request` here first | Then call from `tests/e2e/offline_inference/test_*.py` |

L1 tests under `tests/entrypoints/` may use **FastAPI `TestClient`** or direct handler calls with mocks; they do **not** need `OnlineOmniClient` / `OfflineOmniClient`. **L2+ online and offline e2e** must use `runtime.py` `send_*` helpers.

#### Invalid parameter validation (`tests/dfx/reliability/invalid_param_test/`)

When the user asks for **invalid parameter validation**, **invalid request bodies**, **HTTP 4xx contract tests**, or any case that sends **malformed / out-of-range / mismatched** API payloads against a **live** server:

1. **Do not** place these in `tests/e2e/online_serving/` or `*_expansion.py`. If already drafted there, **move** the `test_*` into the correct `invalid_param_test` script and delete the duplicate from e2e.
2. **Pick the script by HTTP route** (extend an existing file; add a new `test_invalid_<area>.py` only when no in-tree script covers that route family):

| API / area | Script |
|------------|--------|
| Omni chat completions, WebSocket video/realtime paths | `test_invalid_omni_chat.py` |
| `POST /v1/audio/speech`, stream, batch, voices | `test_invalid_audio_speech.py` |
| Audio diffusion endpoints | `test_invalid_audio_diffusion.py` |
| `POST /v1/images/generations` | `test_invalid_image_generation.py` |
| `POST /v1/images/edits` | `test_invalid_image_editing.py` |
| `POST/GET/DELETE /v1/videos*` | `test_invalid_video_generation.py` |
| Sleep / wakeup / server control | `test_invalid_server_control.py` |

3. **Match in-tree style** in the chosen script:

| Element | Convention |
|---------|------------|
| **Module markers** | `pytestmark = [pytest.mark.slow, pytest.mark.<omni or tts or diffusion>]` |
| **Server fixture** | `_PARAMS` / `_QWEN3_TTS_SPEECH`-style list of `pytest.param(OmniServerParams(...), id="...", marks=hardware_marks(...))`; `@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)` |
| **Hardware** | `hardware_marks(res={"cuda": "H100"})` for heavy diffusion/omni/video; `hardware_marks(res={"cuda": "L4"})` for smaller TTS models (must match weekly `-m "slow and L4"` step) |
| **HTTP client** | **Low-level** `online_client.send_*_http_request({..., "err_code": 400, "err_message": (...)} )` — **not** `send_*_request` (success path) |
| **Case shape** | Prefer **one `test_*` per route family** + `@pytest.mark.parametrize("body_spec, err_message", [...])` with stable `id=` per case; or dedicated `test_<route>_malformed_json` when not parametrized |
| **Body helpers** | `_minimal_<endpoint>_json(omni_server)` / `_minimal_*_form_data()`; merge overrides with `body.update(body_spec)` |
| **Known gaps** | `pytest.mark.skip(reason="…#3649")` as `_SKIP_ISSUE_3649` when server validation is not yet strict (mirror neighboring cases) |
| **Sections** | Route banner comments (`# ─── POST /v1/images/generations ───`) like existing files |
| **Shared fixtures** | Reuse `tests/dfx/reliability/invalid_param_test/conftest.py` (`tiny_png_bytes`, env defaults) |

4. **Adding a new model** to invalid-param coverage: append a `pytest.param(OmniServerParams(model="...", stage_config_path=..., server_args=...), ...)` entry to the script’s `_PARAMS` list — do **not** create `test_invalid_<model>.py` unless the route is new.

5. **CI — `.buildkite/cuda/test-weekly.yml` only** (not `test-ready.yml` / `test-merge.yml` / `test-nightly.yml`):

| Weekly step | Command | When your cases run |
|-------------|---------|---------------------|
| **Invalid parameters Test · H100** | `pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and H100"` | Diffusion / omni / video invalid-param tests with `H100` hardware mark |
| **Invalid parameters Test · L4** | `pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and L4"` | TTS / lighter models marked `L4` |

- **Trigger:** `build.env("WEEKLY") == "1"` or PR label `weekly-test`.
- **Default:** extending an existing `invalid_param_test` script needs **no YAML edit** — the weekly steps already sweep the whole directory.
- **Edit YAML only** when adding a **new hardware queue**, a **new top-level script** that must run in isolation, or a **model-specific weekly shard** (mirror neighboring reliability steps).
- Weekly steps **do not** use `source_file_dependencies`.

**Example — append to `test_invalid_image_generation.py`:**

```python
@pytest.mark.parametrize(
    "body_spec, err_message",
    [
        pytest.param({"seed": -1}, ("seed", "greater_than_equal", "0"), id="seed_negative"),
    ],
)
@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)
def test_images_generations_invalid_requests(
    omni_server: OmniServer,
    online_client: OnlineOmniClient,
    body_spec: dict[str, object],
    err_message: str | tuple[str, ...],
) -> None:
    body = _minimal_images_gen_json(omni_server)
    body.update(body_spec)
    online_client.send_images_generations_http_request(
        {"json": body, "timeout": 300, "err_code": 400, "err_message": err_message}
    )
```

See [references/test-routing.md](references/test-routing.md) **Invalid parameter / weekly CI**.

#### Assertion helpers (`tests/helpers/assertions.py`)

**Do not** add module-local helpers such as `_assert_images_generations_payload` or `_send_and_assert_*` in e2e test files. Response/media validation belongs in **`tests/helpers/assertions.py`**, grouped by **category**:

| Category | Existing anchors | When to extend vs add |
|----------|------------------|------------------------|
| **Image** (chat diffusion + DALL-E JSON) | `assert_image_diffusion_response`, `assert_image_valid` | DALL-E `/v1/images/generations` JSON → add **`assert_images_generations_response`** beside image helpers (reuses `assert_image_valid`); do not fork decode logic into tests |
| **Video** | `assert_video_diffusion_response`, `assert_video_valid` | Extend these for new video contracts |
| **Audio** | `assert_audio_diffusion_response`, `assert_audio_speech_response`, `assert_audio_valid` | Extend for new TTS/audio endpoints |
| **Omni multimodal** | `assert_omni_response` | Extend for new modality combos |
| **HTTP errors** | `assert_http_error`, `assert_err_message_in_text` | Used inside low-level `send_*_http_request` and negative tests |

| Do | Don't |
|----|-------|
| Put **general** `assert_*` inside `send_*_request` in `runtime.py`; tests call `send_*_request` only | Call `assert_diffusion_response` after `send_diffusion_request` (general assert already bundled) |
| Put **special** `assert_*` in `assertions.py` and call it **in the test** after `send_*_request` when that case needs extra checks | Put general decode/count/size logic in the test because “this case uses `n=4`” (that belongs in general assert reading `request_config`) |
| Extend the matching category function, or add `assert_<endpoint>_response` next to its category | `_send_and_assert_*` or PIL/base64 loops in `te

…(truncated)