Coverage for tests/streamwise_app/test_streamcast.py: 99%
356 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1#!/usr/bin/env python3
2"""
3Unit tests for StreamCast.
4"""
6import os
7import sys
8import pytest
9import aiofiles
11from PIL import Image
13from http import HTTPStatus
15from quart import Quart
17from unittest.mock import patch
18from unittest.mock import MagicMock
19from unittest.mock import AsyncMock
21# Add current path
22sys.path.append(os.getcwd())
24from tests.test_utils import temp_sys_path
25from tests.torch_mock import TorchMock
26from tests.k8s_mock import K8sMock
27from tests.fantasytalking_mock import FantasyTalkingMock
28from tests.streamwise_app.app_test_helpers import check_app_root
29from tests.streamwise_app.app_test_helpers import check_health
30from tests.streamwise_app.app_test_helpers import check_files
31from tests.streamwise_app.app_test_helpers import check_unknown_route
32from tests.streamwise_app.app_test_helpers import check_job_submit_page
33from tests.streamwise_app.app_test_helpers import check_job_status_page
34from tests.streamwise_app.app_test_helpers import check_api_job_status
35from tests.streamwise_app.app_test_helpers import check_api_job_requests
37from file_utils import binary_to_base64
38from file_utils import read_file_base64
39from file_utils import save_base64_as_binary
41with temp_sys_path("apps", "apps/streamcast"):
42 from character import Character
43 from streamcast_job import split_text_lines
46mock_torch = TorchMock()
47mock_k8s = K8sMock()
48mock_ft = FantasyTalkingMock()
50mock_modules = {
51 "imageio": MagicMock(),
52 "tabulate": MagicMock(),
53 "soundfile": MagicMock(),
54}
55mock_modules.update(mock_torch.get_sub_modules())
56mock_modules.update(mock_k8s.get_sub_modules())
57mock_modules.update(mock_ft.get_sub_modules())
59with patch.dict(sys.modules, mock_modules):
60 from media_utils import get_audio_duration
61 from media_utils import get_video_frames
62 from media_utils import get_video_file_info
64 # with temp_sys_path("streamcast"):
65 from apps.lmm_generator import LMMGenerator
67 with temp_sys_path("apps", "apps/streamcast"):
68 from apps.streamcast.streamcast import StreamCastApp
69 from apps.streamcast.streamcast_job import StreamCastJob
70 from apps.streamcast.streamcast_job import OutputMode
71 from apps.streamcast.streamcast_job import JobStatus
73 with temp_sys_path("wrapper", "wrapper/fantasytalking"):
74 from wrapper_fantasytalking import FantasyTalking
76 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock
79streamcast_app = StreamCastApp()
82@pytest.fixture(name="test_app")
83def _test_app() -> Quart:
84 return streamcast_app.app
87@pytest.mark.asyncio
88async def test_app(test_app: Quart) -> None:
89 """Check that GET / returns 200."""
90 await check_app_root(test_app, "StreamCast")
93@pytest.mark.asyncio
94async def test_health(test_app: Quart) -> None:
95 """Check /health."""
96 await check_health(test_app)
99@pytest.mark.asyncio
100async def test_files(test_app: Quart) -> None:
101 """Check /files endpoint."""
102 await check_files(test_app, "streamcast")
105@pytest.mark.asyncio
106async def test_unknown_route(test_app: Quart) -> None:
107 """Check that an unknown route returns 404."""
108 await check_unknown_route(test_app)
111@pytest.mark.asyncio
112async def test_server() -> None:
113 """Check the HTTP server start."""
114 # TODO this test does not work
115 """
116 test_args = ["streamcast.py"]
117 with patch.object(sys, "argv", test_args):
118 await main()
119 """
120 pass
123@pytest.mark.asyncio
124async def test_index(test_app: Quart) -> None:
125 """Check the HTTP server content via client."""
126 client = test_app.test_client()
127 response = await client.get("/")
128 assert response is not None
129 # TODO sometimes it returns 500 Internal Server Error
130 """
131 assert response.status_code == HTTPStatus.OK
132 text = await response.get_data(as_text=True)
133 assert "index page" in text or len(text) > 0
134 """
137@pytest.mark.asyncio
138async def test_job_submit_page(test_app: Quart) -> None:
139 """Check the web for job submission."""
140 await check_job_submit_page(test_app)
143@pytest.mark.asyncio
144async def test_job_status_page(test_app: Quart) -> None:
145 """Check the web for job status."""
146 await check_job_status_page(test_app)
149@pytest.mark.asyncio
150async def test_api_job_status(test_app: Quart) -> None:
151 """Check the API for job status."""
152 await check_api_job_status(test_app)
155@pytest.mark.asyncio
156async def test_api_job_requests(test_app: Quart) -> None:
157 """Check the API for job requests."""
158 await check_api_job_requests(test_app)
161@pytest.mark.asyncio
162async def test_api_job_submit(test_app: Quart) -> None:
163 """Check the API for job requests."""
164 client = test_app.test_client()
166 # Mock the service manager
167 streamcast_app.service_manager = MagicMock()
169 response = await client.post("/api/job")
170 assert response.status_code == HTTPStatus.BAD_REQUEST
171 response_json = await response.get_json()
172 assert response_json["status"] == "error"
173 assert response_json["error"] == "No JSON body received"
175 response = await client.post("/api/job", json={"test": 1})
176 assert response.status_code == HTTPStatus.BAD_REQUEST
177 response_json = await response.get_json()
178 assert response_json["status"] == "error"
179 assert response_json["error"] == "Missing 'pdf_base64' in request"
181 response = await client.post("/api/job", json={"pdf_base64": "AAA"})
182 assert response.status_code == HTTPStatus.BAD_REQUEST
183 response_json = await response.get_json()
184 assert response_json["status"] == "error"
185 assert response_json["error"] == "Incorrect padding"
187 pdf_base64 = binary_to_base64(b"BLANK")
188 response = await client.post("/api/job", json={"pdf_base64": pdf_base64})
189 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
190 response_json = await response.get_json()
191 assert response_json["status"] == "error"
192 assert "Error generating podcast transcript" in response_json["error"]
194 # Success case
195 pdf_path = "tests/data/blank.pdf"
196 async with aiofiles.open(pdf_path, "rb") as file:
197 pdf_binary = await file.read()
198 pdf_base64 = binary_to_base64(pdf_binary)
199 response = await client.post("/api/job", json={"pdf_base64": pdf_base64})
200 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
201 # TODO fix the generation
202 # response_json = await response.get_json()
203 # assert "job_id" in response_json
204 # assert response_json["status"] == "success"
207def test_parse_args() -> None:
208 test_args = ["streamcast.py", "--k8s", "--num_dialogues", "7"]
209 with patch.object(sys, "argv", test_args):
210 pass
211 # args = streamcast.parse_args()
212 """
213 assert args.num_characters == 2
214 assert args.num_dialogues == 7
215 """
218@pytest.mark.asyncio
219async def test_gen_podcast_transcript() -> None:
220 service_manager = MagicMock()
221 service_manager.get_service_url = MagicMock(
222 return_value="http://mock_service_url:1234"
223 )
224 job_id = "gen_podcast_transcript"
225 gen = LMMGenerator("streamcast", job_id, service_manager)
227 # Mock aiohttp session
228 gen.session = MagicMock()
229 gen.session.post = MagicMock()
230 gen.session.close = AsyncMock()
232 async for line_json in gen.gen_podcast_transcript(pdf_base64="AAAA"):
233 assert line_json is not None
235 async for line_json in gen.gen_podcast_transcript(
236 pdf_base64="AAAA",
237 style_prompt="An epic fantasy story.",
238 scene_prompt="A battle between good and evil.",
239 custom_prompt="Include dragons and magic.",
240 ):
241 assert line_json is not None
243 await gen.stop()
244 del gen
245 del service_manager
248@pytest.mark.asyncio
249async def test_gen_scene() -> None:
250 service_manager = MagicMock()
251 job_id = "test_podcast_gen_sub_scenes"
252 job = StreamCastJob(job_id, service_manager)
254 character = Character("Jane", gender="Female")
256 # TODO we could mock gen_audio() to return a valid audio
257 with pytest.raises(Exception, match="Service 'kokoro' not found"):
258 await job.gen_scene(
259 scene_id=0,
260 character=character,
261 text="Test text.")
263 del job
264 del service_manager
267@pytest.mark.asyncio
268async def test_gen_scene_mock() -> None:
269 service_manager = MagicMock()
270 job_id = "test_podcast_gen_sub_scenes"
271 job = StreamCastJob(job_id, service_manager)
272 job.config["output_mode"] = OutputMode.UNKNOWN
274 job.gen = LMMGeneratorMock()
276 # Generate scene for character
277 character = Character("Jane", gender="Female")
278 video_path = await job.gen_scene(
279 scene_id=0,
280 character=character,
281 text="Test text.")
282 assert video_path.endswith(".mp4")
283 assert os.path.exists(video_path) is True
285 del job
286 del service_manager
289@pytest.mark.asyncio
290async def test_gen_scene_mock_video_audio_synced() -> None:
291 service_manager = MagicMock()
292 job_id = "gen_scene_mock_video_audio_synced"
293 job = StreamCastJob(job_id, service_manager)
294 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
296 job.gen = LMMGeneratorMock()
298 # Generate scene for character
299 character = Character("Jane", gender="Female")
300 video_path = await job.gen_scene(
301 scene_id=0,
302 character=character,
303 text="Test text.")
304 assert video_path.endswith(".mp4")
305 assert os.path.exists(video_path) is True
307 del job
308 del service_manager
311@pytest.mark.asyncio
312async def test_gen_scene_mock_video_audio_unsynced() -> None:
313 service_manager = MagicMock()
314 job_id = "gen_scene_mock_video_audio_unsynced"
315 job = StreamCastJob(job_id, service_manager)
316 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_UNSYNCED
318 job.gen = LMMGeneratorMock()
320 # Generate scene for character
321 character = Character("Jane", gender="Female")
322 video_path = await job.gen_scene(
323 scene_id=0,
324 character=character,
325 text="Test text.")
326 assert video_path.endswith(".mp4")
327 assert os.path.exists(video_path) is True
329 del job
330 del service_manager
333@pytest.mark.asyncio
334async def test_gen_scene_mock_audio() -> None:
335 service_manager = MagicMock()
336 job_id = "gen_scene_mock_audio"
337 job = StreamCastJob(job_id, service_manager)
338 job.config["output_mode"] = OutputMode.AUDIO_ONLY
340 job.gen = LMMGeneratorMock()
342 # Generate scene for character
343 character = Character("Jane", gender="Female")
344 video_path = await job.gen_scene(
345 scene_id=0,
346 character=character,
347 text="Test text.")
348 assert video_path.endswith(".mp4")
349 assert os.path.exists(video_path) is True
351 del job
352 del service_manager
355@pytest.mark.asyncio
356async def test_gen_podcast() -> None:
357 service_manager = MagicMock()
359 job_id = "gen_podcast"
360 job = StreamCastJob(job_id, service_manager)
361 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
363 job.gen = LMMGeneratorMock()
365 assert len(job.characters) == 0
366 with pytest.raises(ValueError, match="Invalid base64-encoded string"):
367 await job.gen_podcast(pdf_base64="AAAAA")
368 job_status = await job.get_status()
369 assert job_status == JobStatus.FAILED
370 assert len(job.characters) == 0
372 # TODO success case
373 # assert job_status == JobStatus.COMPLETED
374 # assert len(job.characters) == 2
376 del job
377 del service_manager
380@pytest.mark.asyncio
381async def test_gen_podcast_all() -> None:
382 service_manager = MagicMock()
383 job_id = "gen_podcast_all"
384 job = StreamCastJob(job_id, service_manager)
385 try:
386 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
387 job.config["resolution"] = "low"
388 job.config["upscaling"] = True
389 job.config["edit_image"] = True
390 job.config["add_subtitles"] = True
391 job.config["debug_image"] = True
392 job.config["speech_speed"] = 1.2
394 job.gen = LMMGeneratorMock()
396 # Wrong type
397 assert len(job.characters) == 0
398 with pytest.raises(TypeError, match="Expected str for base64_str"):
399 await job.gen_podcast(pdf_base64=b"AAAAA") # type: ignore[arg-type]
400 job_status = await job.get_status()
401 assert job_status == JobStatus.FAILED
402 assert len(job.characters) == 0
404 # Wrong base64
405 assert len(job.characters) == 0
406 with pytest.raises(ValueError, match="Invalid base64-encoded string"):
407 await job.gen_podcast(pdf_base64="AAAAA")
408 job_status = await job.get_status()
409 assert job_status == JobStatus.FAILED
410 assert len(job.characters) == 0
412 # Success case
413 assert len(job.characters) == 0
414 pdf_path = "tests/data/blank.pdf"
415 async with aiofiles.open(pdf_path, "rb") as file:
416 pdf_binary = await file.read()
417 pdf_base64 = binary_to_base64(pdf_binary)
418 await job.gen_podcast(pdf_base64=pdf_base64)
419 job_status = await job.get_status()
420 assert job_status == JobStatus.COMPLETED
421 assert len(job.characters) == 2
423 # Check on triggered requests
424 requests = job.get_requests()
425 assert isinstance(requests, dict)
426 assert requests == {} # This is empty because we are mocking all the requests
427 queued_requests = job.get_queued_requests()
428 assert queued_requests == []
429 finally:
430 del job
431 del service_manager
434@pytest.mark.asyncio
435async def test_gen_podcast_nopdf() -> None:
436 service_manager = MagicMock()
438 job_id = "gen_podcast_nopdf"
439 job = StreamCastJob(job_id, service_manager)
440 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
442 with pytest.raises(ValueError, match="Missing 'pdf_base64' in request"):
443 await job.gen_podcast(pdf_base64=None)
444 job_status = await job.get_status()
445 assert job_status == JobStatus.FAILED
447 del job
448 del service_manager
451@pytest.mark.asyncio
452async def test_gen_scene_single() -> None:
453 service_manager = MagicMock()
454 job_id = "gen_scene_single"
455 job = StreamCastJob(job_id, service_manager)
457 # Video and audio synced
458 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
459 with pytest.raises(Exception, match="Service 'fantasytalking' not found"):
460 await job.gen_scene_single(
461 scene_id=0,
462 audio_path="tests/data/audio_4675.wav",
463 image=Image.new("RGB", (100, 100), color="white"),
464 video_prompt="Video prompt.")
466 # Audio only
467 job.config["output_mode"] = OutputMode.AUDIO_ONLY
468 with pytest.raises(Exception, match="Service 'hunyuanframepackf1' not found"):
469 await job.gen_scene_single(
470 scene_id=1,
471 audio_path="tests/data/audio_4675.wav",
472 image=Image.new("RGB", (80, 80), color="blue"),
473 video_prompt="Video prompt.")
475 # Video and audio unsynced
476 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_UNSYNCED
477 with pytest.raises(Exception, match="Service 'hunyuanframepackf1' not found"):
478 await job.gen_scene_single(
479 scene_id=2,
480 audio_path="tests/data/audio_4675.wav",
481 image=Image.new("RGB", (100, 80), color="blue"),
482 video_prompt="Video prompt.")
484 del job
485 del service_manager
488def test_split_text_lines() -> None:
489 text = "This is a test of the split_text_lines function."
490 lines = split_text_lines(text, max_line_length=10)
491 assert isinstance(lines, list)
492 assert len(lines) == 5
494 text = "Short"
495 lines = split_text_lines(text, max_line_length=10)
496 assert isinstance(lines, list)
497 assert len(lines) == 1
498 assert lines[0] == "Short"
500 lines = split_text_lines("", max_line_length=10)
501 assert lines == []
504def assert_approx(
505 a: float,
506 b: float,
507 tol: float = 1e-3
508) -> None:
509 assert abs(a - b) < tol, f"{a} !~= {b}"
512@pytest.mark.asyncio
513async def test_align_audio() -> None:
514 service_manager = MagicMock()
515 job = StreamCastJob("test_video_audio", service_manager)
516 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
518 with pytest.raises(FileNotFoundError):
519 await job.align_audio("notexisting.wav")
521 audio_path = "tests/data/audio_4675.wav"
522 aligned_audio_path, aligned_audio_duration = await job.align_audio(audio_path)
523 assert aligned_audio_path.endswith(".wav")
524 assert_approx(aligned_audio_duration, 3.522)
526 aligned_audio_path, aligned_audio_duration = await job.align_audio(aligned_audio_path)
527 assert aligned_audio_path.endswith(".wav")
528 assert_approx(aligned_audio_duration, 3.522)
531@pytest.mark.asyncio
532async def test_video_audio() -> None:
533 """
534 1. Take an audio file.
535 2. Align it to the FPS (mimic StreamCast).
536 3. Create N frames for that audio (mimic Fantasy Talking).
537 4. Save the frames as video (mimic Fantasy Talking)
538 5. Extract frames from the video (mimic StreamCast).
539 6. Add debugging message to the frames (mimic StreamCast).
540 7. Save the video with audio (mimic StreamCast).
541 """
543 # Mock StreamCast
544 service_manager = MagicMock()
545 job = StreamCastJob("test_video_audio", service_manager)
546 job.config["output_mode"] = OutputMode.VIDEO_AUDIO_SYNCED
547 job.width = 1280
548 job.height = 800
550 # Mock Fantasy Talking
551 fantasy_talking = FantasyTalking()
553 # Verify the input file
554 audio_path = "tests/data/audio_4675.wav"
555 assert os.path.exists(audio_path) is True
556 audio_base64 = await read_file_base64(audio_path)
557 duration_secs = get_audio_duration(audio_base64)
558 assert duration_secs == 4.675
560 # Align to the FPS
561 audio_path_copy = "tests/data/audio_4675.wav"
562 await save_base64_as_binary(audio_path_copy, audio_base64)
563 aligned_audio_path, aligned_audio_duration = await job.align_audio(audio_path)
564 assert aligned_audio_path.endswith(".wav")
565 assert_approx(aligned_audio_duration, 3.522)
567 # Mimic Fantasy Talking preparing the audio
568 ft_audio_path = aligned_audio_path
569 ft_audio_duration, audio_num_frames, video_num_frames = fantasy_talking._get_audio_num_frames(ft_audio_path)
570 assert_approx(ft_audio_duration, 3.522)
571 assert audio_num_frames == 81
572 assert video_num_frames == 81
574 # Mimic Fantasy Talking generating N frames
575 video_frames = [
576 Image.new("RGB", (job.width, job.height), color="blue")
577 for _ in range(video_num_frames)
578 ]
579 assert len(video_frames) == 81
581 # Fantasy Talking save video
582 ft_video_binary = await fantasy_talking._output_video(
583 job_id="test_fantasy_talking",
584 gen_timer=MagicMock(),
585 audio_path=ft_audio_path,
586 video_frames=video_frames,
587 output_type="video_binary")
588 assert ft_video_binary is not None
589 assert len(ft_video_binary) > 1024 # 1 KB at least
591 # StreamCast extract frames from the video
592 video_file_info = get_video_file_info(ft_video_binary)
593 video_info = video_file_info["video"]
594 video_fps = video_info["fps"]
595 assert video_fps == 23
596 video_duration_secs = video_info["duration_seconds"]
597 assert video_duration_secs is not None
598 assert_approx(video_duration_secs, 3.522, tol=0.01)
599 video_num_frames = video_info["num_frames"]
600 assert video_num_frames == 81
601 video_frames = await get_video_frames(ft_video_binary)
602 video_num_frames = len(video_frames)
603 # TODO this breaks in the CI with (ffmpeg version?): 3.52 * 23 = 80.96 != 81
604 # assert_approx(video_duration_secs * video_fps, video_num_frames, tol=1.0 / video_fps)
605 assert video_num_frames == 81, f"{video_num_frames} ({video_duration_secs}x{video_fps}) != 81"
607 # Cleanup
608 await job.close()
609 del service_manager