From bd83ac9b4b6f7f3a64d13a1695d5a51e73164075 Mon Sep 17 00:00:00 2001 From: wangyu <53896905+yenuo26@users.noreply.github.com> Date: Tue, 19 May 2026 14:28:47 +0800 Subject: [PATCH] [CI] invalid_param reliability suite and weekly http_invalid jobs (#3652) Signed-off-by: wangyu <410167048@qq.com> --- .buildkite/test-weekly.yml | 62 ++ tests/conftest.py | 2 + .../invalid_param_test/conftest.py | 23 + .../test_invalid_audio_diffusion.py | 118 +++ .../test_invalid_audio_speech.py | 862 ++++++++++++++++++ .../test_invalid_image_editing.py | 90 ++ .../test_invalid_image_generation.py | 142 +++ .../test_invalid_omni_chat.py | 216 +++++ .../test_invalid_server_control.py | 95 ++ .../test_invalid_video_generation.py | 171 ++++ ..._qwen2_5_omni_autoround_w4a16_expansion.py | 10 +- ...st_qwen3_omni_autoround_w4a16_expansion.py | 12 +- .../test_omnivoice_expansion.py | 13 - tests/e2e/online_serving/test_voxtral_tts.py | 17 - .../test_qwen3_tts_speaker_embedding.py | 66 +- .../openai_api/test_serving_speech.py | 6 +- tests/helpers/assertions.py | 177 ++-- tests/helpers/runtime.py | 749 ++++++++++++++- 18 files changed, 2647 insertions(+), 184 deletions(-) create mode 100644 tests/dfx/reliability/invalid_param_test/conftest.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_audio_diffusion.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_audio_speech.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_image_editing.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_image_generation.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_server_control.py create mode 100644 tests/dfx/reliability/invalid_param_test/test_invalid_video_generation.py diff --git a/.buildkite/test-weekly.yml b/.buildkite/test-weekly.yml index 00c9b25f2..57eadeafd 100644 --- a/.buildkite/test-weekly.yml +++ b/.buildkite/test-weekly.yml @@ -84,3 +84,65 @@ steps: hostPath: path: /mnt/hf-cache type: DirectoryOrCreate + + # Invalid HTTP / WS payloads vs live Omni servers (Issue #3649 tracks known skips). + - group: "Reliability Test - Invalid parameters Test" + depends_on: upload-weekly-pipeline + if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test" + steps: + - label: "Invalid parameters Test · H100" + timeout_in_minutes: 120 + commands: + - pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and H100" + 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 + + # Stable Audio Open, Qwen3-TTS speech/batch/voices paths marked ``L4``. + - label: "Invalid parameters Test · L4" + timeout_in_minutes: 120 + commands: + - pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and L4" + agents: + queue: "gpu_1_queue" + plugins: + - docker#v5.2.0: + image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT + always-pull: true + propagate-environment: true + shm-size: "8gb" + environment: + - "HF_HOME=/fsx/hf_cache" + - "HF_TOKEN" + volumes: + - "/fsx/hf_cache:/fsx/hf_cache" diff --git a/tests/conftest.py b/tests/conftest.py index 760404870..163309b8c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -36,6 +36,7 @@ def pytest_terminal_summary(terminalreporter, exitstatus, config): _ASSERTION_EXPORT_NAMES = ( "assert_audio_speech_response", "assert_diffusion_response", + "assert_http_error", "assert_image_diffusion_response", "assert_image_valid", "assert_omni_response", @@ -54,6 +55,7 @@ _MEDIA_EXPORT_NAMES = ( _STAGE_CONFIG_EXPORT_NAMES = ("modify_stage_config",) _RUNTIME_EXPORT_NAMES = ( "DiffusionResponse", + "HttpResponse", "OmniResponse", "OmniRunner", "OmniRunnerHandler", diff --git a/tests/dfx/reliability/invalid_param_test/conftest.py b/tests/dfx/reliability/invalid_param_test/conftest.py new file mode 100644 index 000000000..13fab0bc8 --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/conftest.py @@ -0,0 +1,23 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Shared env for ``http_invalid`` (real :func:`omni_server`).""" + +from __future__ import annotations + +import io +import os + +import pytest +from PIL import Image + +# Match ``tests/e2e/online_serving/*`` module-level env for subprocess serve. +os.environ.setdefault("VLLM_WORKER_MULTIPROC_METHOD", "spawn") +os.environ.setdefault("VLLM_TEST_CLEAN_GPU_MEMORY", "0") + + +@pytest.fixture +def tiny_png_bytes() -> bytes: + img = Image.new("RGB", (32, 32), color="gray") + buf = io.BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_audio_diffusion.py b/tests/dfx/reliability/invalid_param_test/test_invalid_audio_diffusion.py new file mode 100644 index 000000000..ee1515fa6 --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_audio_diffusion.py @@ -0,0 +1,118 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""``POST /v1/audio/generate`` validation (live Stable Audio Open server).""" + +from __future__ import annotations + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler + +pytestmark = [pytest.mark.slow, pytest.mark.diffusion] + +# https://github.com/vllm-project/vllm-omni/issues/3649 — invalid inputs still HTTP 200 / timeout / missing validation. +_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649") + +_PARAMS = [ + pytest.param( + OmniServerParams( + model="stabilityai/stable-audio-open-1.0", + server_args=[ + "--trust-remote-code", + "--enforce-eager", + "--model-class-name", + "StableAudioPipeline", + ], + ), + id="stable_audio_open", + marks=hardware_marks(res={"cuda": "L4"}), + ), +] + + +@pytest.mark.parametrize( + "overrides, err_message", + [ + pytest.param({"input": ""}, ("cannot be empty", "input"), id="input_empty", marks=_SKIP_ISSUE_3649), + pytest.param( + {"input": " "}, ("cannot be empty", "input"), id="input_whitespace_only", marks=_SKIP_ISSUE_3649 + ), + pytest.param({"input": 123}, ("input", "a valid string"), id="input_wrong_type"), + pytest.param( + {"response_format": "mpeg"}, + ("response_format", "literal_error", "Input should be"), + id="response_format_invalid", + ), + pytest.param({"speed": 0.1}, ("speed", "greater_than_equal", "0.25"), id="speed_too_low"), + pytest.param({"speed": 99.0}, ("speed", "less_than_equal", "4"), id="speed_too_high"), + pytest.param({"stream_format": 1}, ("stream_format", "literal_error", "audio"), id="stream_format_wrong_type"), + pytest.param( + {"stream_format": "sse"}, + ("sse", "stream_format", "value_error", "not a supported stream_format"), + id="stream_format_sse_blocked", + ), + pytest.param( + {"audio_length": -1.0}, ("audio_end_in_s", "higher than", "audio_start_in_s"), id="audio_length_negative" + ), + pytest.param({"audio_length": 0}, "audio_length", id="audio_length_zero", marks=_SKIP_ISSUE_3649), + # ``OmniOpenAIServingAudioGenerate`` only forwards ``audio_start`` when ``audio_length`` is set. + pytest.param( + {"audio_start": -0.5, "audio_length": 5.0}, + ("audio_start_in_s", "between", "0", "512"), + id="audio_start_negative", + ), + pytest.param( + {"negative_prompt": ["noise"]}, ("negative_prompt", "a valid string"), id="negative_prompt_wrong_type" + ), + pytest.param({"guidance_scale": -1.0}, "guidance_scale", id="guidance_scale_negative", marks=_SKIP_ISSUE_3649), + pytest.param({"guidance_scale": 0}, "guidance_scale", id="guidance_scale_zero", marks=_SKIP_ISSUE_3649), + pytest.param( + {"num_inference_steps": 0}, "num_inference_steps", id="num_inference_steps_zero", marks=_SKIP_ISSUE_3649 + ), + pytest.param( + {"num_inference_steps": -1}, ("number of steps", "non-negative"), id="num_inference_steps_negative" + ), + pytest.param( + {"num_inference_steps": 6000}, + "num_inference_steps", + id="num_inference_steps_above_max", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + {"guidance_scale": 1001.0}, "guidance_scale", id="guidance_scale_above_max", marks=_SKIP_ISSUE_3649 + ), + pytest.param( + {"audio_length": 86401.0}, ("Requested audio length", "exceeds maximum"), id="audio_length_above_max" + ), + pytest.param( + {"audio_start": 86401.0, "audio_length": 1.0}, + ("audio_start_in_s", "between", "0", "512"), + id="audio_start_above_max", + ), + ], +) +@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True) +def test_audio_generate_invalid_field_values( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + overrides: dict[str, object], + err_message: str | tuple[str, ...], +) -> None: + body = {"model": omni_server.model, "input": "ambient electronic pad"} + body.update(overrides) + openai_client.send_audio_generate_http_request( + {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True) +def test_audio_generate_missing_input(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_audio_generate_http_request( + { + "json": {"model": omni_server.model}, + "timeout": 120, + "err_code": 400, + "err_message": ("missing", "input", "field required"), + } + ) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_audio_speech.py b/tests/dfx/reliability/invalid_param_test/test_invalid_audio_speech.py new file mode 100644 index 000000000..51a0cbe2b --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_audio_speech.py @@ -0,0 +1,862 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""HTTP + WebSocket validation for Qwen3-TTS Base: ``/v1/audio/speech``, ``/v1/audio/speech/stream``, batch, voices.""" + +from __future__ import annotations + +import io +import json +import uuid +import wave +from typing import Any + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.media import load_test_audio_data_url +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler +from tests.helpers.stage_config import get_deploy_config_path + +# Speech / voice-upload numeric caps for these tests only (not exported from vllm_omni). +# Align where possible with serving_speech defaults: _TTS_MAX_INSTRUCTIONS_LENGTH (500), +# _TTS_MAX_NEW_TOKENS_MAX (4096). Voice form max lengths are not public constants in prod; +# values below keep tests self-contained and deterministic. +_SPEECH_API_MAX_INSTRUCTIONS_CHARS = 500 +_SPEECH_API_MAX_NEW_TOKENS = 4096 +_VOICE_UPLOAD_MAX_CONSENT_LEN = 1024 +_VOICE_UPLOAD_MAX_REF_TEXT_CHARS = 8192 +_VOICE_UPLOAD_MAX_SPEAKER_DESCRIPTION_CHARS = 2048 +REF_AUDIO_URL = load_test_audio_data_url("qwen3_tts/clone_2.wav") +REF_TEXT = "Okay. Yeah. I resent you. I love you. I respect you. But you know what? You blew it! And thanks to you." +# Qwen3-TTS 0.6B expects 1024-dim speaker embeddings (see serving_speech / talker hidden_size). +_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM = 1024 +# Pop-only marker for ``test_speech_invalid_field_values``: omit ``ref_audio`` / ``ref_text`` so Base +# ``speaker_embedding`` cases mirror standalone embedding-only JSON (no ref_audio clash). +_SPEECH_INVALID_EMBEDDING_ONLY_BODY = "__speech_invalid_embedding_only_body__" +# Batch-only: merged items inherit batch ``ref_audio`` / ``ref_text`` because ``_pick`` treats item ``None`` +# as "unset". Strip clone fields from the batch envelope before asserting CustomVoice preset errors. +_SPEECH_BATCH_NO_CLONE_FIELDS = "__speech_batch_no_clone_fields__" + +_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649") + + +pytestmark = [pytest.mark.slow, pytest.mark.tts] + +_L4_SPEECH_HW = hardware_marks(res={"cuda": "L4"}) +_SPEECH_SERVER_ARGS = ["--trust-remote-code", "--disable-log-stats"] + +_QWEN3_TTS_SPEECH = [ + pytest.param( + OmniServerParams( + model="Qwen/Qwen3-TTS-12Hz-0.6B-Base", + stage_config_path=get_deploy_config_path("qwen3_tts.yaml"), + server_args=_SPEECH_SERVER_ARGS, + ), + id="qwen3_tts", + marks=_L4_SPEECH_HW, + ), +] + + +def _apply_batch_overrides(body: dict[str, object], *, loc: str, overrides: dict[str, object]) -> None: + if loc == "batch": + body.update(overrides) + return + items = body["items"] + assert isinstance(items, list) and len(items) >= 1 + first = items[0] + assert isinstance(first, dict) + body["items"] = [{**first, **overrides}] + + +def _pcm_wav_mono_bytes(*, duration_s: float = 1.1, sample_rate: int = 44100) -> bytes: + buf = io.BytesIO() + with wave.open(buf, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(sample_rate) + nframes = max(1, int(sample_rate * duration_s)) + wf.writeframes(b"\x00\x00" * nframes) + return buf.getvalue() + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/audio/speech (JSON) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_malformed_json(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_audio_speech_http_request( + { + "raw_body": "{", + "timeout": 120, + "err_code": 400, + "err_message": ("json", "decode", "invalid"), + } + ) + + +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_missing_required_fields(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_audio_speech_http_request( + { + "json": {"model": omni_server.model}, + "timeout": 120, + "err_code": 400, + "err_message": ("input", "field required", "missing"), + } + ) + + +@pytest.mark.parametrize( + "overrides, err_message", + [ + pytest.param({"input": ""}, ("input", "empty"), id="input_empty"), + pytest.param({"input": " "}, ("input", "empty"), id="input_whitespace_only"), + pytest.param({"voice": ""}, "Invalid voice", id="voice_empty", marks=_SKIP_ISSUE_3649), + pytest.param( + {"instructions": 123}, ("instructions", "string_type", "valid string"), id="instructions_wrong_type" + ), + pytest.param( + {"response_format": "mpeg"}, ("response_format", "literal_error", "wav"), id="response_format_invalid" + ), + pytest.param({"stream_format": 1}, ("stream_format", "literal_error", "audio"), id="stream_format_wrong_type"), + pytest.param( + {"stream_format": "sse"}, ("stream_format", "value_error", "audio"), id="stream_format_sse_blocked" + ), + pytest.param({"stream": "wrong_type"}, ("stream", "bool_parsing", "validation error"), id="stream_wrong_type"), + pytest.param( + { + "voice": "vivian", + "task_type": "CustomVoice", + "ref_audio": None, + "ref_text": None, + }, + ("CustomVoice", "does not support"), + id="customvoice_task_not_supported", + ), + pytest.param( + {"task_type": "InvalidEnum"}, ("task_type", "literal_error", "CustomVoice"), id="task_type_invalid" + ), + pytest.param({"language": ""}, ("language", "invalid language", "chinese"), id="language_empty"), + pytest.param({"ref_audio": "ftp://example.com/a.wav"}, ("ref_audio", "URL"), id="ref_audio_bad_scheme"), + pytest.param( + {"ref_audio": "not_a_valid_uri"}, + ("ref_audio", "url"), + id="ref_audio_invalid_uri", + ), + pytest.param({"ref_text": ["x"]}, ("ref_text", "string_type", "valid string"), id="ref_text_wrong_type"), + pytest.param( + {"x_vector_only_mode": "wrong"}, + ("x_vector_only_mode", "valid boolean"), + id="x_vector_only_mode_string_not_boolean", + ), + pytest.param( + {"speaker_embedding": [0.01] * _QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM}, + ("mutually exclusive", "speaker_embedding", "ref_audio"), + id="speaker_embedding_with_ref_audio_mutually_exclusive", + ), + pytest.param({"max_new_tokens": 0}, ("max_new_tokens", "least 1"), id="max_new_tokens_below_min"), + pytest.param( + {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1}, + ("max_new_tokens", "exceed 4096"), + id="max_new_tokens_above_max", + ), + pytest.param({"seed": -1}, ("seed", "greater_than_equal", "0"), id="seed_negative"), + pytest.param({"seed": 2**63}, ("seed", "less_than_equal", "9223372036854775807"), id="seed_above_max"), + pytest.param( + {"initial_codec_chunk_frames": -1}, + ("initial_codec_chunk_frames", "greater_than_equal", "0"), + id="initial_codec_chunk_frames_negative", + ), + pytest.param( + {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)}, + ("instructions", "too long"), + id="instructions_exceed_api_limit", + ), + # Base clone: whitespace-only ref_text fails serving validation (not pydantic). + pytest.param( + { + "task_type": "Base", + "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=", + "ref_text": " ", + }, + ("ref_text", "base task", "non-empty"), + id="ref_text_whitespace_only_base_clone", + ), + pytest.param( + { + _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True, + "task_type": "Base", + "speaker_embedding": [], + }, + ("speaker_embedding", "non-empty"), + id="base_speaker_embedding_empty", + ), + pytest.param( + { + _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True, + "task_type": "Base", + "x_vector_only_mode": True, + "speaker_embedding": [0.01] * 512, + }, + ("speaker_embedding", "dimensions", str(_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM)), + id="base_speaker_embedding_wrong_dim_short", + ), + pytest.param( + { + _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True, + "task_type": "Base", + "x_vector_only_mode": True, + "speaker_embedding": [0.01] * 2048, + }, + ("speaker_embedding", "2048", "dimensions", str(_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM)), + id="base_speaker_embedding_wrong_dim_long", + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_invalid_field_values( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + overrides: dict[str, object], + err_message: str | tuple[str, ...], +) -> None: + ov = dict(overrides) + embedding_only = bool(ov.pop(_SPEECH_INVALID_EMBEDDING_ONLY_BODY, False)) + body = { + "model": omni_server.model, + "input": "Hello.", + "voice": "clone", + "language": "English", + "response_format": "wav", + "ref_audio": REF_AUDIO_URL, + "ref_text": REF_TEXT, + } + if embedding_only: + body.pop("ref_audio", None) + body.pop("ref_text", None) + body.update(ov) + openai_client.send_audio_speech_http_request( + {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/audio/speech/batch (JSON) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_batch_empty_items(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_audio_speech_batch_http_request( + {"json": {"items": []}, "timeout": 120, "err_code": (400, 422), "err_message": ("items", "least 1 item")} + ) + + +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_batch_missing_items(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_audio_speech_batch_http_request( + { + "json": {"model": omni_server.model}, + "timeout": 120, + "err_code": (400, 422), + "err_message": ("items", "field required", "missing"), + } + ) + + +# Align field-level checks with ``POST /v1/audio/speech`` where ``BatchSpeechRequest`` / +# ``SpeechBatchItem`` expose the same knobs. Not on batch schema (single-speech only): +# ``stream``, ``stream_format``, ``seed``, ``speaker_embedding`` — see dedicated speech tests. +@pytest.mark.parametrize( + "loc, overrides, err_message", + [ + pytest.param("item", {"input": ""}, ("input", "empty"), id="item_input_empty", marks=_SKIP_ISSUE_3649), + pytest.param( + "item", + {"input": " "}, + ("input", "empty"), + id="item_input_whitespace_only", + marks=_SKIP_ISSUE_3649, + ), + pytest.param("batch", {"voice": ""}, "Invalid voice", id="batch_voice_empty", marks=_SKIP_ISSUE_3649), + pytest.param("item", {"voice": ""}, "Invalid voice", id="item_voice_empty", marks=_SKIP_ISSUE_3649), + pytest.param( + "batch", + { + "voice": "nonexistent_voice_xyz", + "task_type": "CustomVoice", + "ref_audio": None, + "ref_text": None, + }, + "Invalid voice", + id="batch_voice_unknown_preset", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + { + _SPEECH_BATCH_NO_CLONE_FIELDS: True, + "voice": "nonexistent_voice_xyz", + "task_type": "CustomVoice", + }, + "Invalid voice", + id="item_voice_unknown_preset", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"instructions": 123}, + ("instructions", "string_type", "valid string"), + id="batch_instructions_wrong_type", + ), + pytest.param( + "item", + {"instructions": 123}, + ("instructions", "string_type", "valid string"), + id="item_instructions_wrong_type", + ), + pytest.param( + "batch", + {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)}, + ("instructions", "too long"), + id="batch_instructions_exceed_api_limit", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)}, + ("instructions", "too long"), + id="item_instructions_exceed_api_limit", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"response_format": "mpeg"}, + ("response_format", "literal_error", "wav"), + id="batch_response_format_invalid", + ), + pytest.param( + "item", + {"response_format": "mpeg"}, + ("response_format", "literal_error", "wav"), + id="item_response_format_invalid", + ), + pytest.param("batch", {"speed": 0.1}, ("speed", "greater_than_equal", "0.25"), id="batch_speed_too_low"), + pytest.param("item", {"speed": 10.0}, ("speed", "less_than_equal", "4"), id="item_speed_too_high"), + pytest.param( + "batch", + {"task_type": "InvalidEnum"}, + ("task_type", "literal_error", "CustomVoice"), + id="batch_task_type_invalid", + ), + pytest.param( + "item", + {"task_type": "InvalidEnum"}, + ("task_type", "literal_error", "CustomVoice"), + id="item_task_type_invalid", + ), + pytest.param( + "batch", + {"language": ""}, + ("language", "invalid language", "chinese"), + id="batch_language_empty", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"language": ""}, + ("language", "invalid language", "chinese"), + id="item_language_empty", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"ref_audio": "ftp://example.com/a.wav"}, + ("ref_audio", "URL"), + id="batch_ref_audio_bad_scheme", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"ref_audio": "ftp://example.com/a.wav"}, + ("ref_audio", "URL"), + id="item_ref_audio_bad_scheme", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"ref_audio": "not_a_valid_uri"}, + ("ref_audio", "url"), + id="batch_ref_audio_invalid_uri", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"ref_audio": "not_a_valid_uri"}, + ("ref_audio", "url"), + id="item_ref_audio_invalid_uri", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"ref_text": ["x"]}, + ("ref_text", "string_type", "valid string"), + id="batch_ref_text_wrong_type", + ), + pytest.param( + "item", + {"ref_text": ["x"]}, + ("ref_text", "string_type", "valid string"), + id="item_ref_text_wrong_type", + ), + pytest.param( + "batch", + {"x_vector_only_mode": "wrong_type"}, + ("x_vector_only_mode", "bool_parsing", "validation error"), + id="batch_x_vector_only_mode_wrong_type", + ), + pytest.param( + "item", + {"x_vector_only_mode": "wrong_type"}, + ("x_vector_only_mode", "bool_parsing", "validation error"), + id="item_x_vector_only_mode_wrong_type", + ), + pytest.param( + "batch", + {"max_new_tokens": 0}, + ("max_new_tokens", "least 1"), + id="batch_max_new_tokens_below_min", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"max_new_tokens": 0}, + ("max_new_tokens", "least 1"), + id="item_max_new_tokens_below_min", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1}, + ("max_new_tokens", "exceed 4096"), + id="batch_max_new_tokens_above_max", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1}, + ("max_new_tokens", "exceed 4096"), + id="item_max_new_tokens_above_max", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "batch", + {"initial_codec_chunk_frames": -1}, + ("initial_codec_chunk_frames", "greater_than_equal", "0"), + id="batch_initial_codec_chunk_frames_negative", + ), + pytest.param( + "item", + {"initial_codec_chunk_frames": -1}, + ("initial_codec_chunk_frames", "greater_than_equal", "0"), + id="item_initial_codec_chunk_frames_negative", + ), + # Base clone: whitespace-only ref_text (same serving check as single speech). + pytest.param( + "batch", + { + "task_type": "Base", + "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=", + "ref_text": " ", + }, + ("ref_text", "base task", "non-empty"), + id="batch_ref_text_whitespace_only_base_clone", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "item", + { + "task_type": "Base", + "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=", + "ref_text": " ", + }, + ("ref_text", "base task", "non-empty"), + id="item_ref_text_whitespace_only_base_clone", + marks=_SKIP_ISSUE_3649, + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_batch_invalid_field_values( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + loc: str, + overrides: dict[str, object], + err_message: str | tuple[str, ...], +) -> None: + ov = dict(overrides) + strip_clone = bool(ov.pop(_SPEECH_BATCH_NO_CLONE_FIELDS, False)) + body = { + "model": omni_server.model, + "voice": "clone", + "ref_audio": REF_AUDIO_URL, + "ref_text": REF_TEXT, + "language": "English", + "response_format": "wav", + "items": [{"input": "Hello."}], + } + if strip_clone: + body.pop("ref_audio", None) + body.pop("ref_text", None) + _apply_batch_overrides(body, loc=loc, overrides=ov) + openai_client.send_audio_speech_batch_http_request( + {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# WebSocket /v1/audio/speech/stream +# ───────────────────────────────────────────────────────────────────────────── + +# ``test_speech_stream_invalid_requests``: build ``session.config`` JSON with ``omni_server.model``. +_SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES = object() + + +@pytest.mark.parametrize( + "send_frames_spec, err_message", + [ + pytest.param("{not-json", "Invalid JSON", id="invalid_json"), + pytest.param( + json.dumps({"type": "input.text", "text": "hello"}), + "Expected session.config", + id="wrong_first_message_type", + ), + pytest.param( + _SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES, + ("invalid session config", "response_format", "pcm"), + id="session_config_stream_audio_requires_pcm", + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_speech_stream_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + send_frames_spec: Any, + err_message: str | tuple[str, ...], +) -> None: + """Bad first WebSocket text frames: non-JSON, wrong ``type``, or invalid ``session.config``.""" + if send_frames_spec is _SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES: + send_frames = json.dumps( + { + "type": "session.config", + "model": omni_server.model, + "response_format": "wav", + "stream_audio": True, + } + ) + else: + send_frames = send_frames_spec + assert isinstance(send_frames, str) + openai_client.send_audio_speech_stream_ws_request( + { + "send_frames": send_frames, + "timeout": 120, + "ws_max_size": None, + "ws_json_type": "error", + "err_message": err_message, + } + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/audio/voices (multipart + embedding-only) and DELETE /v1/audio/voices/{name} +# ───────────────────────────────────────────────────────────────────────────── + +# Templates for ``test_voices_create_invalid_requests``: resolved in ``_finalize_voices_form_data``. +_VF_LONG_CONSENT = object() +_VF_LONG_REF_TEXT = object() +_VF_LONG_SPEAKER_DESC = object() +_VF_EMBEDDING_NAN_JSON = object() + + +def _finalize_voices_form_data(template: dict[str, Any], uuid_hex: str) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, val in template.items(): + if val is _VF_LONG_CONSENT: + out[key] = "x" * (_VOICE_UPLOAD_MAX_CONSENT_LEN + 1) + elif val is _VF_LONG_REF_TEXT: + out[key] = "y" * (_VOICE_UPLOAD_MAX_REF_TEXT_CHARS + 1) + elif val is _VF_LONG_SPEAKER_DESC: + out[key] = "z" * (_VOICE_UPLOAD_MAX_SPEAKER_DESCRIPTION_CHARS + 1) + elif val is _VF_EMBEDDING_NAN_JSON: + out[key] = json.dumps([0.1] * 1023 + [float("nan")]) + elif isinstance(val, str): + out[key] = val.replace("{uuid}", uuid_hex) + else: + out[key] = val + return out + + +def _voices_upload_multipart_files(kind: str | None) -> dict[str, Any] | None: + """``None`` => no multipart file (embedding-only or neither).""" + if kind is None: + return None + if kind == "wav_ok": + return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(), "audio/wav")} + if kind == "wav_short": + return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(duration_s=0.3), "audio/wav")} + if kind == "wav_bad_body": + return {"audio_sample": ("clip.wav", b"not valid wav content", "audio/wav")} + if kind == "wav_pdf_type": + return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(), "application/pdf")} + raise AssertionError(f"unknown multipart kind {kind!r}") + + +@pytest.mark.parametrize( + "multipart_kind, form_template, err_message", + [ + pytest.param( + None, + {"consent": "c1", "name": "v1"}, + ("audio_sample", "speaker_embedding", "provided"), + id="neither_audio_nor_embedding", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + {"name": "v_miss_consent_{uuid}"}, + ("missing", "consent", "field required"), + id="missing_consent", + ), + pytest.param( + "wav_ok", + {"consent": "consent_ok"}, + ("missing", "name", "field required"), + id="missing_name", + ), + pytest.param( + "wav_ok", + {"consent": "", "name": "v_empty_consent_{uuid}"}, + ("missing", "consent", "field required"), + id="empty_consent", + ), + pytest.param( + "wav_ok", + {"consent": " ", "name": "v_ws_consent_{uuid}"}, + "consent", + id="whitespace_consent", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + {"consent": "bad/consent", "name": "v_bad_cons_{uuid}"}, + "consent", + id="consent_path_sep", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + {"consent": _VF_LONG_CONSENT, "name": "v_long_cons_{uuid}"}, + ("consent", "too long", "Failed to save"), + id="consent_too_long", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + {"consent": "consent_ok", "name": ""}, + ("missing", "name", "field required"), + id="empty_name", + ), + pytest.param( + "wav_ok", + {"consent": "consent_ok", "name": "evil/name"}, + ("voice name", "invalid voice name"), + id="invalid_name_path_sep", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + {"consent": "consent_ok", "name": "v_long_ref_{uuid}", "ref_text": _VF_LONG_REF_TEXT}, + "ref_text", + id="ref_text_too_long", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + { + "consent": "consent_ok", + "name": "v_long_desc_{uuid}", + "speaker_description": _VF_LONG_SPEAKER_DESC, + }, + "speaker_description", + id="speaker_description_too_long", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_pdf_type", + {"consent": "consent_ok", "name": "v_bad_mime_{uuid}"}, + ("mime", "unsupported", "audio/mp4"), + id="audio_unsupported_mime", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_short", + {"consent": "consent_ok", "name": "v_short_{uuid}"}, + ("too short", "reference audio"), + id="audio_too_short", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_bad_body", + {"consent": "consent_ok", "name": "v_bad_wav_{uuid}"}, + ("decode", "audio", "Format not recognised"), + id="audio_decode_error", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + None, + { + "speaker_embedding": "not valid json [[[", + "consent": "consent_emb", + "name": "v_bad_json_{uuid}", + }, + ("speaker_embedding", "valid JSON"), + id="speaker_embedding_invalid_json", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + None, + { + "speaker_embedding": json.dumps([]), + "consent": "consent_emb", + "name": "v_empty_emb_{uuid}", + }, + ("speaker_embedding", "non-empty"), + id="speaker_embedding_empty", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + None, + { + "speaker_embedding": _VF_EMBEDDING_NAN_JSON, + "consent": "consent_emb", + "name": "v_nan_emb_{uuid}", + }, + ("speaker_embedding", "finite"), + id="speaker_embedding_nan", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + None, + { + "speaker_embedding": json.dumps([0.01] * 512), + "consent": "consent_emb", + "name": "v_wrong_dim_{uuid}", + }, + ("expected", "dimensions", "1024"), + id="speaker_embedding_wrong_dim", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "wav_ok", + { + "consent": "consent_emb", + "name": "v_both_{uuid}", + "speaker_embedding": json.dumps([0.01] * 1024), + }, + ("mutually exclusive", "audio_sample", "speaker_embedding"), + id="audio_and_embedding_mutually_exclusive", + marks=_SKIP_ISSUE_3649, + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_voices_create_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + multipart_kind: str | None, + form_template: dict[str, Any], + err_message: str | tuple[str, ...], +) -> None: + """Invalid ``POST /v1/audio/voices`` bodies (multipart or embedding-only).""" + uid = uuid.uuid4().hex[:8] + cfg: dict[str, Any] = { + "data": _finalize_voices_form_data(form_template, uid), + "timeout": 120, + "err_code": 400, + "err_message": err_message, + } + files = _voices_upload_multipart_files(multipart_kind) + if files is not None: + cfg["files"] = files + openai_client.send_audio_voices_create_http_request(cfg) + + +@pytest.mark.parametrize( + "voice_name, err_code, err_message", + [ + pytest.param( + "missing-voice-404", + 404, + "not found", + id="not_found", + ), + pytest.param( + " ", + 404, + "not found", + id="whitespace_only", + ), + pytest.param( + "evil/name", + 404, + "not found", + id="path_separator", + ), + pytest.param( + ".", + 405, + ("Method Not Allowed", "detail"), + id="dot_single", + ), + pytest.param( + "..", + 404, + "not found", + id="dot_dot", + ), + pytest.param( + "bad\x00voice", + 404, + "not found", + id="nul_byte", + ), + pytest.param( + "", + 405, + "Method Not Allowed", + id="empty_name_path", + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True) +def test_voices_delete_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + voice_name: str, + err_code: int | tuple[int, ...], + err_message: str | tuple[str, ...], +) -> None: + """DELETE ``/v1/audio/voices/{name}``: missing voice (404), odd segments (often 404), ``.``paths (405).""" + openai_client.send_audio_voices_delete_http_request( + { + "name": voice_name, + "timeout": 120, + "err_code": err_code, + "err_message": err_message, + } + ) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_image_editing.py b/tests/dfx/reliability/invalid_param_test/test_invalid_image_editing.py new file mode 100644 index 000000000..fa7f0f905 --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_image_editing.py @@ -0,0 +1,90 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""``POST /v1/images/edits`` multipart validation (live ``Qwen/Qwen-Image-Edit``).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler + +pytestmark = [pytest.mark.slow, pytest.mark.diffusion] + +_PARAMS = [ + pytest.param( + OmniServerParams(model="Qwen/Qwen-Image-Edit"), + id="qwen_image_edit", + marks=hardware_marks(res={"cuda": "H100"}), + ), +] + +# Populated in ``test_images_edits_invalid_requests`` from ``omni_server.model``. +_IMAGE_EDITS_SERVER_MODEL = object() + + +def _finalize_edits_form_data(template: dict[str, Any], server_model: str) -> dict[str, Any]: + return {k: (server_model if v is _IMAGE_EDITS_SERVER_MODEL else v) for k, v in template.items()} + + +@pytest.mark.parametrize( + "include_image, data_template, err_message, err_code", + [ + pytest.param( + False, + {"prompt": "make it brighter", "model": _IMAGE_EDITS_SERVER_MODEL}, + ("image", "required", "Unprocessable Entity"), + 422, + id="missing_image", + ), + pytest.param( + True, + {"prompt": "edit", "model": "wrong-model-id"}, + ("model", "mismatch"), + 400, + id="model_mismatch", + ), + pytest.param( + True, + {"prompt": "edit", "response_format": "url"}, + ("response_format", "b64_json", "supported"), + 400, + id="unsupported_response_format", + ), + pytest.param( + True, + {"prompt": "edit", "model": _IMAGE_EDITS_SERVER_MODEL, "output_compression": "101"}, + ("less_than_equal", "output_compression", "100"), + 400, + id="output_compression_above_max", + ), + pytest.param( + True, + {"prompt": "edit", "model": _IMAGE_EDITS_SERVER_MODEL, "layers": "2"}, + ("Invalid layers", "3", "1"), + 400, + id="invalid_layers", + ), + ], +) +@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True) +def test_images_edits_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + tiny_png_bytes: bytes, + include_image: bool, + data_template: dict[str, Any], + err_message: str | tuple[str, ...], + err_code: int | tuple[int, ...], +) -> None: + cfg: dict[str, Any] = { + "data": _finalize_edits_form_data(data_template, omni_server.model), + "timeout": 300, + "err_code": err_code, + "err_message": err_message, + } + if include_image: + cfg["files"] = {"image": ("x.png", tiny_png_bytes, "image/png")} + openai_client.send_images_edits_http_request(cfg) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_image_generation.py b/tests/dfx/reliability/invalid_param_test/test_invalid_image_generation.py new file mode 100644 index 000000000..6f7cf984d --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_image_generation.py @@ -0,0 +1,142 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""``POST /v1/images/generations`` validation (live ``Qwen/Qwen-Image``).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler + +pytestmark = [pytest.mark.slow, pytest.mark.diffusion] + +_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649") + +_PARAMS = [ + pytest.param( + OmniServerParams(model="Qwen/Qwen-Image"), + id="qwen_image", + marks=hardware_marks(res={"cuda": "H100"}), + ), +] + +# Full replacement body for ``test_images_generations_invalid_requests`` (no minimal prompt/model merge). +_IMAGES_GEN_MISSING_PROMPT_JSON = object() + + +def _minimal_images_gen_json(omni_server: OmniServer) -> dict[str, object]: + return {"prompt": "a simple red apple icon", "model": omni_server.model} + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/images/generations (JSON) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "body_spec, err_message", + [ + pytest.param(_IMAGES_GEN_MISSING_PROMPT_JSON, ("prompt", "Field required", "Missing"), id="missing_prompt"), + pytest.param( + {"prompt": "cat", "model": "not-the-loaded-checkpoint"}, + ("model", "mismatch"), + id="model_mismatch", + ), + pytest.param({"prompt": ""}, ("prompt", "empty"), id="prompt_empty", marks=_SKIP_ISSUE_3649), + pytest.param({"prompt": " "}, ("prompt", "empty"), id="prompt_whitespace_only", marks=_SKIP_ISSUE_3649), + pytest.param({"prompt": 123}, ("prompt", "string_type", "valid string"), id="prompt_wrong_type"), + pytest.param({"n": 0}, ("n", "greater_than_equal", "1"), id="n_below_min"), + pytest.param({"n": 11}, ("n", "less_than_equal", "10"), id="n_above_max"), + pytest.param({"size": "1024"}, ("size", "value_error", "WIDTHxHEIGHT"), id="size_missing_height_token"), + pytest.param({"size": "not-a-size"}, ("size", "value_error", "WIDTHxHEIGHT"), id="size_missing_separator"), + pytest.param({"size": "1024xabc"}, ("Invalid size format", "integers"), id="size_non_integer_height"), + pytest.param({"size": "1024x"}, ("Invalid size format", "integers"), id="size_incomplete_height"), + pytest.param({"size": "0x1024"}, ("Invalid size", "positive integers"), id="size_non_positive_width"), + pytest.param( + {"response_format": "url"}, + ("response_format", "value_error", "b64_json"), + id="response_format_not_b64_json", + ), + pytest.param( + {"use_system_prompt": "bogus"}, + ("use_system_prompt", "value_error", "dynamic"), + id="use_system_prompt_invalid", + ), + pytest.param( + {"num_inference_steps": 0}, + ("num_inference_steps", "'greater_than_equal", "1"), + id="num_inference_steps_below_min", + ), + pytest.param( + {"num_inference_steps": 201}, + ("num_inference_steps", "less_than_equal", "200"), + id="num_inference_steps_above_max", + ), + pytest.param( + {"guidance_scale": -1.0}, ("guidance_scale", "greater_than_equal", "0"), id="guidance_scale_negative" + ), + pytest.param( + {"guidance_scale": 21.0}, ("guidance_scale", "less_than_equal", "20"), id="guidance_scale_above_max" + ), + pytest.param( + {"true_cfg_scale": -1.0}, ("true_cfg_scale", "greater_than_equal", "0"), id="true_cfg_scale_negative" + ), + pytest.param( + {"true_cfg_scale": 21.0}, ("true_cfg_scale", "less_than_equal", "20"), id="true_cfg_scale_above_max" + ), + pytest.param({"layers": 2}, ("layers", "value_error", "3", "10"), id="layers_below_min"), + pytest.param({"layers": 11}, ("layers", "value_error", "3", "10"), id="layers_above_max"), + pytest.param({"seed": -1}, ("seed", "greater_than_equal", "0"), id="seed_negative", marks=_SKIP_ISSUE_3649), + pytest.param( + {"seed": 2**32}, + ("seed", "less_than_equal", "4294967295"), + id="seed_above_uint32", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + {"output_format": "gif"}, + ("output_format", "value_error", "b64_json"), + id="output_format_invalid", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + {"vae_use_slicing": "wrong_type"}, + ("vae_use_slicing", "bool_parsing", "validation error"), + id="vae_use_slicing_wrong_type", + ), + pytest.param({"user": 123}, ("user", "string_type", "valid string"), id="user_wrong_type"), + pytest.param( + {"negative_prompt": ["noise"]}, + ("negative_prompt", "string_type", "valid string"), + id="negative_prompt_wrong_type", + ), + pytest.param( + {"generator_device": 1}, + ("generator_device", "string_type", "valid string"), + id="generator_device_wrong_type", + ), + pytest.param( + {"lora": {"foo": "bar"}}, ("lora", "both name and path", "required"), id="lora_missing_required_fields" + ), + ], +) +@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True) +def test_images_generations_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + body_spec: Any, + err_message: str | tuple[str, ...], +) -> None: + """Invalid JSON bodies for ``POST /v1/images/generations`` (missing prompt, model mismatch, bad fields).""" + if body_spec is _IMAGES_GEN_MISSING_PROMPT_JSON: + body: dict[str, object] = {"size": "1024x1024"} + else: + assert isinstance(body_spec, dict) + body = _minimal_images_gen_json(omni_server) + body.update(body_spec) + openai_client.send_images_generations_http_request( + {"json": body, "timeout": 300, "err_code": 400, "err_message": err_message} + ) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py b/tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py new file mode 100644 index 000000000..9d1782e76 --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py @@ -0,0 +1,216 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Invalid inputs on Qwen3-Omni: ``POST /v1/chat/completions``, ``WS /v1/video/chat/stream``, ``WS /v1/realtime``.""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from tests.helpers.mark import hardware_test +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler +from tests.helpers.stage_config import get_deploy_config_path + +pytestmark = [pytest.mark.slow, pytest.mark.omni] + +_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649") + + +def _minimal_chat_json(omni_server: OmniServer) -> dict[str, object]: + """Minimal valid chat body; individual tests override one offending field.""" + return { + "model": omni_server.model, + "messages": [{"role": "user", "content": "hello"}], + "stream": False, + } + + +_QWEN3_OMNI_SERVER = [ + pytest.param( + OmniServerParams( + model="Qwen/Qwen3-Omni-30B-A3B-Instruct", + stage_config_path=get_deploy_config_path("qwen3_omni_moe.yaml"), + server_args=["--async-chunk"], + ), + id="qwen3_omni", + ), +] + + +def _chat_completions_request_without_expectations(omni_server: OmniServer, case_id: str) -> dict[str, Any]: + """Build ``send_chat_completions_http_request`` kwargs excluding ``err_code`` / ``err_message``.""" + if case_id == "malformed_json": + return {"raw_body": "{not-json", "timeout": 120} + if case_id == "missing_messages": + return {"json": {"model": omni_server.model, "stream": False}, "timeout": 120} + if case_id == "model_mismatch": + return { + "json": { + "model": "this-model-is-not-served-on-this-instance", + "messages": [{"role": "user", "content": "hello"}], + "stream": False, + }, + "timeout": 120, + } + body = _minimal_chat_json(omni_server) + if case_id == "stream_wrong_type": + body["stream"] = "not-a-boolean" # type: ignore[assignment] + elif case_id == "temperature_wrong_type": + body["temperature"] = "hot" # type: ignore[assignment] + elif case_id == "max_tokens_wrong_type": + body["max_tokens"] = "not-an-int" # type: ignore[assignment] # type: ignore[assignment] + elif case_id == "modalities_list_bad": + body["modalities"] = [123] # type: ignore[assignment] + elif case_id == "response_format_json_schema_incomplete": + body["response_format"] = {"type": "json_schema"} + elif case_id == "logprobs_wrong_type": + body["logprobs"] = "yes" # type: ignore[assignment] + elif case_id == "logprobs_top_without_enabled": + body["logprobs"] = False + body["top_logprobs"] = 5 + elif case_id == "speaker_unknown": + body["speaker"] = "zz_invalid_qwen3_omni_chat_speaker_xyz" + else: + raise AssertionError(f"unknown chat completions invalid case_id {case_id!r}") + return {"json": body, "timeout": 120} + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/chat/completions +# ───────────────────────────────────────────────────────────────────────────── + + +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize( + "case_id, err_code, err_message", + [ + pytest.param( + "malformed_json", + 400, + ("json_invalid", "decode error"), + id="malformed_json", + ), + pytest.param("missing_messages", 400, ("messages", "missing", "Field required"), id="missing_messages"), + pytest.param( + "model_mismatch", + 404, + ("model", "does not exist", "NotFoundError"), + id="model_mismatch", + ), + pytest.param("stream_wrong_type", 400, ("stream", "bool_parsing", "not-a-boolean"), id="invalid_stream_type"), + pytest.param( + "temperature_wrong_type", + 400, + ("temperature", "float_parsing", "a valid number"), + id="invalid_temperature_type", + ), + pytest.param( + "max_tokens_wrong_type", 400, ("max_tokens", "int_parsing", "integer"), id="invalid_max_tokens_type" + ), + pytest.param( + "modalities_list_bad", + 400, + ("modalities", "value_error", ""), + id="modalities_list_bad_element", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "response_format_json_schema_incomplete", + 400, + ("response_format", "value_error", "json_schema"), + id="invalid_response_format_json_schema", + ), + pytest.param("logprobs_wrong_type", 400, "logprobs", id="logprobs_wrong_type", marks=_SKIP_ISSUE_3649), + pytest.param( + "logprobs_top_without_enabled", + 400, + ("top_logprobs", "logprobs", "True"), + id="logprobs_top_without_logprobs_true", + ), + pytest.param( + "speaker_unknown", + 400, + ("Invalid speaker", "Supported"), + id="speaker_unknown_preset", + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True) +def test_chat_completions_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + case_id: str, + err_code: int | tuple[int, ...], + err_message: str | tuple[str, ...], +) -> None: + cfg = _chat_completions_request_without_expectations(omni_server, case_id) + cfg["err_code"] = err_code + cfg["err_message"] = err_message + openai_client.send_chat_completions_http_request(cfg)[0] + + +# ───────────────────────────────────────────────────────────────────────────── +# WS /v1/video/chat/stream +# ───────────────────────────────────────────────────────────────────────────── + +# First-frame JSON with ``omni_server.model`` for ``test_video_chat_stream_invalid_requests``. +_VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR = object() + + +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize( + "send_frames_spec, err_message", + [ + pytest.param("{not-json", "Invalid JSON", id="invalid_json"), + pytest.param( + _VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR, + "Invalid session config", + id="session_modalities_scalar", + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True) +def test_video_chat_stream_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + send_frames_spec: Any, + err_message: str, +) -> None: + if send_frames_spec is _VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR: + send_frames = json.dumps({"type": "session.config", "model": omni_server.model, "modalities": "text"}) + else: + send_frames = send_frames_spec + assert isinstance(send_frames, str) + openai_client.send_video_chat_stream_ws_request( + { + "send_frames": send_frames, + "timeout": 120, + "ws_max_size": None, + "ws_json_type": "error", + "err_message": err_message, + } + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# WS /v1/realtime +# ───────────────────────────────────────────────────────────────────────────── + + +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) +@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True) +def test_realtime_rejected_when_async_chunk_enabled( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, +) -> None: + openai_client.send_realtime_ws_request( + { + "timeout": 120, + "ws_max_size": None, + "ws_json_type": "error", + "ws_error_code": "unsupported", + "err_message": "async_chunk", + } + ) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_server_control.py b/tests/dfx/reliability/invalid_param_test/test_invalid_server_control.py new file mode 100644 index 000000000..0a8240c4b --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_server_control.py @@ -0,0 +1,95 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Omni sleep/wakeup JSON validation (live ``Qwen/Qwen-Image`` process).""" + +from __future__ import annotations + +from typing import Any, Literal + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler + +pytestmark = [pytest.mark.slow, pytest.mark.diffusion, pytest.mark.omni] + +_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649") + +_QWEN_IMAGE = [ + pytest.param( + OmniServerParams(model="Qwen/Qwen-Image"), + id="qwen_image", + marks=hardware_marks(res={"cuda": "H100"}), + ), +] + + +# ───────────────────────────────────────────────────────────────────────────── +# POST omni sleep / omni wakeup (invalid JSON) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "endpoint, json_body, err_message", + [ + pytest.param("sleep", {"level": 2}, ("stage_ids", "Field required", "Missing"), id="sleep_missing_stage_ids"), + pytest.param( + "sleep", + {"stage_ids": [], "level": 2}, + ("stage_ids", "Field required", "Missing"), + id="sleep_empty_stage_ids", + marks=_SKIP_ISSUE_3649, + ), + pytest.param( + "sleep", + {"stage_ids": "not-a-list", "level": 2}, + ("stage_ids", "list_type", "a valid list"), + id="sleep_invalid_stage_ids_type", + ), + pytest.param( + "sleep", + {"stage_ids": [0], "level": "high"}, + ("level", "int_parsing", "a valid integer"), + id="sleep_invalid_level_type", + ), + pytest.param( + "sleep", + {"stage_ids": [0], "level": -1}, + ("level", "greater"), + id="sleep_negative_level", + marks=_SKIP_ISSUE_3649, + ), + pytest.param("wakeup", {}, ("stage_ids", "missing", "Field required"), id="wakeup_missing_stage_ids"), + pytest.param( + "wakeup", + {"stage_ids": None}, + ("stage_ids", "list_type", "a valid list"), + id="wakeup_invalid_stage_ids_type", + ), + pytest.param( + "wakeup", + {"stage_ids": []}, + ("stage_ids", "length"), + id="wakeup_empty_stage_ids", + marks=_SKIP_ISSUE_3649, + ), + ], +) +@pytest.mark.parametrize("omni_server", _QWEN_IMAGE, indirect=True) +def test_omni_sleep_wakeup_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + endpoint: Literal["sleep", "wakeup"], + json_body: dict[str, Any], + err_message: str | tuple[str, ...], +) -> None: + cfg: dict[str, Any] = { + "json": json_body, + "timeout": 120, + "err_code": 400, + "err_message": err_message, + } + if endpoint == "sleep": + openai_client.send_omni_sleep_http_request(cfg) + else: + openai_client.send_omni_wakeup_http_request(cfg) diff --git a/tests/dfx/reliability/invalid_param_test/test_invalid_video_generation.py b/tests/dfx/reliability/invalid_param_test/test_invalid_video_generation.py new file mode 100644 index 000000000..c0dd133c9 --- /dev/null +++ b/tests/dfx/reliability/invalid_param_test/test_invalid_video_generation.py @@ -0,0 +1,171 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""HTTP video job routes (live ``Wan-AI/Wan2.2-T2V-A14B-Diffusers``). + +WebSocket ``/v1/video/chat/stream`` and ``/v1/realtime`` invalid-input checks live in +``test_invalid_omni_chat.py`` with the same Qwen3-Omni server module fixture (Wan cannot serve those paths). +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from tests.helpers.mark import hardware_marks +from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler + +pytestmark = [pytest.mark.slow, pytest.mark.diffusion] + +_WAN_T2V = [ + pytest.param( + OmniServerParams(model="Wan-AI/Wan2.2-T2V-A14B-Diffusers"), + id="wan22_t2v", + marks=hardware_marks(res={"cuda": "H100"}), + ), +] + + +def _minimal_video_form_data() -> dict[str, str]: + return {"prompt": "a cat walking on grass"} + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/videos (async) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "form_extra, err_message", + [ + pytest.param(None, ("prompt", "Field required", "Missing"), id="missing_prompt"), + pytest.param({"seconds": "0"}, ("seconds", "string_pattern_mismatch", "1"), id="seconds_zero"), + pytest.param({"size": "bad"}, ("size", "string_pattern_mismatch"), id="size_bad"), + pytest.param( + {"num_inference_steps": "0"}, + ("num_inference_steps", "greater_than_equal", "1"), + id="num_inference_steps_zero", + ), + pytest.param( + {"guidance_scale": "25"}, ("guidance_scale", "less_than_equal", "20"), id="guidance_scale_above_max" + ), + ], +) +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_post_videos_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + form_extra: dict[str, str] | None, + err_message: str | tuple[str, ...], +) -> None: + data: dict[str, str] = {} if form_extra is None else {**_minimal_video_form_data(), **form_extra} + openai_client.send_videos_create_http_request( + {"data": data, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# POST /v1/videos/sync +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "form_extra, err_message", + [ + pytest.param(None, ("prompt", "Field required", "Missing"), id="missing_prompt"), + pytest.param({"seconds": "0"}, ("seconds", "string_pattern_mismatch", "1"), id="seconds_zero"), + pytest.param({"size": "bad"}, ("size", "string_pattern_mismatch"), id="size_bad"), + pytest.param( + {"num_inference_steps": "201"}, + ("num_inference_steps", "less_than_equal", "200"), + id="num_inference_steps_above_max", + ), + ], +) +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_post_videos_sync_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + form_extra: dict[str, str] | None, + err_message: str | tuple[str, ...], +) -> None: + data: dict[str, str] = {} if form_extra is None else {**_minimal_video_form_data(), **form_extra} + openai_client.send_videos_sync_http_request( + {"data": data, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# GET /v1/videos (list) +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "params, err_message", + [ + pytest.param({"limit": -1}, ("limit", "greater_than_equal", "0"), id="invalid_limit_negative"), + pytest.param({"limit": 101}, ("limit", "less_than_equal", "100"), id="limit_above_max"), + pytest.param({"order": "newest"}, ("order", "literal_error", "asc", "desc"), id="invalid_order"), + ], +) +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_get_videos_list_invalid_requests( + omni_server: OmniServer, + openai_client: OpenAIClientHandler, + params: dict[str, Any], + err_message: str | tuple[str, ...], +) -> None: + openai_client.send_videos_list_http_request( + {"params": params, "timeout": 120, "err_code": 400, "err_message": err_message} + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# GET /v1/videos/{video_id} +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_get_video_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_video_retrieve_http_request( + { + "video_id": "does-not-exist", + "timeout": 120, + "err_code": 404, + "err_message": ("not found", "video"), + } + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# DELETE /v1/videos/{video_id} +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_delete_video_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_video_delete_http_request( + { + "video_id": "does-not-exist", + "timeout": 120, + "err_code": 404, + "err_message": ("not found", "video"), + } + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# GET /v1/videos/{video_id}/content +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True) +def test_get_video_content_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None: + openai_client.send_video_content_http_request( + { + "video_id": "does-not-exist", + "timeout": 120, + "err_code": 404, + "err_message": ("not found", "video"), + } + ) diff --git a/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16_expansion.py b/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16_expansion.py index 250f37b94..190857e3b 100644 --- a/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16_expansion.py +++ b/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16_expansion.py @@ -70,7 +70,7 @@ def test_text_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_omni_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -93,7 +93,7 @@ def test_audio_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_omni_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -114,7 +114,7 @@ def test_image_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_omni_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -135,7 +135,7 @@ def test_video_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_omni_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -162,4 +162,4 @@ def test_mix_to_audio(omni_runner, omni_runner_handler): "modalities": ["audio"], } response = omni_runner_handler.send_omni_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" diff --git a/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16_expansion.py b/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16_expansion.py index 6cd630e77..9111740ea 100644 --- a/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16_expansion.py +++ b/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16_expansion.py @@ -87,7 +87,7 @@ def test_text_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -111,7 +111,7 @@ def test_audio_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -133,7 +133,7 @@ def test_image_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -155,7 +155,7 @@ def test_video_to_text(omni_runner, omni_runner_handler): "modalities": ["text"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" assert response.text_content and len(response.text_content.strip()) > 0 @@ -177,7 +177,7 @@ def test_video_to_audio(omni_runner, omni_runner_handler): "modalities": ["audio"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" # ------------------------------------------------------------------ @@ -204,4 +204,4 @@ def test_mix_to_audio(omni_runner, omni_runner_handler): "modalities": ["audio"], } response = omni_runner_handler.send_request(request_config) - assert response.success, f"Request failed: {response.error_message}" + assert response.success, "Request failed" diff --git a/tests/e2e/online_serving/test_omnivoice_expansion.py b/tests/e2e/online_serving/test_omnivoice_expansion.py index 7ad3b9b25..631c799cd 100644 --- a/tests/e2e/online_serving/test_omnivoice_expansion.py +++ b/tests/e2e/online_serving/test_omnivoice_expansion.py @@ -105,16 +105,3 @@ class TestOmniVoiceVoiceCloning: "min_audio_bytes": _DEFAULT_MIN_AUDIO_BYTES, } openai_client.send_audio_speech_request(request_config) - - @hardware_test(res={"cuda": "L4"}, num_cards=1) - def test_voice_clone_invalid_ref_audio_format(self, omni_server, openai_client) -> None: - """Test that invalid ref_audio format returns a clear error.""" - request_config = { - "model": omni_server.model, - "input": get_prompt("text"), - "ref_audio": "not_a_valid_uri", - "response_format": "wav", - "timeout": 180.0, - "status_code": (400, 422), - } - openai_client.send_audio_speech_request(request_config) diff --git a/tests/e2e/online_serving/test_voxtral_tts.py b/tests/e2e/online_serving/test_voxtral_tts.py index dff53d539..a1b5fdbb8 100644 --- a/tests/e2e/online_serving/test_voxtral_tts.py +++ b/tests/e2e/online_serving/test_voxtral_tts.py @@ -91,20 +91,3 @@ class TestVoxtralTTSFixedVoice: "timeout": 120.0, } ) - - @pytest.mark.advanced_model - @pytest.mark.tts - @hardware_test(res={"cuda": "H100"}, num_cards=1) - def test_speech_invalid_voice_rejected(self, omni_server, openai_client) -> None: - """Request with a non-existent voice should be rejected (HTTP 200 with JSON or 4xx, depending on server).""" - openai_client.send_audio_speech_request( - { - "model": omni_server.model, - "input": "This voice does not exist.", - "voice": "nonexistent_voice_xyz", - "language": "English", - "response_format": "wav", - "timeout": 120.0, - "err_message": "Invalid speaker", - } - ) diff --git a/tests/entrypoints/openai_api/test_qwen3_tts_speaker_embedding.py b/tests/entrypoints/openai_api/test_qwen3_tts_speaker_embedding.py index 48e3670b4..1f71cf8fb 100644 --- a/tests/entrypoints/openai_api/test_qwen3_tts_speaker_embedding.py +++ b/tests/entrypoints/openai_api/test_qwen3_tts_speaker_embedding.py @@ -5,6 +5,10 @@ E2E tests for speaker_embedding API parameter. Tests Base models (0.6B-Base and 1.7B-Base) which support voice cloning via pre-computed ECAPA-TDNN embeddings, bypassing ref_audio extraction. + +Invalid combinations (mutually exclusive with ref_audio, empty embedding, +wrong embedding dimensions) live in +``tests/dfx/reliability/http_invalid/test_invalid_audio_speech.py``. """ import os @@ -147,45 +151,6 @@ class TestSpeakerEmbeddingBase: assert len(response.content) > MIN_AUDIO_BYTES assert_not_silence(response.content) - @pytest.mark.advanced_model - @pytest.mark.tts - @hardware_test(res={"cuda": "L4"}, num_cards=1) - def test_speaker_embedding_mutually_exclusive_with_ref_audio(self, omni_server) -> None: - """Sending both speaker_embedding and ref_audio returns 400.""" - url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech" - payload = { - "model": omni_server.model, - "input": SYN_TEXT, - "task_type": "Base", - "speaker_embedding": DUMMY_EMBEDDING_1024, - "ref_audio": "https://example.com/audio.wav", - "response_format": "wav", - } - with httpx.Client(timeout=30.0) as client: - response = client.post(url, json=payload) - - assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}" - assert "mutually exclusive" in response.text - - @pytest.mark.advanced_model - @pytest.mark.tts - @hardware_test(res={"cuda": "L4"}, num_cards=1) - def test_speaker_embedding_empty_rejected(self, omni_server) -> None: - """Empty speaker_embedding list returns 400.""" - url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech" - payload = { - "model": omni_server.model, - "input": SYN_TEXT, - "task_type": "Base", - "speaker_embedding": [], - "response_format": "wav", - } - with httpx.Client(timeout=30.0) as client: - response = client.post(url, json=payload) - - assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}" - assert "non-empty" in response.text - # ── 1.7B-Base model tests (2048-dim embeddings) ── @@ -216,26 +181,3 @@ class TestSpeakerEmbedding1_7B: assert response.headers.get("content-type") == "audio/wav" assert verify_wav_audio(response.content), "Response is not valid WAV" assert len(response.content) > MIN_AUDIO_BYTES, f"Audio too small: {len(response.content)} bytes" - - @pytest.mark.advanced_model - @pytest.mark.tts - @hardware_test(res={"cuda": "L4"}, num_cards=1) - def test_1024_dim_on_1_7b_model_rejected(self, omni_server) -> None: - """1024-dim speaker_embedding on a 1.7B model (expects 2048) should fail gracefully.""" - url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech" - payload = { - "model": omni_server.model, - "input": SYN_TEXT, - "task_type": "Base", - "speaker_embedding": DUMMY_EMBEDDING_1024, - "x_vector_only_mode": True, - "response_format": "wav", - "max_new_tokens": MAX_NEW_TOKENS, - } - with httpx.Client(timeout=120.0) as client: - response = client.post(url, json=payload) - - # Wrong dimensions should produce an error, not silent garbage. - assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}" - assert "speaker_embedding has 1024 dimensions" in response.text - assert "expected 2048" in response.text diff --git a/tests/entrypoints/openai_api/test_serving_speech.py b/tests/entrypoints/openai_api/test_serving_speech.py index 72767ec02..2440fe586 100644 --- a/tests/entrypoints/openai_api/test_serving_speech.py +++ b/tests/entrypoints/openai_api/test_serving_speech.py @@ -1,5 +1,6 @@ # tests/entrypoints/openai/test_serving_speech.py import asyncio +import json import logging import os import struct @@ -564,7 +565,6 @@ class TestSpeechAPI: def test_upload_voice_embedding_success(self, client): """Upload a voice via speaker_embedding JSON.""" - import json emb = [0.1] * 1024 data = { @@ -585,7 +585,6 @@ class TestSpeechAPI: def test_upload_voice_embedding_appears_in_listing(self, client): """Embedding-uploaded voice appears in list with correct source.""" - import json emb = [0.2] * 2048 data = { @@ -606,7 +605,6 @@ class TestSpeechAPI: def test_upload_voice_embedding_and_audio_mutually_exclusive(self, client): """Providing both audio_sample and speaker_embedding returns 400.""" - import json emb = [0.1] * 1024 files = {"audio_sample": ("test.wav", b"fake", "audio/wav")} @@ -632,7 +630,6 @@ class TestSpeechAPI: def test_upload_voice_embedding_nan_rejected(self, client): """NaN values in speaker_embedding return 400.""" - import json data = { "speaker_embedding": json.dumps([0.1] * 1023 + [float("nan")]), @@ -930,7 +927,6 @@ class TestTTSMethods: def test_upload_voice_embedding_wrong_dims_rejected(self, speech_server): """Embedding uploads must match the loaded Qwen3-TTS model before being stored.""" - import json speech_server._tts_model_type = "qwen3_tts" speech_server.engine_client.model_config = SimpleNamespace( diff --git a/tests/helpers/assertions.py b/tests/helpers/assertions.py index 65a7c1404..5898b624b 100644 --- a/tests/helpers/assertions.py +++ b/tests/helpers/assertions.py @@ -1,12 +1,13 @@ """Assertion and response validation helpers for tests.""" import io +import json import tempfile import threading import wave from io import BytesIO from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal if TYPE_CHECKING: from tests.helpers.runtime import DiffusionResponse @@ -495,62 +496,12 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve ) -def _speech_combined_error_text_for_assert(response: Any) -> str: - """Join ``error_message`` and UTF-8 body (e.g. HTTP 200 + JSON) for substring checks.""" - parts: list[str] = [] - em = getattr(response, "error_message", None) - if em: - parts.append(str(em)) - ab = getattr(response, "audio_bytes", None) - if ab: - parts.append(ab.decode("utf-8", errors="replace")) - return "\n".join(parts) - - -def _assert_speech_error_substrings(haystack: str, err_expected: Any) -> None: - if isinstance(err_expected, (list, tuple)): - assert any(str(s) in haystack for s in err_expected), ( - f"Expected error text to contain one of {err_expected!r}, got: {haystack!r}" - ) - else: - assert str(err_expected) in haystack, f"Expected error text to contain {str(err_expected)!r}, got: {haystack!r}" - - def assert_audio_speech_response(response: Any, request_config: dict[str, Any], run_level: str) -> None: - """Validate speech API results. + """Validate speech API results from :class:`~tests.helpers.runtime.OmniResponse`. - If ``status_code`` is set: assert HTTP status, then optional ``err_message`` (4xx: ``error_message`` only; - HTTP 200: ``error_message`` and/or body, e.g. JSON with HTTP 200). - - If only ``err_message`` is set: assert substring(s) on combined error text. Otherwise success/min_audio/etc. + Success path only. Use :meth:`OpenAIClientHandler.send_audio_speech_http_request` with ``err_code`` / + ``err_message`` for expected HTTP errors. """ - # When present, only compare HTTP status and skip all other checks (e.g. expected 4xx). - # ``status_code`` may be a single int or a collection of allowed codes (e.g. ``(400, 422)``). - expected_status = request_config.get("status_code") - err_expected = request_config.get("err_message") - if expected_status is not None: - actual = getattr(response, "status_code", None) - assert actual is not None, ( - "request_config has status_code but response has no .status_code " - "(set OmniResponse.status_code or pass a response object with status_code)" - ) - if isinstance(expected_status, (list, tuple, set, frozenset)): - assert actual in expected_status, f"Expected HTTP status in {expected_status!r}, got {actual}" - else: - assert actual == int(expected_status), f"Expected HTTP status {int(expected_status)}, got {actual}" - - if err_expected is not None: - if actual is not None and actual != 200: - msg = (getattr(response, "error_message", None) or "") + "" - _assert_speech_error_substrings(msg, err_expected) - else: - _assert_speech_error_substrings(_speech_combined_error_text_for_assert(response), err_expected) - return - - if err_expected is not None: - _assert_speech_error_substrings(_speech_combined_error_text_for_assert(response), err_expected) - return - assert response.success, "The request failed." # Optional floor on decoded audio size (models with very short clips may use a lower value). @@ -599,10 +550,128 @@ def assert_diffusion_response(response: "DiffusionResponse", request_config: dic assert_audio_diffusion_response(response=response, request_config=request_config, run_level=run_level) +def _http_response_body_materialize(resp: Any) -> tuple[bytes, dict[str, Any] | None]: + """Serialize ``HttpResponse``-like body to UTF-8 bytes and parse a JSON object when possible.""" + jb = getattr(resp, "json_body", None) + if jb is not None: + raw = json.dumps(jb, ensure_ascii=False).encode("utf-8") + if isinstance(jb, dict): + return raw, jb + try: + parsed = json.loads(raw.decode("utf-8", errors="replace")) + except json.JSONDecodeError: + return raw, None + return raw, parsed if isinstance(parsed, dict) else None + err = getattr(resp, "error_message", None) + raw = (err or "").encode("utf-8", errors="replace") + try: + parsed = json.loads(raw.decode("utf-8", errors="replace")) + except json.JSONDecodeError: + return raw, None + return raw, parsed if isinstance(parsed, dict) else None + + +def assert_err_message_in_text( + haystack: str, + err_message: str | tuple[str, ...] | list[str] | set[str] | frozenset[str], + *, + sequence_match: Literal["all", "any"] = "all", +) -> None: + """Assert ``err_message`` appears in ``haystack`` (case-insensitive). + + For non-string sequences: ``sequence_match='all'`` requires every substring (HTTP-style); + ``sequence_match='any'`` requires at least one (WebSocket first-frame JSON helpers). + """ + hl = haystack.lower() + if isinstance(err_message, (list, tuple, set, frozenset)): + if sequence_match == "all": + missing = [s for s in err_message if str(s).lower() not in hl] + assert not missing, ( + f"Expected error text to contain all of {err_message!r}; missing {missing!r}. haystack={haystack!r}" + ) + else: + assert any(str(s).lower() in hl for s in err_message), ( + f"Expected error text to contain one of {err_message!r}. haystack={haystack!r}" + ) + else: + assert str(err_message).lower() in hl, f"Expected error text to contain {str(err_message)!r}, got: {haystack!r}" + + +def assert_http_error( + resp: Any, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + websocket_json_message: bool = False, +) -> dict[str, Any] | None: + """Validate a raw-HTTP :class:`~tests.helpers.runtime.HttpResponse`-like object. + + Used by :class:`~tests.helpers.runtime.OpenAIClientHandler` ``send_*_http_request`` helpers when + ``request_config`` contains optional ``err_code`` and/or ``err_message``. + + When ``websocket_json_message=True``, only ``json_body`` is checked (first JSON WebSocket text frame). + Tuple/list ``err_message`` then uses **any** substring match; HTTP mode still requires **all** pieces. + + - ``err_code``: exact HTTP ``int``, or membership if a non-string sequence (e.g. ``(400, 422)``). + When ``err_code`` is set and the actual status is a client error in ``400..499`` other than ``404``, + the JSON body must include FastAPI ``detail`` and/or OpenAI-style ``error`` (2xx skips this). + - ``err_message``: substring match (case-insensitive) against serialized ``json_body`` and ``error_message``. + If ``err_message`` is a non-string sequence (``list`` / ``tuple`` / ``set`` / ``frozenset``), **every** + element must appear as a substring; a plain ``str`` still requires that single substring. + """ + if websocket_json_message: + if err_code is not None: + raise ValueError("assert_http_error: err_code is incompatible with websocket_json_message=True") + jb_ws = getattr(resp, "json_body", None) + assert jb_ws is not None, resp + if err_message is None: + return jb_ws if isinstance(jb_ws, dict) else None + assert_err_message_in_text( + json.dumps(jb_ws, ensure_ascii=False), + err_message, + sequence_match="any", + ) + return jb_ws if isinstance(jb_ws, dict) else None + + if err_code is None and err_message is None: + return None + + actual = getattr(resp, "status_code", None) + assert actual is not None, "response missing status_code" + + if err_code is not None: + if isinstance(err_code, int): + assert actual == err_code, (resp, err_code) + else: + allowed = tuple(err_code) + assert actual in allowed, (resp, allowed) + + body_bytes, payload = _http_response_body_materialize(resp) + + if err_code is not None and actual is not None and 400 <= actual < 500 and actual != 404: + assert payload is not None, getattr(resp, "error_message", resp) + assert "detail" in payload or "error" in payload, payload + + if err_message is not None: + pieces: list[str] = [] + jb = getattr(resp, "json_body", None) + if jb is not None: + pieces.append(json.dumps(jb, ensure_ascii=False)) + em = getattr(resp, "error_message", None) + if em: + pieces.append(str(em)) + haystack = "\n".join(pieces) if pieces else body_bytes.decode("utf-8", errors="replace") + assert_err_message_in_text(haystack, err_message, sequence_match="all") + + return payload + + __all__ = [ "assert_audio_diffusion_response", "assert_audio_speech_response", "assert_diffusion_response", + "assert_err_message_in_text", + "assert_http_error", "assert_image_diffusion_response", "assert_image_valid", "assert_omni_response", diff --git a/tests/helpers/runtime.py b/tests/helpers/runtime.py index 66ded055d..8b007dc16 100644 --- a/tests/helpers/runtime.py +++ b/tests/helpers/runtime.py @@ -1,5 +1,6 @@ """Server/client/runner runtime primitives for tests.""" +import asyncio import base64 import concurrent.futures import copy @@ -19,6 +20,7 @@ from dataclasses import asdict, dataclass from io import BytesIO from pathlib import Path from typing import Any, NamedTuple +from urllib.parse import quote import psutil import requests @@ -37,6 +39,7 @@ from vllm.logger import init_logger from tests.helpers.assertions import ( assert_audio_speech_response, assert_diffusion_response, + assert_http_error, assert_omni_response, ) from tests.helpers.env import run_post_test_cleanup, run_pre_test_cleanup @@ -86,6 +89,16 @@ def cleanup_dist_env_and_memory(shutdown_ray: bool = False): logger.warning("torch._C._host_emptyCache() only available in Pytorch >=2.5") +def _parse_response_json(r: requests.Response) -> dict[str, Any] | list[Any] | None: + try: + data = r.json() + if isinstance(data, (dict, list)): + return data + except Exception: + pass + return None + + def _split_request_config_by_per_output_sizes(cfg: dict[str, Any]) -> list[dict[str, Any]] | None: """If ``extra_body`` has list ``height``/``width``, return one config per index (scalar h/w, ``num_outputs_per_prompt=1``).""" eb = cfg.get("extra_body") @@ -647,6 +660,8 @@ class OmniServerStageCli(OmniServer): @dataclass class OmniResponse: + """Decoded multimodal / chat output from the OpenAI SDK or offline runner (not raw ``requests``).""" + text_content: str | None = None audio_data: list[str] | None = None audio_content: str | None = None @@ -656,16 +671,15 @@ class OmniResponse: #: OpenAI client call through response parsing and local post-process (e.g. audio decode). e2e_latency: float | None = None success: bool = False - error_message: str | None = None prompt_tokens: int | None = None cached_tokens: int | None = None - # Set for HTTP-level assertions (e.g. ``assert_audio_speech_response`` with ``status_code`` in config). - status_code: int | None = None logprobs: list | None = None @dataclass class DiffusionResponse: + """Decoded diffusion output from chat completions or offline runner (not raw ``requests``).""" + text_content: str | None = None images: list[Image.Image] | None = None audios: list[Any] | None = None @@ -674,7 +688,67 @@ class DiffusionResponse: #: ``chat.completions.create`` through local image / audio decode. e2e_latency: float | None = None success: bool = False + + +@dataclass +class HttpResponse: + """Normalized view of a ``requests`` response from :class:`OpenAIClientHandler` HTTP helpers.""" + + status_code: int + success: bool error_message: str | None = None + json_body: dict[str, Any] | list[Any] | None = None + + +@dataclass +class WebSocketJsonResponse: + """First JSON object delivered as a text WebSocket frame (streaming endpoints).""" + + json_body: dict[str, Any] + + +def _merge_http_expectation_kwargs( + base: dict[str, Any] | None, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, +) -> dict[str, Any]: + cfg = dict(base or {}) + if err_code is not None: + cfg["err_code"] = err_code + if err_message is not None: + cfg["err_message"] = err_message + return cfg + + +def _merge_ws_expectation_kwargs( + base: dict[str, Any] | None, + *, + err_message: str | tuple[str, ...] | list[str] | None = None, + ws_json_type: str | None = None, + ws_error_code: str | None = None, +) -> dict[str, Any]: + cfg = dict(base or {}) + if err_message is not None: + cfg["err_message"] = err_message + if ws_json_type is not None: + cfg["ws_json_type"] = ws_json_type + if ws_error_code is not None: + cfg["ws_error_code"] = ws_error_code + return cfg + + +def _run_ws_expectations_from_request_config(cfg: dict[str, Any], resp: WebSocketJsonResponse) -> None: + jb = resp.json_body + want_type = cfg.get("ws_json_type") + if want_type is not None: + assert jb.get("type") == want_type, (jb, want_type) + want_code = cfg.get("ws_error_code") + if want_code is not None: + assert jb.get("code") == want_code, (jb, want_code) + err_message = cfg.get("err_message") + if err_message is not None: + assert_http_error(resp, err_message=err_message, websocket_json_message=True) def _merge_diffusion_responses(parts: list[DiffusionResponse]) -> DiffusionResponse: @@ -739,8 +813,8 @@ class OpenAIClientHandler: result.e2e_latency = time.perf_counter() - wall_start result.success = True except Exception as e: - result.error_message = f"Stream processing error: {str(e)}" - print(f"Error: {result.error_message}") + msg = f"Stream processing error: {str(e)}" + print(f"Error: {msg}") return result def _process_non_stream_omni_response(self, chat_completion, *, wall_start: float) -> OmniResponse: @@ -771,8 +845,8 @@ class OpenAIClientHandler: result.logprobs = chat_completion.choices[0].logprobs.content result.success = True except Exception as e: - result.error_message = f"Non-stream processing error: {str(e)}" - print(f"Error: {result.error_message}") + msg = f"Non-stream processing error: {str(e)}" + print(f"Error: {msg}") return result def _process_diffusion_response(self, chat_completion, *, wall_start: float) -> DiffusionResponse: @@ -811,11 +885,592 @@ class OpenAIClientHandler: result.e2e_latency = time.perf_counter() - wall_start result.success = True except Exception as e: - result.error_message = f"Diffusion response processing error: {str(e)}" - print(f"Error: {result.error_message}") + msg = f"Diffusion response processing error: {str(e)}" + print(f"Error: {msg}") return result + def _http_response_from_requests(self, r: requests.Response) -> HttpResponse: + payload = _parse_response_json(r) + ok = 200 <= r.status_code < 300 + return HttpResponse( + status_code=r.status_code, + success=ok, + error_message=None if ok else (r.text[:8000] if r.text else None), + json_body=payload, + ) + + def send_health_http_request( + self, + request_config: dict[str, Any] | None = None, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/health`` (raw ``requests``). + + ``request_config``: optional ``timeout`` plus optional ``err_code`` / ``err_message`` for + :func:`~tests.helpers.assertions.assert_http_error` (also as keyword-only args). + """ + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = requests.get(self._build_url("/health"), timeout=float(cfg.get("timeout", 120.0))) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_models_http_request( + self, + request_config: dict[str, Any] | None = None, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/v1/models``. Optional ``timeout`` and HTTP assertions (see :func:`~tests.helpers.assertions.assert_http_error`).""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = requests.get( + self._build_url("/v1/models"), + headers={"Accept": "application/json"}, + timeout=float(cfg.get("timeout", 120.0)), + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_chat_completions_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/chat/completions`` with ``json`` or ``raw_body`` (malformed-body / contract tests).""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/chat/completions", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_omni_sleep_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/omni/sleep`` — ``json`` or ``raw_body``, ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/omni/sleep", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_omni_wakeup_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/omni/wakeup``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/omni/wakeup", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_voices_list_http_request( + self, + request_config: dict[str, Any] | None = None, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/v1/audio/voices``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = requests.get( + self._build_url("/v1/audio/voices"), + headers={"Accept": "application/json"}, + timeout=float(cfg.get("timeout", 120.0)), + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_voices_create_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/audio/voices`` (multipart): ``data`` / ``files`` / ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_form_endpoint("/v1/audio/voices", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_voices_delete_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """DELETE ``/v1/audio/voices/{name}`` — requires ``name``, optional ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + name = cfg["name"] + timeout = float(cfg.get("timeout", 120.0)) + path = f"/v1/audio/voices/{quote(str(name), safe='')}" + r = requests.delete( + self._build_url(path), + headers={"Accept": "application/json"}, + timeout=timeout, + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_speech_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/audio/speech`` with ``json`` or ``raw_body``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/audio/speech", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_speech_batch_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/audio/speech/batch``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/audio/speech/batch", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_audio_generate_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/audio/generate``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/audio/generate", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_images_generations_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/images/generations`` — ``json`` or ``raw_body``, ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_json_endpoint("/v1/images/generations", cfg, default_timeout=300.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_images_edits_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/images/edits`` — ``data`` / ``files`` / ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_form_endpoint("/v1/images/edits", cfg, default_timeout=300.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_videos_create_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/videos`` (async job) — multipart ``data`` / ``files``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_form_endpoint("/v1/videos", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_videos_sync_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """POST ``/v1/videos/sync``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = self._post_form_endpoint("/v1/videos/sync", cfg, default_timeout=120.0) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_videos_list_http_request( + self, + request_config: dict[str, Any] | None = None, + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/v1/videos`` — optional ``params``, ``timeout``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + r = requests.get( + self._build_url("/v1/videos"), + params=cfg.get("params"), + headers={"Accept": "application/json"}, + timeout=float(cfg.get("timeout", 120.0)), + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_video_retrieve_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/v1/videos/{video_id}``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + video_id = cfg["video_id"] + timeout = float(cfg.get("timeout", 120.0)) + r = requests.get( + self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}"), + headers={"Accept": "application/json"}, + timeout=timeout, + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_video_delete_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """DELETE ``/v1/videos/{video_id}``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + video_id = cfg["video_id"] + timeout = float(cfg.get("timeout", 120.0)) + r = requests.delete( + self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}"), + headers={"Accept": "application/json"}, + timeout=timeout, + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def send_video_content_http_request( + self, + request_config: dict[str, Any], + *, + err_code: int | tuple[int, ...] | list[int] | None = None, + err_message: str | tuple[str, ...] | list[str] | None = None, + ) -> list[HttpResponse]: + """GET ``/v1/videos/{video_id}/content``.""" + cfg = _merge_http_expectation_kwargs( + request_config, + err_code=err_code, + err_message=err_message, + ) + video_id = cfg["video_id"] + timeout = float(cfg.get("timeout", 120.0)) + r = requests.get( + self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}/content"), + timeout=timeout, + ) + resp = self._http_response_from_requests(r) + assert_http_error( + resp, + err_code=cfg.get("err_code"), + err_message=cfg.get("err_message"), + ) + return [resp] + + def _build_ws_url(self, path: str) -> str: + """Turn HTTP ``base_url`` into ``ws`` / ``wss`` for WebSocket helpers.""" + base = self.base_url.rstrip("/") + suffix = "/" + path.lstrip("/") + if base.startswith("http://"): + return "ws://" + base.removeprefix("http://") + suffix + if base.startswith("https://"): + return "wss://" + base.removeprefix("https://") + suffix + raise ValueError(f"Unsupported base_url for WebSocket: {base!r}") + + def _send_websocket_first_json_request( + self, + path: str, + cfg: dict[str, Any], + ) -> list[WebSocketJsonResponse]: + """Connect, optionally send text frames, return first JSON text frame as :class:`WebSocketJsonResponse`. + + ``request_config`` keys: + + - ``send_frames``: optional ``str`` or sequence of ``str`` raw WebSocket text frames (omit when the server + speaks first, e.g. ``/v1/realtime`` rejection path). + - ``timeout``: seconds to wait for the first inbound text frame (default ``120``). + - ``ws_max_size``: passed through as ``max_size`` to :func:`websockets.connect` when the key is present. + """ + send_frames_raw = cfg.get("send_frames") + if send_frames_raw is None: + frames: list[str] = [] + elif isinstance(send_frames_raw, str): + frames = [send_frames_raw] + else: + frames = list(send_frames_raw) + + timeout = float(cfg.get("timeout", 120.0)) + uri = self._build_ws_url(path) + + connect_kw: dict[str, Any] = {} + if "ws_max_size" in cfg: + connect_kw["max_size"] = cfg["ws_max_size"] + + async def _recv_first_json_object() -> WebSocketJsonResponse: + import websockets + + async with websockets.connect(uri, **connect_kw) as ws: + for frame in frames: + await ws.send(frame) + raw = await asyncio.wait_for(ws.recv(), timeout=timeout) + if not isinstance(raw, str): + raise AssertionError(f"Expected JSON text frame from {uri}, got {type(raw).__name__}") + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise AssertionError(f"Expected JSON text frame from {uri}, body={raw[:500]!r}") from exc + if not isinstance(data, dict): + raise AssertionError(f"Expected JSON object from {uri}, got {type(data).__name__}") + return WebSocketJsonResponse(json_body=data) + + resp = asyncio.run(_recv_first_json_object()) + _run_ws_expectations_from_request_config(cfg, resp) + return [resp] + + def send_audio_speech_stream_ws_request( + self, + request_config: dict[str, Any], + *, + err_message: str | tuple[str, ...] | list[str] | None = None, + ws_json_type: str | None = None, + ws_error_code: str | None = None, + ) -> list[WebSocketJsonResponse]: + """WebSocket ``/v1/audio/speech/stream`` — send ``send_frames`` then read first JSON text frame.""" + cfg = _merge_ws_expectation_kwargs( + request_config, + err_message=err_message, + ws_json_type=ws_json_type, + ws_error_code=ws_error_code, + ) + return self._send_websocket_first_json_request("/v1/audio/speech/stream", cfg) + + def send_video_chat_stream_ws_request( + self, + request_config: dict[str, Any], + *, + err_message: str | tuple[str, ...] | list[str] | None = None, + ws_json_type: str | None = None, + ws_error_code: str | None = None, + ) -> list[WebSocketJsonResponse]: + """WebSocket ``/v1/video/chat/stream`` — send ``send_frames`` then read first JSON text frame.""" + cfg = _merge_ws_expectation_kwargs( + request_config, + err_message=err_message, + ws_json_type=ws_json_type, + ws_error_code=ws_error_code, + ) + return self._send_websocket_first_json_request("/v1/video/chat/stream", cfg) + + def send_realtime_ws_request( + self, + request_config: dict[str, Any] | None = None, + *, + err_message: str | tuple[str, ...] | list[str] | None = None, + ws_json_type: str | None = None, + ws_error_code: str | None = None, + ) -> list[WebSocketJsonResponse]: + """WebSocket ``/v1/realtime`` — optional outbound frames, then first JSON text frame (often server-initiated).""" + cfg = _merge_ws_expectation_kwargs( + request_config, + err_message=err_message, + ws_json_type=ws_json_type, + ws_error_code=ws_error_code, + ) + return self._send_websocket_first_json_request("/v1/realtime", cfg) + def send_omni_request(self, request_config: dict[str, Any], request_num: int = 1) -> list[OmniResponse]: + """Chat completions via the OpenAI Python SDK (not raw HTTP).""" responses: list[OmniResponse] = [] stream = request_config.get("stream", False) modalities = request_config.get("modalities", ["text", "audio"]) @@ -939,8 +1594,8 @@ class OpenAIClientHandler: result.audio_format = result.audio_format.headers.get("content-type", "") except Exception as e: - result.error_message = f"Audio speech stream processing error: {str(e)}" - print(f"Error: {result.error_message}") + msg = f"Audio speech stream processing error: {str(e)}" + print(f"Error: {msg}") return result @@ -978,8 +1633,8 @@ class OpenAIClientHandler: result.audio_format = result.audio_format.headers.get("content-type", "") except Exception as e: - result.error_message = f"Audio speech non-stream processing error: {str(e)}" - print(f"Error: {result.error_message}") + msg = f"Audio speech non-stream processing error: {str(e)}" + print(f"Error: {msg}") return result @@ -1201,6 +1856,7 @@ class OpenAIClientHandler: Send OpenAI requests for diffusion models. If ``extra_body`` has list ``height``/``width``, sends one chat completion per index in parallel (scalar h/w, ``num_outputs_per_prompt=1`` each) and merges images in list order. + Args: request_config: A single request configuration dict, or a list of request configuration dicts (one request per element) @@ -1300,10 +1956,13 @@ class OpenAIClientHandler: self, request_config: dict[str, Any], request_num: int = 1 ) -> list[DiffusionResponse]: """ - Send native /v1/videos requests. + Send native /v1/videos requests: multipart ``form_data`` job create, poll until done, download content. + + For raw HTTP to video routes without polling, use ``send_videos_create_http_request``, etc. """ if request_num != 1: raise NotImplementedError("Concurrent video diffusion requests are not currently implemented") + form_data = request_config.get("form_data") if not isinstance(form_data, dict): raise ValueError("Video request_config must contain 'form_data'") @@ -1346,6 +2005,54 @@ class OpenAIClientHandler: self._print_client_stat("[diffusion] request#1 completed") return [result] + def _post_json_endpoint( + self, + path: str, + request_config: dict[str, Any], + *, + default_timeout: float, + ) -> requests.Response: + url = self._build_url(path) + timeout = float(request_config.get("timeout", default_timeout)) + if "raw_body" in request_config: + raw = request_config["raw_body"] + payload = raw.encode("utf-8") if isinstance(raw, str) else raw + return requests.post( + url, + data=payload, + headers={"Content-Type": "application/json", "Accept": "application/json"}, + timeout=timeout, + ) + if "json" not in request_config: + raise ValueError(f"{path} request_config must include 'json' or 'raw_body'") + return requests.post( + url, + json=request_config["json"], + headers={"Accept": "application/json", "Content-Type": "application/json"}, + timeout=timeout, + ) + + def _post_form_endpoint( + self, + path: str, + request_config: dict[str, Any], + *, + default_timeout: float = 120.0, + ) -> requests.Response: + url = self._build_url(path) + timeout = float(request_config.get("timeout", default_timeout)) + data = request_config.get("data") + files = request_config.get("files") + if data is None and not files: + data = {} + return requests.post( + url, + data=data, + files=files, + headers={"Accept": "application/json"} if not files else {"Accept": "application/json"}, + timeout=timeout, + ) + def _wait_until_video_completed( self, video_id: str, poll_interval_seconds: int = 2, timeout_seconds: int = 300 ) -> None: @@ -1693,9 +2400,9 @@ class OmniRunnerHandler: result.text_content = text_content result.success = True except Exception as e: - result.error_message = f"Output processing error: {str(e)}" + msg = f"Output processing error: {str(e)}" result.success = False - print(f"Error: {result.error_message}") + print(f"Error: {msg}") return result def _process_diffusion_output(self, outputs: list[OmniRequestOutput]) -> DiffusionResponse: @@ -1831,15 +2538,11 @@ class OmniRunnerHandler: mm_out = stage_out.request_output.outputs[0].multimodal_output break if mm_out is None: - result = OmniResponse(success=False, error_message="No audio output from pipeline") - assert result.success, result.error_message - return result + raise AssertionError("No audio output from pipeline") audio_data = mm_out.get("audio") if audio_data is None: - result = OmniResponse(success=False, error_message="No audio tensor in multimodal output") - assert result.success, result.error_message - return result + raise AssertionError("No audio tensor in multimodal output") sr_raw = mm_out.get("sr") sr_val = sr_raw[-1] if isinstance(sr_raw, list) and sr_raw else sr_raw @@ -2004,6 +2707,8 @@ def iter_omni_runner( __all__ = [ "DiffusionResponse", + "HttpResponse", + "WebSocketJsonResponse", "OmniResponse", "OmniRunner", "OmniRunnerHandler",