diff --git a/tests/conftest.py b/tests/conftest.py index a8edf2d0e..09696658d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -52,6 +52,10 @@ def default_vllm_config(): @pytest.fixture(autouse=True) def clean_gpu_memory_between_tests(): + clean_gpu_memory() + + +def clean_gpu_memory(): if os.getenv("VLLM_TEST_CLEAN_GPU_MEMORY", "0") != "1": yield return @@ -528,7 +532,6 @@ def convert_audio_to_text(audio_data): temperature=0.0, word_timestamps=True, condition_on_previous_text=False, - initial_prompt="Please transcribe with proper word spacing.", )["text"] if text: return text @@ -709,31 +712,5 @@ class OmniServer: def __exit__(self, exc_type, exc_val, exc_tb): if self.proc: - try: - parent = psutil.Process(self.proc.pid) - children = parent.children(recursive=True) - for child in children: - try: - child.terminate() - except psutil.NoSuchProcess: - pass - - gone, still_alive = psutil.wait_procs(children, timeout=10) - - for child in still_alive: - try: - child.kill() - except psutil.NoSuchProcess: - pass - - try: - parent.terminate() - parent.wait(timeout=10) - except (psutil.NoSuchProcess, psutil.TimeoutExpired): - try: - parent.kill() - except psutil.NoSuchProcess: - pass - - except psutil.NoSuchProcess: - pass + self._kill_process_tree(self.proc.pid) + clean_gpu_memory() diff --git a/tests/e2e/offline_inference/conftest.py b/tests/e2e/offline_inference/conftest.py index 8bae08760..276fd8844 100644 --- a/tests/e2e/offline_inference/conftest.py +++ b/tests/e2e/offline_inference/conftest.py @@ -10,6 +10,7 @@ import pytest from vllm.distributed.parallel_state import cleanup_dist_env_and_memory from vllm.sampling_params import SamplingParams +from tests.conftest import clean_gpu_memory from vllm_omni.entrypoints.omni import Omni from vllm_omni.outputs import OmniRequestOutput @@ -335,6 +336,7 @@ class OmniRunner: self.close() del self.omni cleanup_dist_env_and_memory() + clean_gpu_memory() def close(self): """Close and cleanup the Omni instance.""" diff --git a/tests/e2e/offline_inference/test_cache_dit.py b/tests/e2e/offline_inference/test_cache_dit.py index 546ff7436..9eaee00c3 100644 --- a/tests/e2e/offline_inference/test_cache_dit.py +++ b/tests/e2e/offline_inference/test_cache_dit.py @@ -40,42 +40,49 @@ def test_cache_dit(model_name: str): "residual_diff_threshold": 0.24, "max_continuous_cached_steps": 3, } + m = None + try: + m = Omni( + model=model_name, + cache_backend="cache_dit", + cache_config=cache_config, + ) - m = Omni( - model=model_name, - cache_backend="cache_dit", - cache_config=cache_config, - ) + # Use minimal settings for fast testing + height = 256 + width = 256 + num_inference_steps = 4 # Minimal steps for fast test - # Use minimal settings for fast testing - height = 256 - width = 256 - num_inference_steps = 4 # Minimal steps for fast test + outputs = m.generate( + "a photo of a cat sitting on a laptop keyboard", + height=height, + width=width, + num_inference_steps=num_inference_steps, + guidance_scale=0.0, + generator=torch.Generator("cuda").manual_seed(42), + num_outputs_per_prompt=1, # Single output for speed + ) + # Extract images from request_output[0]['images'] + first_output = outputs[0] + assert first_output.final_output_type == "image" + if not hasattr(first_output, "request_output") or not first_output.request_output: + raise ValueError("No request_output found in OmniRequestOutput") - outputs = m.generate( - "a photo of a cat sitting on a laptop keyboard", - height=height, - width=width, - num_inference_steps=num_inference_steps, - guidance_scale=0.0, - generator=torch.Generator("cuda").manual_seed(42), - num_outputs_per_prompt=1, # Single output for speed - ) - # Extract images from request_output[0]['images'] - first_output = outputs[0] - assert first_output.final_output_type == "image" - if not hasattr(first_output, "request_output") or not first_output.request_output: - raise ValueError("No request_output found in OmniRequestOutput") + req_out = first_output.request_output[0] + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): + raise ValueError("Invalid request_output structure or missing 'images' key") - req_out = first_output.request_output[0] - if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): - raise ValueError("Invalid request_output structure or missing 'images' key") + images = req_out.images - images = req_out.images - - # Verify generation succeeded - assert images is not None - assert len(images) == 1 - # Check image size - assert images[0].width == width - assert images[0].height == height + # Verify generation succeeded + assert images is not None + assert len(images) == 1 + # Check image size + assert images[0].width == width + assert images[0].height == height + except Exception as e: + print(f"Test failed with error: {e}") + raise + finally: + if m is not None and hasattr(m, "close"): + m.close() diff --git a/tests/e2e/offline_inference/test_t2i_model.py b/tests/e2e/offline_inference/test_t2i_model.py index f491fa72e..09fadc93a 100644 --- a/tests/e2e/offline_inference/test_t2i_model.py +++ b/tests/e2e/offline_inference/test_t2i_model.py @@ -34,33 +34,41 @@ elif is_rocm(): @pytest.mark.parametrize("model_name", models) def test_diffusion_model(model_name: str): - m = Omni(model=model_name) - # high resolution may cause OOM on L4 - height = 256 - width = 256 - outputs = m.generate( - "a photo of a cat sitting on a laptop keyboard", - height=height, - width=width, - num_inference_steps=2, - guidance_scale=0.0, - generator=torch.Generator("cuda").manual_seed(42), - num_outputs_per_prompt=2, - ) - # Extract images from request_output[0]['images'] - first_output = outputs[0] - assert first_output.final_output_type == "image" - if not hasattr(first_output, "request_output") or not first_output.request_output: - raise ValueError("No request_output found in OmniRequestOutput") + m = None + try: + m = Omni(model=model_name) + # high resolution may cause OOM on L4 + height = 256 + width = 256 + outputs = m.generate( + "a photo of a cat sitting on a laptop keyboard", + height=height, + width=width, + num_inference_steps=2, + guidance_scale=0.0, + generator=torch.Generator("cuda").manual_seed(42), + num_outputs_per_prompt=2, + ) + # Extract images from request_output[0]['images'] + first_output = outputs[0] + assert first_output.final_output_type == "image" + if not hasattr(first_output, "request_output") or not first_output.request_output: + raise ValueError("No request_output found in OmniRequestOutput") - req_out = first_output.request_output[0] - if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): - raise ValueError("Invalid request_output structure or missing 'images' key") + req_out = first_output.request_output[0] + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): + raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out.images + images = req_out.images - assert len(images) == 2 - # check image size - assert images[0].width == width - assert images[0].height == height - images[0].save("image_output.png") + assert len(images) == 2 + # check image size + assert images[0].width == width + assert images[0].height == height + images[0].save("image_output.png") + except Exception as e: + print(f"Test failed with error: {e}") + raise + finally: + if m is not None and hasattr(m, "close"): + m.close() diff --git a/tests/e2e/offline_inference/test_teacache.py b/tests/e2e/offline_inference/test_teacache.py index dc8004f2f..7ce915b96 100644 --- a/tests/e2e/offline_inference/test_teacache.py +++ b/tests/e2e/offline_inference/test_teacache.py @@ -36,42 +36,49 @@ def test_teacache(model_name: str): cache_config = { "rel_l1_thresh": 0.2, # Default threshold } + m = None + try: + m = Omni( + model=model_name, + cache_backend="tea_cache", + cache_config=cache_config, + ) - m = Omni( - model=model_name, - cache_backend="tea_cache", - cache_config=cache_config, - ) + # Use minimal settings for fast testing + height = 256 + width = 256 + num_inference_steps = 4 # Minimal steps for fast test - # Use minimal settings for fast testing - height = 256 - width = 256 - num_inference_steps = 4 # Minimal steps for fast test + outputs = m.generate( + "a photo of a cat sitting on a laptop keyboard", + height=height, + width=width, + num_inference_steps=num_inference_steps, + guidance_scale=0.0, + generator=torch.Generator("cuda").manual_seed(42), + num_outputs_per_prompt=1, # Single output for speed + ) + # Extract images from request_output[0]['images'] + first_output = outputs[0] + assert first_output.final_output_type == "image" + if not hasattr(first_output, "request_output") or not first_output.request_output: + raise ValueError("No request_output found in OmniRequestOutput") - outputs = m.generate( - "a photo of a cat sitting on a laptop keyboard", - height=height, - width=width, - num_inference_steps=num_inference_steps, - guidance_scale=0.0, - generator=torch.Generator("cuda").manual_seed(42), - num_outputs_per_prompt=1, # Single output for speed - ) - # Extract images from request_output[0]['images'] - first_output = outputs[0] - assert first_output.final_output_type == "image" - if not hasattr(first_output, "request_output") or not first_output.request_output: - raise ValueError("No request_output found in OmniRequestOutput") + req_out = first_output.request_output[0] + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): + raise ValueError("Invalid request_output structure or missing 'images' key") - req_out = first_output.request_output[0] - if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): - raise ValueError("Invalid request_output structure or missing 'images' key") + images = req_out.images - images = req_out.images - - # Verify generation succeeded - assert images is not None - assert len(images) == 1 - # Check image size - assert images[0].width == width - assert images[0].height == height + # Verify generation succeeded + assert images is not None + assert len(images) == 1 + # Check image size + assert images[0].width == width + assert images[0].height == height + except Exception as e: + print(f"Test failed with error: {e}") + raise + finally: + if m is not None and hasattr(m, "close"): + m.close() diff --git a/tests/e2e/online_serving/test_qwen3_omni.py b/tests/e2e/online_serving/test_qwen3_omni.py index 33a05c1ce..c458a1686 100644 --- a/tests/e2e/online_serving/test_qwen3_omni.py +++ b/tests/e2e/online_serving/test_qwen3_omni.py @@ -8,21 +8,22 @@ import os os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" -import base64 import concurrent.futures -import ctypes -import signal -import socket -import subprocess -import sys import time from pathlib import Path import openai import pytest -from vllm.assets.video import VideoAsset -from vllm.utils.network_utils import get_open_port +from tests.conftest import ( + OmniServer, + convert_audio_to_text, + cosine_similarity_text, + dummy_messages_from_mix_data, + generate_synthetic_audio, + generate_synthetic_image, + generate_synthetic_video, +) from vllm_omni.utils import is_rocm models = ["Qwen/Qwen3-Omni-30B-A3B-Instruct"] @@ -32,114 +33,27 @@ if is_rocm(): # ROCm stage config optimized for MI325 GPU stage_configs = [str(Path(__file__).parent / "stage_configs" / "rocm" / "qwen3_omni_ci.yaml")] else: - stage_configs = [str(Path(__file__).parent / "stage_configs" / "qwen3_omni_ci.yaml")] + stage_configs = [str(Path(__file__).parent.parent / "stage_configs" / "qwen3_omni_ci.yaml")] # Create parameter combinations for model and stage config test_params = [(model, stage_config) for model in models for stage_config in stage_configs] -class OmniServer: - """Omniserver for vLLM-Omni tests.""" - - def __init__( - self, - model: str, - serve_args: list[str], - *, - env_dict: dict[str, str] | None = None, - ) -> None: - self.model = model - self.serve_args = serve_args - self.env_dict = env_dict - self.proc: subprocess.Popen | None = None - self.host = "127.0.0.1" - self.port = get_open_port() - - def _start_server(self) -> None: - """Start the vLLM-Omni server subprocess.""" - env = os.environ.copy() - env["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" - if self.env_dict is not None: - env.update(self.env_dict) - - cmd = [ - sys.executable, - "-m", - "vllm_omni.entrypoints.cli.main", - "serve", - self.model, - "--omni", - "--host", - self.host, - "--port", - str(self.port), - ] + self.serve_args - - # Helper to ensure child process dies when parent dies - libc = ctypes.CDLL("libc.so.6") - - def preexec_fn(): - # Ensure the child process receives SIGTERM when the parent (this test runner) dies. - # This prevents orphaned processes if the test is killed unexpectedly. - PR_SET_PDEATHSIG = 1 - libc.prctl(PR_SET_PDEATHSIG, signal.SIGTERM) - - print(f"Launching OmniServer with: {' '.join(cmd)}") - self.proc = subprocess.Popen( - cmd, - env=env, - cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), # Set working directory to vllm-omni root - start_new_session=True, - preexec_fn=preexec_fn, - ) - - # Wait for server to be ready - max_wait = 600 # 10 minutes - start_time = time.time() - while time.time() - start_time < max_wait: - try: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(1) - result = sock.connect_ex((self.host, self.port)) - if result == 0: - print(f"Server ready on {self.host}:{self.port}") - return - except Exception: - pass - time.sleep(2) - - raise RuntimeError(f"Server failed to start within {max_wait} seconds") - - def __enter__(self): - self._start_server() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - if self.proc: - try: - os.killpg(self.proc.pid, signal.SIGTERM) - except ProcessLookupError: - pass - - try: - self.proc.wait(timeout=30) - except subprocess.TimeoutExpired: - try: - os.killpg(self.proc.pid, signal.SIGKILL) - except ProcessLookupError: - pass - self.proc.wait() - - -@pytest.fixture +@pytest.fixture(scope="module") def omni_server(request): """Start vLLM-Omni server as a subprocess with actual model weights. - Uses module scope so the server starts only once for all tests. + Uses session scope so the server starts only once for the entire test session. Multi-stage initialization can take 10-20+ minutes. """ model, stage_config_path = request.param - with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server: + + print(f"Starting OmniServer with model: {model}") + print("This may take 10-20+ minutes for initialization...") + + with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"]) as server: + print("OmniServer started successfully") yield server + print("OmniServer stopped") @pytest.fixture @@ -151,15 +65,6 @@ def client(omni_server): ) -@pytest.fixture(scope="session") -def base64_encoded_video() -> str: - """Base64 encoded video for testing.""" - video = VideoAsset(name="baby_reading", num_frames=4) - with open(video.video_path, "rb") as f: - content = f.read() - return base64.b64encode(content).decode("utf-8") - - def get_system_prompt(): return { "role": "system", @@ -193,71 +98,50 @@ def dummy_messages_from_video_data( ] +def get_prompt(prompt_type="text_only"): + prompts = { + "text_only": "What is the capital of China?", + "mix": "What is recited in the audio? What is in this image? Describe the video briefly.", + } + return prompts.get(prompt_type, prompts["text_only"]) + + +def get_max_batch_size(size_type="few"): + batch_sizes = {"few": 5, "medium": 100, "large": 256} + return batch_sizes.get(size_type, 5) + + +@pytest.mark.skipif(is_rocm(), reason="Test skipped on AMD environment due to known output issues") @pytest.mark.parametrize("omni_server", test_params, indirect=True) -def test_video_to_audio_concurrent( - client: openai.OpenAI, - omni_server, - base64_encoded_video: str, -) -> None: - """Test processing video with multiple concurrent completions, generating audio output via OpenAI API.""" - # Create data URL for the base64 encoded video - video_data_url = f"data:video/mp4;base64,{base64_encoded_video}" +def test_mix_to_text_audio_001(client: openai.OpenAI, omni_server) -> None: + """ + Test multi-modal input processing and text/audio output generation via OpenAI API. + Deploy Setting: default yaml + Input Modal: text + audio + video + image + Output Modal: text + audio + Input Setting: stream=True + Datasets: single request + """ - messages = dummy_messages_from_video_data(video_data_url) - - # Test multiple concurrent completions - num_concurrent_requests = 5 - - with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor: - # Submit multiple completion requests concurrently - futures = [ - executor.submit( - client.chat.completions.create, - model=omni_server.model, - messages=messages, - ) - for _ in range(num_concurrent_requests) - ] - - # Wait for all requests to complete and collect results - chat_completions = [future.result() for future in concurrent.futures.as_completed(futures)] - - # Verify all completions succeeded - assert len(chat_completions) == num_concurrent_requests - - for chat_completion in chat_completions: - assert len(chat_completion.choices) == 2 # 1 for text output, 1 for audio output - - # Verify text output - text_choice = chat_completion.choices[0] - assert text_choice.finish_reason == "length" - - # Verify we got a response - text_message = text_choice.message - assert text_message.content is not None and len(text_message.content) >= 10 - assert text_message.role == "assistant" - - # Verify audio output - audio_choice = chat_completion.choices[1] - assert audio_choice.finish_reason == "stop" - audio_message = audio_choice.message - - # Check if audio was generated - if hasattr(audio_message, "audio") and audio_message.audio: - assert audio_message.audio.data is not None - assert len(audio_message.audio.data) > 0 - - # Test streaming completion - chat_completion = client.chat.completions.create( - model=omni_server.model, - messages=messages, - stream=True, + # Test single completion + e2e_list = list() + video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(224, 224, 300)['base64']}" + image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(224, 224)['base64']}" + audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(5, 1)['base64']}" + messages = dummy_messages_from_mix_data( + system_prompt=get_system_prompt(), + video_data_url=video_data_url, + image_data_url=image_data_url, + audio_data_url=audio_data_url, + content_text=get_prompt("mix"), ) - # Collect text and audio data from stream + # Test single completion + start_time = time.perf_counter() + chat_completion = client.chat.completions.create(model=omni_server.model, messages=messages, stream=True) + text_content = "" audio_data = None - for chunk in chat_completion: for choice in chunk.choices: if hasattr(choice, "delta"): @@ -268,18 +152,92 @@ def test_video_to_audio_concurrent( modality = getattr(chunk, "modality", None) if modality == "audio" and content: - # Audio chunk - decode base64 content + # Audio chunk - content if audio_data is None: - audio_data = base64.b64decode(content) + audio_data = content else: - audio_data += base64.b64decode(content) + audio_data += content elif modality == "text" and content: # Text chunk - accumulate text content text_content += content if content else "" - # Verify text output - assert text_content is not None and len(text_content) >= 2 + # Verify E2E + current_e2e = time.perf_counter() - start_time + print(f"the request e2e is: {current_e2e}") + # TODO: Verify the E2E latency after confirmation baseline. + e2e_list.append(current_e2e) - # Verify audio output - assert audio_data is not None - assert len(audio_data) > 0 + print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}") + # Verify all completions succeeded + assert audio_data is not None, "No audio output is generated" + + # Verify text output success + assert text_content is not None and len(text_content) >= 2, "No text output is generated" + assert any( + keyword in text_content.lower() for keyword in ["square", "quadrate", "sphere", "globe", "circle", "round"] + ), "The output does not contain any of the keywords." + + # Verify text output same as audio output + audio_content = convert_audio_to_text(audio_data) + print(f"text content is: {text_content}") + print(f"audio content is: {audio_content}") + similarity = cosine_similarity_text(audio_content.lower(), text_content.lower()) + print(f"similarity is: {similarity}") + assert similarity > 0.9, "The audio content is not same as the text" + + +@pytest.mark.skipif(is_rocm(), reason="Test skipped on AMD environment due to known output issues") +@pytest.mark.parametrize("omni_server", test_params, indirect=True) +def test_text_to_text_audio_001(client: openai.OpenAI, omni_server) -> None: + """ + Test text input processing and text/audio output generation via OpenAI API. + Deploy Setting: default yaml + Input Modal: text + Output Modal: text + audio + Datasets: few requests + """ + + num_concurrent_requests = get_max_batch_size() + messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt()) + + e2e_list = list() + with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor: + # Submit multiple completion requests concurrently + futures = [ + executor.submit(client.chat.completions.create, model=omni_server.model, messages=messages) + for _ in range(num_concurrent_requests) + ] + start_time = time.perf_counter() + # Wait for all requests to complete and collect results + chat_completions = list() + for future in concurrent.futures.as_completed(futures): + chat_completions.append(future.result()) + # Verify E2E + current_e2e = time.perf_counter() - start_time + print(f"the request e2e is: {current_e2e}") + # TODO: Verify the E2E latency after confirmation baseline. + e2e_list.append(current_e2e) + + print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}") + # Verify all completions succeeded + assert len(chat_completions) == num_concurrent_requests, "Not all requests succeeded." + for chat_completion in chat_completions: + # Verify audio output success + audio_message = chat_completion.choices[1].message + audio_data = audio_message.audio.data + assert audio_data is not None, "No audio output is generated" + assert audio_message.audio.expires_at > time.time(), "The generated audio has expired." + + # Verify text output success + text_choice = chat_completion.choices[0] + text_content = text_choice.message.content + assert text_choice.message.content is not None, "No text output is generated" + assert "beijing" in text_choice.message.content.lower(), "The output do not contain keywords." + + # Verify text output same as audio output + audio_content = convert_audio_to_text(audio_data) + print(f"text content is: {text_content}") + print(f"audio content is: {audio_content}") + similarity = cosine_similarity_text(audio_content.lower(), text_content.lower()) + print(f"similarity is: {similarity}") + assert similarity > 0.9, "The audio content is not same as the text" diff --git a/tests/e2e/online_serving/test_qwen3_omni_expansion.py b/tests/e2e/online_serving/test_qwen3_omni_expansion.py index a779993bf..563eaf628 100644 --- a/tests/e2e/online_serving/test_qwen3_omni_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_omni_expansion.py @@ -111,69 +111,6 @@ def test_text_to_text_001(test_config: tuple[str, str]) -> None: assert "beijing" in text_choice.message.content.lower(), "The output do not contain keywords." -@pytest.mark.parametrize("test_config", test_params) -def test_text_to_text_audio_001(test_config: tuple[str, str]) -> None: - """Test processing text, generating text and audio output via OpenAI API.""" - - model, stage_config_path = test_config - num_concurrent_requests = get_max_batch_size() - stage_config_path = modify_stage_config( - stage_config_path, - { - 0: {"runtime.max_batch_size": num_concurrent_requests}, - 1: {"runtime.max_batch_size": num_concurrent_requests}, - }, - ) - with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "90"]) as server: - messages = dummy_messages_from_mix_data( - system_prompt=get_system_prompt(), content_text="What is the capital of China?" - ) - - # Test single completion - api_client = client(server) - e2e_list = list() - with concurrent.futures.ThreadPoolExecutor(max_workers=num_concurrent_requests) as executor: - # Submit multiple completion requests concurrently - futures = [ - executor.submit(api_client.chat.completions.create, model=server.model, messages=messages) - for _ in range(num_concurrent_requests) - ] - start_time = time.perf_counter() - # Wait for all requests to complete and collect results - chat_completions = list() - for future in concurrent.futures.as_completed(futures): - chat_completions.append(future.result()) - # Verify E2E - current_e2e = time.perf_counter() - start_time - print(f"the request e2e is: {current_e2e}") - # TODO: Verify the E2E latency after confirmation baseline. - e2e_list.append(current_e2e) - - print(f"the avg e2e is: {sum(e2e_list) / len(e2e_list)}") - # Verify all completions succeeded - assert len(chat_completions) == num_concurrent_requests, "Not all requests succeeded." - for chat_completion in chat_completions: - # Verify audio output success - audio_message = chat_completion.choices[1].message - audio_data = audio_message.audio.data - assert audio_data is not None, "No audio output is generated" - assert audio_message.audio.expires_at > time.time(), "The generated audio has expired." - - # Verify text output success - text_choice = chat_completion.choices[0] - text_content = text_choice.message.content - assert text_choice.message.content is not None, "No text output is generated" - assert "beijing" in text_choice.message.content.lower(), "The output do not contain keywords." - - # Verify text output same as audio output - audio_content = convert_audio_to_text(audio_data) - print(f"text content is: {text_content}") - print(f"audio content is: {audio_content}") - assert cosine_similarity_text(audio_content.lower(), text_content.lower()) > 0.9, ( - "The audio content is not same as the text" - ) - - @pytest.mark.parametrize("test_config", test_params) def test_audio_to_text_001(test_config: tuple[str, str]) -> None: """Test processing text, generating text output via OpenAI API.""" diff --git a/tests/e2e/stage_configs/qwen3_omni_ci.yaml b/tests/e2e/stage_configs/qwen3_omni_ci.yaml index 24eeeb8d6..4b5638529 100644 --- a/tests/e2e/stage_configs/qwen3_omni_ci.yaml +++ b/tests/e2e/stage_configs/qwen3_omni_ci.yaml @@ -8,7 +8,7 @@ stage_args: - stage_id: 0 runtime: devices: "0,1" - max_batch_size: 1 + max_batch_size: 5 engine_args: model_stage: thinker model_arch: Qwen3OmniMoeForConditionalGeneration @@ -39,7 +39,7 @@ stage_args: - stage_id: 1 runtime: devices: "1" - max_batch_size: 1 + max_batch_size: 5 engine_args: model_stage: talker model_arch: Qwen3OmniMoeForConditionalGeneration