vllm-project--vllm-omni
wehub-resource-sync
124 行
4.4 KiB
Python
124 行
4.4 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
|
|
|
"""Unit tests for the audio-streaming continuity helper.
|
|
|
|
The helper is intentionally vllm-free: it operates on a flat timeline of
|
|
``(arrival_time_s, num_bytes)`` chunks plus PCM format params. These tests
|
|
exercise the underrun arithmetic against hand-crafted timelines so future
|
|
refactors of the streaming backend cannot silently regress the metric.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from vllm_omni.benchmarks.audio_continuity import (
|
|
ContinuityStats,
|
|
compute_continuity_stats,
|
|
)
|
|
|
|
pytestmark = [pytest.mark.core_model, pytest.mark.benchmark, pytest.mark.cpu]
|
|
|
|
|
|
# PCM s16le mono at 24 kHz -> 48 000 bytes/sec, the Qwen3-TTS / VoxCPM2 default.
|
|
_SR = 24_000
|
|
_BPS = _SR * 2 # 48 000
|
|
|
|
|
|
def test_empty_timeline_is_continuous() -> None:
|
|
stats = compute_continuity_stats([], [], sample_rate=_SR)
|
|
assert stats == ContinuityStats(
|
|
max_underrun_s=0.0,
|
|
underrun_event_count=0,
|
|
is_continuous=True,
|
|
)
|
|
|
|
|
|
def test_single_chunk_has_no_underrun() -> None:
|
|
# One large chunk, no later arrivals to check against.
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.05],
|
|
chunk_bytes=[_BPS], # one second of audio in a single chunk
|
|
sample_rate=_SR,
|
|
)
|
|
assert stats.max_underrun_s == 0.0
|
|
assert stats.underrun_event_count == 0
|
|
assert stats.is_continuous is True
|
|
|
|
|
|
def test_chunks_faster_than_realtime_have_no_underrun() -> None:
|
|
# 100ms of audio per chunk, but arrive every 50ms — buffer always builds.
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.0, 0.05, 0.10, 0.15],
|
|
chunk_bytes=[_BPS // 10] * 4,
|
|
sample_rate=_SR,
|
|
)
|
|
assert stats.max_underrun_s == 0.0
|
|
assert stats.underrun_event_count == 0
|
|
assert stats.is_continuous is True
|
|
|
|
|
|
def test_single_gap_above_threshold_breaks_continuity() -> None:
|
|
# First chunk carries 100ms of audio at t=0. Next chunk arrives at t=0.5s,
|
|
# so the player ran dry from t=0.1 to t=0.5 -> 400ms underrun.
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.0, 0.5],
|
|
chunk_bytes=[_BPS // 10, _BPS // 10], # 100ms each
|
|
sample_rate=_SR,
|
|
threshold_s=0.1,
|
|
)
|
|
assert stats.max_underrun_s == pytest.approx(0.4, abs=1e-6)
|
|
assert stats.underrun_event_count == 1
|
|
assert stats.is_continuous is False
|
|
|
|
|
|
def test_underrun_is_max_across_gaps_not_sum() -> None:
|
|
# Two gaps: a 200ms shortfall and a 50ms shortfall. max_underrun_s
|
|
# must surface the 200ms one, not the sum.
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.0, 0.3, 0.45],
|
|
chunk_bytes=[_BPS // 10, _BPS // 10, _BPS // 10], # 100ms chunks
|
|
sample_rate=_SR,
|
|
threshold_s=0.1,
|
|
)
|
|
# At t=0.3 the player consumed 300ms but only 100ms received -> 200ms deficit
|
|
# At t=0.45 player consumed 450ms, received 200ms -> 250ms deficit.
|
|
assert stats.max_underrun_s == pytest.approx(0.25, abs=1e-6)
|
|
assert stats.underrun_event_count == 2
|
|
assert stats.is_continuous is False
|
|
|
|
|
|
def test_continuity_threshold_boundary_inclusive() -> None:
|
|
# First chunk = 100ms audio at t=0. Next arrives at t=0.2 -> 100ms deficit
|
|
# exactly. With threshold=0.1, continuity_ok stays True (boundary).
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.0, 0.2],
|
|
chunk_bytes=[_BPS // 10, _BPS // 10],
|
|
sample_rate=_SR,
|
|
threshold_s=0.1,
|
|
)
|
|
assert stats.max_underrun_s == pytest.approx(0.1, abs=1e-6)
|
|
assert stats.is_continuous is True
|
|
|
|
|
|
def test_mismatched_lengths_returns_zero_stats() -> None:
|
|
# Defensive: never crash on malformed timeline.
|
|
stats = compute_continuity_stats(
|
|
chunk_arrival_times_s=[0.0, 0.1],
|
|
chunk_bytes=[100],
|
|
sample_rate=_SR,
|
|
)
|
|
assert stats == ContinuityStats(0.0, 0, True)
|
|
|
|
|
|
def test_threshold_is_configurable() -> None:
|
|
# 200ms underrun: violates 100ms threshold, OK against 500ms threshold.
|
|
arrivals = [0.0, 0.3]
|
|
bytes_list = [_BPS // 10, _BPS // 10]
|
|
strict = compute_continuity_stats(arrivals, bytes_list, sample_rate=_SR, threshold_s=0.1)
|
|
lenient = compute_continuity_stats(arrivals, bytes_list, sample_rate=_SR, threshold_s=0.5)
|
|
assert strict.is_continuous is False
|
|
assert lenient.is_continuous is True
|
|
assert strict.max_underrun_s == lenient.max_underrun_s
|