Coverage for tests/streamwise_app/test_gen_video_chunked.py: 98%
121 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1#!/usr/bin/env python3
3import sys
4import os
5import pytest
6import logging
8from PIL import Image
10from typing import Tuple
12from unittest.mock import patch
13from unittest.mock import MagicMock
15# Add current path
16sys.path.append(os.getcwd())
18from file_utils import read_file_base64
20from tests.torch_mock import TorchMock
21from tests.test_utils import temp_sys_path
23mock_torch = TorchMock()
25mock_modules = {}
26mock_modules.update(mock_torch.get_sub_modules())
28with patch.dict(sys.modules, mock_modules):
29 with temp_sys_path("apps"):
30 from apps.gen_video_chunked import GenVideoChunked
31 from apps.gen_video_chunked import SubVideoInfo
33 from media_utils import get_video_frames
34 from media_utils import empty_audio_file
35 from media_utils import get_video_file_info
37 from video import HUNYUANFRAMEPACK_FPS
39 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock
40 from tests.streamwise_app.lmm_generator_mock import MOCK_COLORS_RGB
43@pytest.mark.asyncio
44async def test_gen_slide_video_chunks_noservice() -> None:
45 gen = LMMGeneratorMock()
46 logger = logging
47 gen_video_chunked = GenVideoChunked(
48 video_id=0,
49 gen=gen,
50 job_path="tests/data",
51 logger=logger, # type: ignore[arg-type]
52 )
54 with pytest.raises(FileNotFoundError, match="Audio file not found: audio.wav"):
55 await gen_video_chunked.gen_video_chunked(
56 audio_path="audio.wav",
57 image=Image.new("RGB", (640, 480), color="blue"),
58 prompt="A test slide",
59 neg_prompt="",
60 width=640,
61 height=480,
62 num_steps=10,
63 )
65 # TODO
66 with pytest.raises(Exception): # , match="Error generating video.+hunyuanframepackf1"):
67 await gen_video_chunked.gen_video_chunked(
68 audio_path="tests/data/audio_4675.wav",
69 image=Image.new("RGB", (640, 480), color="blue"),
70 prompt="A test slide",
71 neg_prompt="",
72 width=640,
73 height=480,
74 num_steps=10,
75 )
78def test_scene_info() -> None:
79 video = []
80 video.append(SubVideoInfo(0, 2))
81 video.append(SubVideoInfo(2, 5))
83 scene0 = video[0]
84 assert scene0 is not None
85 assert scene0.get_seconds() == 2
86 assert scene0.get_frames(fps=23) == (0, 46)
87 assert str(scene0) == "SubVideoInfo(0.000-2.000s)"
89 scene1 = video[1]
90 assert scene1 is not None
91 assert scene1.get_seconds() == 3
92 assert scene1.get_frames(fps=20) == (40, 100)
93 assert str(scene1) == "SubVideoInfo(2.000-5.000s)"
95 # assert video.get_num_audio_frames() == 118
98@pytest.mark.asyncio
99async def test_gen_subvideo() -> None:
100 service_manager = MagicMock()
101 service_manager.get_service_url = MagicMock(
102 return_value="http://mock_service_url:1234"
103 )
105 # Chunk generator
106 gen = LMMGeneratorMock()
107 logger = logging
108 gen_video_chunked = GenVideoChunked(
109 video_id=1,
110 gen=gen,
111 job_path="tests/data",
112 logger=logger, # type: ignore[arg-type]
113 )
115 # Chunk components
116 subvideo_info = SubVideoInfo(
117 start_seconds=0.0,
118 end_seconds=4.0,
119 )
120 assert subvideo_info.get_start_frame(HUNYUANFRAMEPACK_FPS) == 0
121 assert subvideo_info.get_end_frame(HUNYUANFRAMEPACK_FPS) == 120 # 4 secs x 30 FPS
123 audio_base64 = await read_file_base64("tests/data/audio_4675.wav")
124 video_frames = [
125 Image.new("RGB", (640, 480), color=color)
126 for color in ["blue", "green", "red", "yellow"]
127 ]
129 # Generation
130 subvideo_task = await gen_video_chunked._gen_subvideo(
131 subvideo_id=0,
132 subvideo_info=subvideo_info,
133 video_frames=video_frames,
134 audio_base64=audio_base64,
135 width=640,
136 height=480,
137 num_steps=10,
138 )
139 assert subvideo_task is not None
140 subvideo_binary = await subvideo_task
141 assert subvideo_binary is not None
143 # Check data
144 subvideo_frames = await get_video_frames(subvideo_binary)
145 assert 90 <= len(subvideo_frames) <= 94 # == 1 + (4.0 * 23) # 4 secs x 23 FPS = 92
146 frame0 = subvideo_frames[0]
147 assert frame0.size == (640, 480)
149 # Check metadata
150 video_file_info = get_video_file_info(subvideo_binary)
151 video_info = video_file_info["video"]
152 assert video_info["codec"] == "h264"
153 assert video_info["duration_seconds"] is not None
154 assert 3.9 <= video_info["duration_seconds"] <= 4.1 # 4 seconds
157@pytest.mark.asyncio
158async def test_gen_slide_video_chunks() -> None:
159 service_manager = MagicMock()
160 service_manager.get_service_url = MagicMock(
161 return_value="http://mock_service_url:1234"
162 )
163 # job_id = "test_gen_slide_video_chunks"
164 # job = StreamPersonaJob(job_id, service_manager)
165 # Mocking generation
166 # job = await _mock_generation(job)
168 gen = LMMGeneratorMock()
169 logger = logging
170 gen_video_chunked = GenVideoChunked(
171 video_id=2,
172 gen=gen,
173 job_path="tests/data",
174 logger=logger, # type: ignore[arg-type]
175 )
177 # Failure case: audio too short (4.6s < 5s minimum)
178 with pytest.raises(ValueError, match="Audio too short for chunked generation"):
179 await gen_video_chunked.gen_video_chunked(
180 audio_path="tests/data/audio_4675.wav",
181 image=Image.new("RGB", (320, 200), color="pink"),
182 prompt="A test slide 1",
183 neg_prompt="",
184 width=1280,
185 height=800,
186 num_steps=20,
187 )
189 # Success case with 2 chunks
190 audio_path = empty_audio_file(duration_seconds=7.0)
191 video_binary = await gen_video_chunked.gen_video_chunked(
192 audio_path=audio_path,
193 image=Image.new("RGB", (640, 480), color="pink"),
194 prompt="A test slide 1",
195 neg_prompt="",
196 width=1280,
197 height=800,
198 num_steps=20,
199 )
200 assert video_binary is not None
202 # Check metadata
203 video_file_info = get_video_file_info(video_binary)
204 video_info = video_file_info["video"]
205 assert video_info["codec"] == "h264"
206 assert video_info["num_frames"] is not None
207 assert 1 + 161 <= video_info["num_frames"] <= 1 + 161 + 4 # 7 secs * 23 FPS = 161
208 assert video_info["width"] == 1280
209 assert video_info["height"] == 800
210 assert video_info["fps"] == 23
211 assert video_info["duration_seconds"] is not None
212 assert 6.9 <= video_info["duration_seconds"] <= 7.1 # 7 seconds
213 audio_info = video_file_info["audio"]
214 assert audio_info["codec"] == "aac"
215 assert audio_info["duration_seconds"] is not None
216 assert 6.9 <= audio_info["duration_seconds"] <= 7.1 # 7 seconds
218 # Check actual data
219 video_frames = await get_video_frames(video_binary)
220 assert 1 + 161 <= len(video_frames) <= 1 + 161 + 4 # 7 secs * 23 FPS = 161
222 expected_color = MOCK_COLORS_RGB["gen_video_audio_from_video"]
223 frame = video_frames[0]
224 assert frame.size == (1280, 800)
225 pixel_color = frame.convert("RGB").getpixel((0, 0))
226 assert is_pixel_color(pixel_color, expected_color), ( # type: ignore[arg-type]
227 f"Frame 0: {pixel_color}!={expected_color}")
228 for ix, frame in enumerate(video_frames):
229 assert frame.size == (1280, 800)
230 pixel_color = frame.convert("RGB").getpixel((0, 0))
231 assert is_pixel_color(pixel_color, expected_color), ( # type: ignore[arg-type]
232 f"Frame {ix}: {pixel_color}!={expected_color}")
234 # TODO test with upscaling
235 # TODO test with debug
238def is_pixel_color(
239 pixel_color: Tuple[int, int, int],
240 expected_color: Tuple[int, int, int],
241) -> bool:
242 """Check if pixel_color is approximately equal to expected_color."""
243 r, g, b = pixel_color
244 er, eg, eb = expected_color
245 if abs(r - er) > 2:
246 return False
247 if abs(g - eg) > 2:
248 return False
249 if abs(b - eb) > 2:
250 return False
251 return True
254"""
255@pytest.mark.timeout(5)
256@pytest.mark.asyncio
257async def test_gen_scene_chunks() -> None:
258 service_manager = MagicMock()
260 job_id = "gen_scene_chunks"
261 job = StreamCastJob(job_id, service_manager)
263 with pytest.raises(Exception, match="Service 'hunyuanframepackf1' not found"):
264 await job.gen_scene_chunks(
265 scene_id=0,
266 audio_path="tests/data/audio_4675.wav",
267 image=Image.new("RGB", (80, 80), color="blue"),
268 video_prompt="Video prompt.")
270 # Mocking the LMM generator
271 job.gen = await _get_gen_mock(job.gen)
273 # TODO go further than this
274 with pytest.raises(ValueError, match="No sub-scenes generated for scene 0"):
275 video_binary = await job.gen_scene_chunks(
276 scene_id=0,
277 audio_path="tests/data/audio_4675.wav",
278 image=Image.new("RGB", (80, 80), color="blue"),
279 video_prompt="Video prompt.")
280 assert video_binary is not None
282 del job
283 del service_manager
286def _get_soundfile_mock() -> MagicMock:
287 soundfile_mock = MagicMock()
288 sample_rate = 16 * 1000
289 audio_data = [1.0] * sample_rate * 5 # 5 seconds of dummy audio
290 soundfile_mock.read = MagicMock(return_value=(audio_data, sample_rate))
291 soundfile_mock.write = MagicMock()
292 return soundfile_mock
295@pytest.mark.asyncio
296async def test_podcast_gen_sub_scene() -> None:
297 # TODO
298 service_manager = MagicMock()
299 job = StreamCastJob("test_podcast_gen_sub_scene", service_manager)
300 audio_base64 = await read_file_base64("tests/data/audio_4675.wav")
302 job.gen = await _get_gen_mock(job.gen)
304 scene_frames = [
305 Image.new("RGB", (80, 50), color="pink")
306 for _ in range(109)
307 ]
309 with patch.dict(sys.modules, {
310 'soundfile': _get_soundfile_mock(),
311 }):
312 sub_scene_task = await job.gen_sub_scene(
313 scene_id=1,
314 sub_scene_id=2,
315 sub_scene_info=SubVideoInfo(1, 2),
316 scene_frames=scene_frames,
317 audio_base64=audio_base64,
318 )
319 assert sub_scene_task is not None
320 await sub_scene_task
322 # No frames should return None
323 with patch.dict(sys.modules, {
324 'soundfile': _get_soundfile_mock(),
325 }):
326 sub_scene_task = await job.gen_sub_scene(
327 scene_id=2,
328 sub_scene_id=3,
329 sub_scene_info=SubSceneInfo(2, 3),
330 scene_frames=[],
331 audio_base64=audio_base64,
332 )
333 assert sub_scene_task is None
335 await job.close()
336 del service_manager
337"""