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.

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 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:

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:

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 § 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):

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:

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.

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:

# 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):

@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):

@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):

@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):

@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):

@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):

@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):

@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):

@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: 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.
• 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).

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:

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.

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).

# 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.

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

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

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:

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):

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):

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

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):

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

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):

• Trigger: build.env("WEEKLY") == "1" (all weekly jobs) or PR label weekly-test (Reliability / Perf, then filtered by source_file_dependencies).
• 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 Reliability / Perf jobs declare source_file_dependencies. Filtering applies only on the weekly-test label; WEEKLY=1 on main keeps every job in those groups. CPU coverage (WEEKLY=1 only) and the E2E Tests group (NON_CRITICAL=1 only) do not use this field.

Example — append to test_invalid_image_generation.py:

@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 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:

Workflow when generating tests:

1. Implement or reuse general assert_* in assertions.py (by category).
2. Wire it into send_*_request in runtime.py if not already bundled.
3. In the test: request_config → send_*_request. Add a separate assert_* import/call only for special case validation.
4. Update source_file_dependencies when runtime.py or assertions.py changes.

Example — DALL-E /v1/images/generations:

# tests/helpers/assertions.py — general contract (image category)
def assert_images_generations_response(resp_body: dict, request_config: dict, *, run_level: str | None = None) -> None:
    ...  # data[], n, size → assert_image_valid

# tests/helpers/runtime.py — general assert INSIDE send
def send_images_generations_request(self, request_config: dict[str, Any], ...) -> list[HttpResponse]:
    responses = self.send_images_generations_http_request(request_config, ...)
    if request_config.get("err_code") is None:
        body = responses[0].json_body
        assert isinstance(body, dict)
        assert_images_generations_response(body, request_config, run_level=self.run_level)
    return responses

# tests/e2e/.../test_foo_expansion.py — one case per test_* (no case_id branching)
@pytest.mark.parametrize("omni_server", [pytest.param(OmniServerParams(model=MODEL), marks=SINGLE_L4)], indirect=True)
def test_foo_images_generations_default_1024(omni_server, online_client) -> None:
    online_client.send_images_generations_request({"json": body, "timeout": 300})

@pytest.mark.parametrize("omni_server", [pytest.param(OmniServerParams(model=MODEL), marks=SINGLE_L4)], indirect=True)
def test_foo_images_generations_sizes_256_512_1024(omni_server, online_client) -> None:
    for size in ("256x256", "512x512", "1024x1024"):
        online_client.send_images_generations_request({"json": {**body, "size": size}, "timeout": 300})

# chat diffusion — separate test; send_diffusion_request bundles assert_diffusion_response
@pytest.mark.parametrize("omni_server", [pytest.param(OmniServerParams(model=MODEL), marks=SINGLE_L4)], indirect=True)
def test_foo_chat_completions_t2i_fallback(omni_server, online_client) -> None:
    online_client.send_diffusion_request({...})

L1 tests under tests/entrypoints/ may keep minimal asserts next to the handler under test when validating a single branch; still prefer assertions.py when the same JSON/media check appears in more than one file or level (e.g. L1 protocol test + L4 e2e).