From bde3834f48d38f5c84eb0a0b3848ab0daf8922b6 Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Mon, 19 Oct 2026 09:56:27 +0800 Subject: [PATCH 1/6] [Bugfix] Preserve audio-in-video fixture requirements Propagate TTS and FFmpeg failures instead of caching video-only output for embed_audio=True. Regenerate legacy cached MP4s that lack the requested audio stream, and cover failure propagation, real AAC decoding, and cache reuse with CPU regressions. Refs #8892 Signed-off-by: psv666 <2693925048@qq.com> --- tests/helpers/media.py | 110 ++++++++++---------- tests/helpers/tests/test_synthetic_video.py | 89 ++++++++++++++++ 2 files changed, 143 insertions(+), 56 deletions(-) create mode 100644 tests/helpers/tests/test_synthetic_video.py diff --git a/tests/helpers/media.py b/tests/helpers/media.py index 18da70fa4a6..8029af608ad 100644 --- a/tests/helpers/media.py +++ b/tests/helpers/media.py @@ -308,16 +308,12 @@ def _mux_mp4_bytes_with_synthetic_audio( duration_sec = num_frames / fps if fps > 0 else 0.0 duration_int = max(1, int(math.ceil(duration_sec))) - try: - audio_result = generate_synthetic_audio( - duration=duration_int, - num_channels=1, - sample_rate=sample_rate, - ) - audio_pcm = audio_result["np_array"] - except Exception as e: - logger.warning("Synthetic video: generate_synthetic_audio failed (%s); using video-only MP4.", e) - return video_mp4_bytes + audio_result = generate_synthetic_audio( + duration=duration_int, + num_channels=1, + sample_rate=sample_rate, + ) + audio_pcm = audio_result["np_array"] try: import imageio_ffmpeg @@ -326,47 +322,38 @@ def _mux_mp4_bytes_with_synthetic_audio( except Exception: ffmpeg_exe = "ffmpeg" - try: - with tempfile.TemporaryDirectory(prefix="syn_vid_mux_") as tmp: - vid_path = os.path.join(tmp, "video.mp4") - wav_path = os.path.join(tmp, "audio.wav") - out_path = os.path.join(tmp, "out.mp4") - with open(vid_path, "wb") as f: - f.write(video_mp4_bytes) - sf.write(wav_path, audio_pcm, sample_rate, format="WAV", subtype="PCM_16") - cmd = [ - ffmpeg_exe, - "-y", - "-nostdin", - "-hide_banner", - "-loglevel", - "error", - "-i", - vid_path, - "-i", - wav_path, - "-c:v", - "copy", - "-c:a", - "aac", - "-b:a", - "128k", - "-shortest", - "-movflags", - "+faststart", - out_path, - ] - subprocess.run(cmd, check=True, stdin=subprocess.DEVNULL, timeout=300) - with open(out_path, "rb") as f: - return f.read() - except ( - FileNotFoundError, - subprocess.CalledProcessError, - subprocess.TimeoutExpired, - OSError, - ) as e: - logger.warning("Synthetic video: audio mux failed (%s); using video-only MP4.", e) - return video_mp4_bytes + with tempfile.TemporaryDirectory(prefix="syn_vid_mux_") as tmp: + vid_path = os.path.join(tmp, "video.mp4") + wav_path = os.path.join(tmp, "audio.wav") + out_path = os.path.join(tmp, "out.mp4") + with open(vid_path, "wb") as f: + f.write(video_mp4_bytes) + sf.write(wav_path, audio_pcm, sample_rate, format="WAV", subtype="PCM_16") + cmd = [ + ffmpeg_exe, + "-y", + "-nostdin", + "-hide_banner", + "-loglevel", + "error", + "-i", + vid_path, + "-i", + wav_path, + "-c:v", + "copy", + "-c:a", + "aac", + "-b:a", + "128k", + "-shortest", + "-movflags", + "+faststart", + out_path, + ] + subprocess.run(cmd, check=True, stdin=subprocess.DEVNULL, timeout=300) + with open(out_path, "rb") as f: + return f.read() def generate_synthetic_video( @@ -383,6 +370,9 @@ def generate_synthetic_video( ``width`` / ``height`` / ``num_frames`` / ``embed_audio`` unless ``force_regenerate`` is true. Cache root: ``cache_dir`` if given, else the default temp subdirectory. + + ``embed_audio=True`` requires successful TTS generation and FFmpeg muxing. + Cached MP4s without the requested audio stream are regenerated. """ root = _resolve_synthetic_media_cache_dir(cache_dir) root.mkdir(parents=True, exist_ok=True) @@ -390,11 +380,19 @@ def generate_synthetic_video( if not force_regenerate and cache_path.is_file(): video_bytes = cache_path.read_bytes() - return { - "np_array": _np_array_from_mp4_bytes(video_bytes), - "base64": base64.b64encode(video_bytes).decode("utf-8"), - "file_path": str(cache_path.resolve()), - } + has_required_audio = True + if embed_audio: + import av + + with av.open(io.BytesIO(video_bytes)) as container: + has_required_audio = bool(container.streams.audio) + if has_required_audio: + return { + "np_array": _np_array_from_mp4_bytes(video_bytes), + "base64": base64.b64encode(video_bytes).decode("utf-8"), + "file_path": str(cache_path.resolve()), + } + logger.warning("Cached synthetic video %s has no audio stream; regenerating.", cache_path) import cv2 import imageio diff --git a/tests/helpers/tests/test_synthetic_video.py b/tests/helpers/tests/test_synthetic_video.py new file mode 100644 index 00000000000..ec617f5e46e --- /dev/null +++ b/tests/helpers/tests/test_synthetic_video.py @@ -0,0 +1,89 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import base64 +import io +import subprocess +from pathlib import Path + +import av +import numpy as np +import pytest + +from tests.helpers import media + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +@pytest.fixture +def synthetic_audio(mocker): + """Keep video tests independent of the host's TTS installation.""" + sample_rate = 48000 + samples = np.arange(sample_rate, dtype=np.float32) / sample_rate + waveform = (0.1 * np.sin(2 * np.pi * 440 * samples)).reshape(-1, 1) + return mocker.patch.object(media, "generate_synthetic_audio", return_value={"np_array": waveform}) + + +def test_audio_generation_failure_is_not_cached(mocker, tmp_path): + failure = RuntimeError("TTS dependency unavailable") + mocker.patch.object(media, "generate_synthetic_audio", side_effect=failure) + + with pytest.raises(RuntimeError, match="TTS dependency unavailable") as caught: + media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + + assert caught.value is failure + assert not list(tmp_path.glob("*.mp4")) + + +@pytest.mark.parametrize( + "failure", + [ + FileNotFoundError("FFmpeg unavailable"), + subprocess.CalledProcessError(1, ["ffmpeg"]), + subprocess.TimeoutExpired(["ffmpeg"], 300), + ], + ids=["missing-ffmpeg", "ffmpeg-error", "ffmpeg-timeout"], +) +def test_audio_mux_failure_is_propagated(mocker, synthetic_audio, failure): + mocker.patch("imageio_ffmpeg.get_ffmpeg_exe", return_value="ffmpeg") + mocker.patch.object(media.subprocess, "run", side_effect=failure) + + with pytest.raises(type(failure)) as caught: + media._mux_mp4_bytes_with_synthetic_audio(b"video", num_frames=6) + + assert caught.value is failure + + +def test_audio_in_video_repairs_video_only_cache(synthetic_audio, tmp_path): + video_only = media.generate_synthetic_video(64, 64, 6, cache_dir=tmp_path) + # Reproduce a cache entry written by the old silent audio-mux fallback. + cache_path = tmp_path / "synth_video_w64_h64_nf6_ea1.mp4" + original_bytes = Path(video_only["file_path"]).read_bytes() + cache_path.write_bytes(original_bytes) + + result = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + + assert result["file_path"] == str(cache_path) + raw = base64.b64decode(result["base64"]) + assert raw == cache_path.read_bytes() + assert raw != original_bytes + assert result["np_array"].shape == (6, 64, 64, 3) + with av.open(io.BytesIO(raw)) as container: + assert len(container.streams.video) == 1 + assert len(container.streams.audio) == 1 + audio_frames = list(container.decode(audio=0)) + assert sum(frame.samples for frame in audio_frames) > 0 + assert any(np.any(frame.to_ndarray()) for frame in audio_frames) + + +@pytest.mark.parametrize("embed_audio", [False, True], ids=["video-only", "with-audio"]) +def test_valid_video_cache_is_reused(mocker, synthetic_audio, tmp_path, embed_audio): + result = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) + encode = mocker.patch("imageio.get_writer", side_effect=AssertionError("valid cache was regenerated")) + + cached = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) + + encode.assert_not_called() + assert cached["base64"] == result["base64"] + assert cached["file_path"] == result["file_path"] + assert np.array_equal(cached["np_array"], media._np_array_from_mp4_bytes(base64.b64decode(result["base64"]))) From 7408f57d45de53d13ce2f1a3c9c48dacb7e2d0b8 Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Mon, 19 Oct 2026 15:35:47 +0800 Subject: [PATCH 2/6] [Bugfix] Recover corrupt synthetic video caches atomically A partially written MP4 may expose audio metadata before its payload is complete. Treat probe/decode failures and incomplete cached media as cache misses, and publish completed files through a same-directory temporary file and os.replace. Cover corrupt headers, truncated audio, atomic publication, and failed replacement cleanup. Signed-off-by: psv666 <2693925048@qq.com> --- tests/helpers/media.py | 46 ++++++++++--- tests/helpers/tests/test_synthetic_video.py | 71 +++++++++++++++++++++ 2 files changed, 107 insertions(+), 10 deletions(-) diff --git a/tests/helpers/media.py b/tests/helpers/media.py index 8029af608ad..7ddab53df75 100644 --- a/tests/helpers/media.py +++ b/tests/helpers/media.py @@ -372,7 +372,8 @@ def generate_synthetic_video( default temp subdirectory. ``embed_audio=True`` requires successful TTS generation and FFmpeg muxing. - Cached MP4s without the requested audio stream are regenerated. + Cached MP4s with corrupt/incomplete media or without the requested audio + stream are regenerated. Cache files are published atomically. """ root = _resolve_synthetic_media_cache_dir(cache_dir) root.mkdir(parents=True, exist_ok=True) @@ -384,15 +385,34 @@ def generate_synthetic_video( if embed_audio: import av - with av.open(io.BytesIO(video_bytes)) as container: - has_required_audio = bool(container.streams.audio) + try: + with av.open(io.BytesIO(video_bytes)) as container: + has_required_audio = False + if container.streams.audio: + audio_stream = container.streams.audio[0] + samples = sum(frame.samples for frame in container.decode(audio=0)) + expected_samples = 0 + if audio_stream.duration is not None and audio_stream.time_base is not None: + expected_samples = math.ceil( + audio_stream.duration * audio_stream.time_base * audio_stream.codec_context.sample_rate + ) + has_required_audio = samples > 0 and samples >= expected_samples + except av.error.FFmpegError: + has_required_audio = False if has_required_audio: - return { - "np_array": _np_array_from_mp4_bytes(video_bytes), - "base64": base64.b64encode(video_bytes).decode("utf-8"), - "file_path": str(cache_path.resolve()), - } - logger.warning("Cached synthetic video %s has no audio stream; regenerating.", cache_path) + try: + cached_frames = _np_array_from_mp4_bytes(video_bytes) + except RuntimeError: + cached_frames = None + if cached_frames is not None and len(cached_frames) == num_frames: + return { + "np_array": cached_frames, + "base64": base64.b64encode(video_bytes).decode("utf-8"), + "file_path": str(cache_path.resolve()), + } + logger.warning( + "Cached synthetic video %s is incomplete or lacks the requested audio; regenerating.", cache_path + ) import cv2 import imageio @@ -473,7 +493,13 @@ class _BouncingBall(TypedDict): else video_only_bytes ) - cache_path.write_bytes(video_bytes) + temp_cache = tempfile.NamedTemporaryFile(dir=root, prefix=f".{cache_path.name}.", suffix=".tmp", delete=False) + try: + with temp_cache: + temp_cache.write(video_bytes) + os.replace(temp_cache.name, cache_path) + finally: + Path(temp_cache.name).unlink(missing_ok=True) return { "np_array": np.array(video_frames), diff --git a/tests/helpers/tests/test_synthetic_video.py b/tests/helpers/tests/test_synthetic_video.py index ec617f5e46e..1ad8155d6da 100644 --- a/tests/helpers/tests/test_synthetic_video.py +++ b/tests/helpers/tests/test_synthetic_video.py @@ -87,3 +87,74 @@ def test_valid_video_cache_is_reused(mocker, synthetic_audio, tmp_path, embed_au assert cached["base64"] == result["base64"] assert cached["file_path"] == result["file_path"] assert np.array_equal(cached["np_array"], media._np_array_from_mp4_bytes(base64.b64decode(result["base64"]))) + + +@pytest.mark.parametrize("corruption", ["empty", "invalid-header", "truncated-audio"]) +def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruption): + result = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + cache_path = Path(result["file_path"]) + raw = cache_path.read_bytes() + if corruption == "empty": + corrupt = b"" + elif corruption == "invalid-header": + corrupt = b"not an MP4" + else: + with av.open(io.BytesIO(raw)) as container: + packets = [packet for packet in container.demux(audio=0) if packet.pos is not None and packet.size] + packet = packets[len(packets) // 2] + corrupt = raw[: packet.pos + packet.size // 2] + # +faststart leaves the stream metadata readable despite missing data. + with av.open(io.BytesIO(corrupt)) as container: + assert container.streams.audio + cache_path.write_bytes(corrupt) + + repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + + repaired_bytes = base64.b64decode(repaired["base64"]) + assert repaired_bytes == cache_path.read_bytes() + assert repaired_bytes != corrupt + assert repaired["np_array"].shape == (6, 64, 64, 3) + with av.open(io.BytesIO(repaired_bytes)) as container: + assert sum(frame.samples for frame in container.decode(audio=0)) > 0 + + +def test_video_cache_is_published_atomically(monkeypatch, synthetic_audio, tmp_path): + original = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + cache_path = Path(original["file_path"]) + old_bytes = cache_path.read_bytes() + replace = media.os.replace + published = [] + + def observe_replace(source, destination): + source = Path(source) + assert Path(destination) == cache_path + assert source.parent == cache_path.parent + # Until publication, readers must still see the old complete MP4. + assert cache_path.read_bytes() == old_bytes + completed = source.read_bytes() + with av.open(io.BytesIO(completed)) as container: + assert len(list(container.decode(video=0))) == 6 + with av.open(io.BytesIO(completed)) as container: + assert sum(frame.samples for frame in container.decode(audio=0)) > 0 + published.append(completed) + replace(source, destination) + + monkeypatch.setattr(media.os, "replace", observe_replace) + result = media.generate_synthetic_video(64, 64, 6, embed_audio=True, force_regenerate=True, cache_dir=tmp_path) + + assert published == [cache_path.read_bytes()] + assert base64.b64decode(result["base64"]) == published[0] + assert list(tmp_path.iterdir()) == [cache_path] + + +def test_failed_cache_publication_preserves_old_file(mocker, synthetic_audio, tmp_path): + original = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + cache_path = Path(original["file_path"]) + old_bytes = cache_path.read_bytes() + mocker.patch.object(media.os, "replace", side_effect=OSError("publication failed")) + + with pytest.raises(OSError, match="publication failed"): + media.generate_synthetic_video(64, 64, 6, embed_audio=True, force_regenerate=True, cache_dir=tmp_path) + + assert cache_path.read_bytes() == old_bytes + assert list(tmp_path.iterdir()) == [cache_path] From a385408e7397f301b94e5d24051331e571ad9e43 Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Mon, 19 Oct 2026 16:03:34 +0800 Subject: [PATCH 3/6] [Bugfix] Preserve shared fixture cache permissions Create the atomic cache staging file exclusively with normal umask-controlled permissions. Keep every video packet in the audio truncation regression so its failure depends on the decoded audio sample count, and verify creation permissions under shared and restrictive umasks. Signed-off-by: psv666 <2693925048@qq.com> --- tests/helpers/media.py | 9 ++++-- tests/helpers/tests/test_synthetic_video.py | 33 +++++++++++++++++---- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/tests/helpers/media.py b/tests/helpers/media.py index 7ddab53df75..25e41434803 100644 --- a/tests/helpers/media.py +++ b/tests/helpers/media.py @@ -20,6 +20,7 @@ import tempfile import threading import time +import uuid from concurrent.futures.process import BrokenProcessPool from contextlib import contextmanager from pathlib import Path @@ -493,13 +494,15 @@ class _BouncingBall(TypedDict): else video_only_bytes ) - temp_cache = tempfile.NamedTemporaryFile(dir=root, prefix=f".{cache_path.name}.", suffix=".tmp", delete=False) + temp_cache_path = root / f".{cache_path.name}.{uuid.uuid4().hex}.tmp" + # Exclusive creation preserves the normal 0666 & ~umask cache permissions. + temp_cache = temp_cache_path.open("xb") try: with temp_cache: temp_cache.write(video_bytes) - os.replace(temp_cache.name, cache_path) + os.replace(temp_cache_path, cache_path) finally: - Path(temp_cache.name).unlink(missing_ok=True) + temp_cache_path.unlink(missing_ok=True) return { "np_array": np.array(video_frames), diff --git a/tests/helpers/tests/test_synthetic_video.py b/tests/helpers/tests/test_synthetic_video.py index 1ad8155d6da..63b1e0acabb 100644 --- a/tests/helpers/tests/test_synthetic_video.py +++ b/tests/helpers/tests/test_synthetic_video.py @@ -3,6 +3,8 @@ import base64 import io +import os +import stat import subprocess from pathlib import Path @@ -100,12 +102,21 @@ def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruptio corrupt = b"not an MP4" else: with av.open(io.BytesIO(raw)) as container: - packets = [packet for packet in container.demux(audio=0) if packet.pos is not None and packet.size] - packet = packets[len(packets) // 2] - corrupt = raw[: packet.pos + packet.size // 2] - # +faststart leaves the stream metadata readable despite missing data. + packets = [ + (packet.stream.type, packet.pos, packet.size) + for packet in container.demux() + if packet.pos is not None and packet.size + ] + video_end = max(pos + size for kind, pos, size in packets if kind == "video") + audio_tail = next(pos for kind, pos, _ in packets if kind == "audio" and pos >= video_end) + corrupt = raw[:audio_tail] + # Keep every video packet and truncate at an audio packet boundary: + # neither video completeness nor a decode exception can reject this. + assert len(media._np_array_from_mp4_bytes(corrupt)) == 6 with av.open(io.BytesIO(corrupt)) as container: - assert container.streams.audio + stream = container.streams.audio[0] + samples = sum(frame.samples for frame in container.decode(audio=0)) + assert 0 < samples < stream.duration * stream.time_base * stream.codec_context.sample_rate cache_path.write_bytes(corrupt) repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) @@ -158,3 +169,15 @@ def test_failed_cache_publication_preserves_old_file(mocker, synthetic_audio, tm assert cache_path.read_bytes() == old_bytes assert list(tmp_path.iterdir()) == [cache_path] + + +@pytest.mark.parametrize("creation_umask", [0o022, 0o002, 0o077]) +def test_video_cache_respects_creation_permissions(synthetic_audio, tmp_path, creation_umask): + previous_umask = os.umask(creation_umask) + try: + result = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + finally: + os.umask(previous_umask) + + cache_path = Path(result["file_path"]) + assert stat.S_IMODE(cache_path.stat().st_mode) == 0o666 & ~creation_umask From 01893cc01099b10e5a4181c0f28ae8bca4a20d2d Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Tue, 20 Oct 2026 11:26:55 +0800 Subject: [PATCH 4/6] [CI/Build] Cover video-only fixture cache recovery Exercise empty and invalid video-only caches through the actual decode failure path, and add truncated video cases with and without audio. Assert that partial videos remain decodable and are repaired to the complete frame count, preserving audio only when requested. Signed-off-by: psv666 <2693925048@qq.com> --- tests/helpers/tests/test_synthetic_video.py | 53 +++++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/tests/helpers/tests/test_synthetic_video.py b/tests/helpers/tests/test_synthetic_video.py index 63b1e0acabb..dedaf81dcc7 100644 --- a/tests/helpers/tests/test_synthetic_video.py +++ b/tests/helpers/tests/test_synthetic_video.py @@ -91,9 +91,18 @@ def test_valid_video_cache_is_reused(mocker, synthetic_audio, tmp_path, embed_au assert np.array_equal(cached["np_array"], media._np_array_from_mp4_bytes(base64.b64decode(result["base64"]))) -@pytest.mark.parametrize("corruption", ["empty", "invalid-header", "truncated-audio"]) -def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruption): - result = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) +@pytest.mark.parametrize( + "corruption, embed_audio", + [ + pytest.param("empty", False, id="empty-video-only"), + pytest.param("empty", True, id="empty-with-audio"), + pytest.param("invalid-header", False, id="invalid-header-video-only"), + pytest.param("invalid-header", True, id="invalid-header-with-audio"), + pytest.param("truncated-audio", True, id="truncated-audio"), + ], +) +def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruption, embed_audio): + result = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) cache_path = Path(result["file_path"]) raw = cache_path.read_bytes() if corruption == "empty": @@ -117,16 +126,50 @@ def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruptio stream = container.streams.audio[0] samples = sum(frame.samples for frame in container.decode(audio=0)) assert 0 < samples < stream.duration * stream.time_base * stream.codec_context.sample_rate + if not embed_audio: + with pytest.raises(RuntimeError): + media._np_array_from_mp4_bytes(corrupt) cache_path.write_bytes(corrupt) - repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=True, cache_dir=tmp_path) + repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) repaired_bytes = base64.b64decode(repaired["base64"]) assert repaired_bytes == cache_path.read_bytes() assert repaired_bytes != corrupt assert repaired["np_array"].shape == (6, 64, 64, 3) with av.open(io.BytesIO(repaired_bytes)) as container: - assert sum(frame.samples for frame in container.decode(audio=0)) > 0 + if embed_audio: + assert sum(frame.samples for frame in container.decode(audio=0)) > 0 + else: + assert not container.streams.audio + synthetic_audio.assert_not_called() + + +@pytest.mark.parametrize("embed_audio", [False, True], ids=["video-only", "with-audio"]) +def test_truncated_video_cache_is_regenerated(synthetic_audio, tmp_path, embed_audio): + result = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) + cache_path = Path(result["file_path"]) + raw = cache_path.read_bytes() + with av.open(io.BytesIO(raw)) as container: + last_video = max(packet.pos for packet in container.demux(video=0) if packet.pos is not None and packet.size) + corrupt = raw[:last_video] + # The video still opens, so the frame-count check must reject video-only caches. + assert 0 < len(media._np_array_from_mp4_bytes(corrupt)) < 6 + cache_path.write_bytes(corrupt) + + repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) + + repaired_bytes = base64.b64decode(repaired["base64"]) + assert repaired["np_array"].shape == (6, 64, 64, 3) + assert repaired_bytes == cache_path.read_bytes() + assert repaired_bytes != corrupt + assert len(media._np_array_from_mp4_bytes(repaired_bytes)) == 6 + with av.open(io.BytesIO(repaired_bytes)) as container: + if embed_audio: + assert sum(frame.samples for frame in container.decode(audio=0)) > 0 + else: + assert not container.streams.audio + synthetic_audio.assert_not_called() def test_video_cache_is_published_atomically(monkeypatch, synthetic_audio, tmp_path): From 4d2214126838a4e0d66980eb17983240d91ed7c4 Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Tue, 20 Oct 2026 11:55:00 +0800 Subject: [PATCH 5/6] [CI/Build] Isolate cached video frame-count coverage Signed-off-by: psv666 <2693925048@qq.com> --- tests/helpers/tests/test_synthetic_video.py | 25 ++++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/tests/helpers/tests/test_synthetic_video.py b/tests/helpers/tests/test_synthetic_video.py index dedaf81dcc7..3883b589d84 100644 --- a/tests/helpers/tests/test_synthetic_video.py +++ b/tests/helpers/tests/test_synthetic_video.py @@ -146,15 +146,28 @@ def test_corrupt_video_cache_is_regenerated(synthetic_audio, tmp_path, corruptio @pytest.mark.parametrize("embed_audio", [False, True], ids=["video-only", "with-audio"]) -def test_truncated_video_cache_is_regenerated(synthetic_audio, tmp_path, embed_audio): +def test_incomplete_video_cache_is_regenerated(synthetic_audio, tmp_path, embed_audio): result = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) cache_path = Path(result["file_path"]) - raw = cache_path.read_bytes() - with av.open(io.BytesIO(raw)) as container: - last_video = max(packet.pos for packet in container.demux(video=0) if packet.pos is not None and packet.size) - corrupt = raw[:last_video] - # The video still opens, so the frame-count check must reject video-only caches. + if embed_audio: + # Keep a complete audio stream while violating the six-frame cache key. + shorter = media.generate_synthetic_video(64, 64, 5, embed_audio=True, cache_dir=tmp_path) + corrupt = base64.b64decode(shorter["base64"]) + else: + raw = cache_path.read_bytes() + with av.open(io.BytesIO(raw)) as container: + last_video = max( + packet.pos for packet in container.demux(video=0) if packet.pos is not None and packet.size + ) + corrupt = raw[:last_video] + # Both modes must reach frame-count validation rather than a decode failure. assert 0 < len(media._np_array_from_mp4_bytes(corrupt)) < 6 + if embed_audio: + with av.open(io.BytesIO(corrupt)) as container: + stream = container.streams.audio[0] + samples = sum(frame.samples for frame in container.decode(audio=0)) + assert samples > 0 + assert samples >= stream.duration * stream.time_base * stream.codec_context.sample_rate cache_path.write_bytes(corrupt) repaired = media.generate_synthetic_video(64, 64, 6, embed_audio=embed_audio, cache_dir=tmp_path) From ac6c5fd8e92281195513e93114d89fbcaf52c6c5 Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Tue, 20 Oct 2026 14:34:20 +0800 Subject: [PATCH 6/6] [CI/Build] Exercise real duplex commit helpers in PCM regression test Signed-off-by: psv666 <2693925048@qq.com> --- .../test_auto_response_commit_consume.py | 50 ++++++------------- 1 file changed, 14 insertions(+), 36 deletions(-) diff --git a/tests/engine/duplex/test_auto_response_commit_consume.py b/tests/engine/duplex/test_auto_response_commit_consume.py index 915754e8299..1fcada36fab 100644 --- a/tests/engine/duplex/test_auto_response_commit_consume.py +++ b/tests/engine/duplex/test_auto_response_commit_consume.py @@ -29,28 +29,6 @@ def _samples(audio: str) -> list[float]: return [struct.unpack_from(" None: - self.payload = payload - self.operation_id = operation_id - self.byte_count = 4 - - def commit(self) -> None: - return None - - -class _Buffer: - def __init__(self, payloads: list[dict[str, object]]) -> None: - self._payloads = list(payloads) - self._index = 0 - - def prepare_commit(self, *, operation_id: str, chunk_period_ms: int) -> _Reservation: - del chunk_period_ms - payload = self._payloads[self._index] - self._index += 1 - return _Reservation(payload, operation_id) - - class _Model: def __init__(self) -> None: self.submitted: list[str] = [] @@ -84,12 +62,6 @@ async def test_auto_response_commit_consumes_retained_pcm( lambda *args, **kwargs: {"type": "input.committed"}, ) state = AuraServingSessionState() - state.audio_buffer = _Buffer( - [ - {"format": "pcm_f32le", "sample_rate_hz": 16000, "audio": _pcm([1.0]), "is_speech": True}, - {"format": "pcm_f32le", "sample_rate_hz": 16000, "audio": _pcm([2.0]), "is_speech": True}, - ] - ) session = SimpleNamespace( turn_id=0, epoch=0, @@ -121,14 +93,20 @@ async def start_append(payload: dict[str, object], **kwargs: object) -> bool: ) return await attempt._submit() - runner = SimpleNamespace( - session=session, - model_state=state, - emit=lambda payload: None, - _start_append=start_append, - ) - for item_id in item_ids: - await DuplexSessionRunner._commit_and_start_auto_response(runner, {}, realtime_item_id=item_id) + # Keep the real commit helpers bound without starting a session or a stage. + runner = DuplexSessionRunner.__new__(DuplexSessionRunner) + runner.session = session + runner.model_state = state + monkeypatch.setattr(runner, "emit", lambda payload: None) + monkeypatch.setattr(runner, "_start_append", start_append) + for index, item_id in enumerate(item_ids): + state.audio_buffer.prepare_append( + {"format": "pcm_f32le", "sample_rate_hz": 16000, "audio": _pcm([float(index + 1)]), "is_speech": True}, + operation_id=f"append-{index}", + chunk_period_ms=1000, + allow_emit=False, + ) + await runner._commit_and_start_auto_response({}, realtime_item_id=item_id) assert [_samples(audio) for audio in model.submitted] == [[1.0], [2.0]] assert state.committed_audio_payload is None