Coverage for tests/streamwise_app/test_streamdub.py: 100%
389 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 StreamDub.
4"""
6import base64
7import json
8import os
9import sys
10import pytest
12from dataclasses import asdict
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.streamwise_app.app_test_helpers import check_app_root
27from tests.streamwise_app.app_test_helpers import check_health
28from tests.streamwise_app.app_test_helpers import check_files
29from tests.streamwise_app.app_test_helpers import check_unknown_route
30from tests.streamwise_app.app_test_helpers import check_job_submit_page
31from tests.streamwise_app.app_test_helpers import check_job_status_page
32from tests.streamwise_app.app_test_helpers import check_api_job_status
33from tests.streamwise_app.app_test_helpers import check_api_job_requests
35mock_torch = TorchMock()
37mock_modules = {}
38mock_modules.update(mock_torch.get_sub_modules())
40with patch.dict(sys.modules, mock_modules):
41 with temp_sys_path("apps", "apps/streamdub"):
42 from apps.streamdub.streamdub import StreamDubApp
44scene_mocks_base = {
45 'scenedetect': MagicMock(),
46 'scenedetect.detectors': MagicMock(),
47 'scenedetect.stats_manager': MagicMock(),
48}
50with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
51 with temp_sys_path("apps", "apps/streamdub"):
52 from apps.streamdub.streamdub_job import StreamDubJob
53 from apps.streamdub.streamdub_job import _is_empty_transcript
56streamdub_app = StreamDubApp()
59@pytest.fixture(name="test_app")
60def _test_app() -> Quart:
61 return streamdub_app.app
64@pytest.mark.asyncio
65async def test_app(test_app: Quart) -> None:
66 """Check that GET / returns 200."""
67 await check_app_root(test_app, "StreamDub")
70@pytest.mark.asyncio
71async def test_health(test_app: Quart) -> None:
72 """Check /health."""
73 await check_health(test_app)
76@pytest.mark.asyncio
77async def test_files(test_app: Quart) -> None:
78 """Check /files endpoint."""
79 await check_files(test_app, "streamdub")
82@pytest.mark.asyncio
83async def test_unknown_route(test_app: Quart) -> None:
84 """Check that an unknown route returns 404."""
85 await check_unknown_route(test_app)
88@pytest.mark.asyncio
89async def test_job_submit_page(test_app: Quart) -> None:
90 """Check the web page for job submission."""
91 await check_job_submit_page(test_app)
94@pytest.mark.asyncio
95async def test_job_status_page(test_app: Quart) -> None:
96 """Check the web page for job status."""
97 await check_job_status_page(test_app)
100@pytest.mark.asyncio
101async def test_api_job_status(test_app: Quart) -> None:
102 """Check the API for job status (returns UNKNOWN for nonexistent jobs)."""
103 await check_api_job_status(test_app)
106@pytest.mark.asyncio
107async def test_api_job_requests(test_app: Quart) -> None:
108 """Check the API for job requests listing (returns empty for nonexistent jobs)."""
109 await check_api_job_requests(test_app)
112@pytest.mark.asyncio
113async def test_submit_job(test_app: Quart) -> None:
114 """Check the API for job requests."""
115 client = test_app.test_client()
117 response = await client.post("/api/job", json={"video_base64": "AAAA"})
118 assert response.status_code == HTTPStatus.BAD_REQUEST
119 response_json = await response.get_json()
120 assert "error" in response_json
121 assert response_json["error"] == "Service manager not initialized"
123 # Mock the service manager
124 streamdub_app.service_manager = MagicMock()
125 streamdub_app.service_manager.get_service_url = MagicMock(
126 return_value="http://mock_service_url:1234"
127 )
129 response = await client.post("/api/job", json={"video_base64": "AAAA"})
130 # assert response.status_code == HTTPStatus.OK
131 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
132 response_json = await response.get_json()
133 assert "error" in response_json
134 # assert "job_id" in response_json
135 # assert response_json["status"] == "success"
138@pytest.mark.asyncio
139async def test_streamdub_job_no_video() -> None:
140 """StreamDubJob.gen_dub with missing video raises ValueError."""
141 job = StreamDubJob(
142 job_id="test_no_video",
143 service_manager=MagicMock(),
144 )
145 with pytest.raises(ValueError, match="Missing 'video_base64'"):
146 await job.gen_dub(video_base64=None)
149@pytest.mark.asyncio
150async def test_streamdub_job_generate_no_video() -> None:
151 """StreamDubJob.generate with missing video_base64 raises ValueError."""
152 job = StreamDubJob(
153 job_id="test_generate_no_video",
154 service_manager=MagicMock(),
155 )
156 with pytest.raises(ValueError, match="Missing 'video_base64'"):
157 await job.generate(job_config={})
160@pytest.mark.asyncio
161async def test_streamdub_job_detect_scenes_missing_file() -> None:
162 """StreamDubJob.detect_scenes raises FileNotFoundError when video file is absent."""
163 job = StreamDubJob(
164 job_id="test_detect_no_file",
165 service_manager=MagicMock(),
166 )
167 with pytest.raises(FileNotFoundError):
168 await job.detect_scenes()
171@pytest.mark.asyncio
172async def test_streamdub_job_gen_dub_no_scenes() -> None:
173 """gen_dub with a valid video but no detected scenes raises ValueError."""
174 job = StreamDubJob(
175 job_id="test_gen_dub_no_scenes",
176 service_manager=MagicMock(),
177 )
178 # "AAAA" is valid base64; scenedetect mocks return empty scene list
179 with pytest.raises(ValueError, match="No scenes detected"):
180 await job.gen_dub(video_base64="AAAA")
183@pytest.mark.asyncio
184async def test_gen_dub_writes_scenes_json() -> None:
185 """gen_dub must write scenes.json after chunking audio so the UI can display scenes."""
186 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
187 with temp_sys_path("apps", "apps/streamdub"):
188 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
189 from apps.scene import SceneSegment
191 service_manager = AsyncMock()
192 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
194 job_id = "test_scenes_json"
195 job = _StreamDubJob(job_id=job_id, service_manager=service_manager)
197 # Build two fake scenes (audio_path not yet set — chunk_audio_into_scenes sets it)
198 fake_scenes = [
199 SceneSegment(scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0),
200 SceneSegment(scene_id=1, start_frame=30, end_frame=60, start_sec=1.0, end_sec=2.0),
201 ]
203 # Patch detect_scenes and chunk_audio_into_scenes so we control the scene list
204 async def fake_detect_scenes(**kwargs: object) -> list:
205 return fake_scenes
207 async def fake_chunk_audio(**kwargs: object) -> list:
208 # Simulate chunking setting audio_path on each scene (mirrors real behaviour)
209 for scene in fake_scenes:
210 scene.audio_path = f"scene_{scene.scene_id:03d}.wav"
211 return []
213 with patch.object(job, "detect_scenes", side_effect=fake_detect_scenes), \
214 patch.object(job, "chunk_audio_into_scenes", side_effect=fake_chunk_audio), \
215 patch.object(job, "gen_dub_scene", side_effect=ValueError("stop")):
216 try:
217 await job.gen_dub(video_base64="AAAA")
218 except (ValueError, Exception):
219 pass # We only care that scenes.json was written before gen_dub_scene is called
221 scenes_json_path = os.path.join(job.job_path, "scenes.json")
222 assert os.path.exists(scenes_json_path), "scenes.json must be written after chunk_audio_into_scenes"
223 with open(scenes_json_path) as f:
224 data = json.load(f)
225 assert len(data) == 2
226 assert data[0]["scene_id"] == 0
227 assert data[0]["audio_path"] == "scene_000.wav"
228 assert data[1]["scene_id"] == 1
229 assert data[1]["audio_path"] == "scene_001.wav"
231 await job.close()
234@pytest.mark.asyncio
235async def test_gen_dub_scene_uses_voice_cloning() -> None:
236 """gen_dub_scene must call gen_audio with voice_sample when original scene audio is available."""
237 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
238 with temp_sys_path("apps", "apps/streamdub"):
239 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
240 from apps.scene import SceneSegment
242 service_manager = AsyncMock()
243 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
245 job_id = "test_voice_clone"
246 job = _StreamDubJob(job_id=job_id, service_manager=service_manager, config={"add_subtitles": False})
248 # Write a dummy original scene audio file that the job should read and forward
249 original_audio_content = b"RIFF....WAVEfmt " # minimal dummy WAV bytes
250 original_audio_b64 = base64.b64encode(original_audio_content).decode()
251 scene_audio_path = os.path.join(job.job_path, "scene_000.wav")
252 os.makedirs(job.job_path, exist_ok=True)
253 with open(scene_audio_path, "wb") as f:
254 f.write(original_audio_content)
256 # Create a scene with transcript already set (skip transcription / translation)
257 fake_scene = SceneSegment(
258 scene_id=0,
259 start_frame=0,
260 end_frame=30,
261 start_sec=0.0,
262 end_sec=1.0,
263 )
264 fake_scene.audio_path = "scene_000.wav"
265 fake_scene.transcript = "Hola mundo" # pre-translated text
267 dubbed_audio_b64 = base64.b64encode(b"dubbed_audio").decode()
268 dubbed_video_binary = b"dubbed_video"
270 gen_audio_mock = AsyncMock(return_value=dubbed_audio_b64)
271 gen_video_mock = AsyncMock(return_value=dubbed_video_binary)
273 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello world")), \
274 patch.object(job, "translate_scene", new=AsyncMock(return_value="Hola mundo")), \
275 patch.object(job.gen, "gen_audio", gen_audio_mock), \
276 patch.object(job, "gen_video_lip_synced", gen_video_mock):
277 await job.gen_dub_scene(fake_scene, lang_code="e")
279 # Verify that gen_audio was called with voice_sample set to the original audio base64
280 gen_audio_mock.assert_called_once()
281 call_kwargs = gen_audio_mock.call_args.kwargs
282 assert call_kwargs.get("voice_sample") == original_audio_b64, (
283 "voice_sample must equal the base64-encoded original scene audio"
284 )
285 assert call_kwargs.get("text") == "Hola mundo", (
286 "gen_audio must receive the translated text"
287 )
289 await job.close()
292@pytest.mark.asyncio
293async def test_gen_dub_scene_falls_back_when_no_audio() -> None:
294 """gen_dub_scene calls gen_audio without voice_sample when the original scene audio is missing."""
295 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
296 with temp_sys_path("apps", "apps/streamdub"):
297 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
298 from apps.scene import SceneSegment
300 service_manager = AsyncMock()
301 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
303 job = _StreamDubJob(job_id="test_fallback", service_manager=service_manager, config={"add_subtitles": False})
305 # Do NOT write a scene audio file — simulate missing audio
306 fake_scene = SceneSegment(
307 scene_id=0,
308 start_frame=0,
309 end_frame=30,
310 start_sec=0.0,
311 end_sec=1.0,
312 )
313 fake_scene.audio_path = "scene_000.wav"
314 fake_scene.transcript = "Hola mundo"
316 dubbed_audio_b64 = base64.b64encode(b"dubbed_audio").decode()
317 dubbed_video_binary = b"dubbed_video"
319 gen_audio_mock = AsyncMock(return_value=dubbed_audio_b64)
320 gen_video_mock = AsyncMock(return_value=dubbed_video_binary)
322 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello world")), \
323 patch.object(job, "translate_scene", new=AsyncMock(return_value="Hola mundo")), \
324 patch.object(job.gen, "gen_audio", gen_audio_mock), \
325 patch.object(job, "gen_video_lip_synced", gen_video_mock):
326 await job.gen_dub_scene(fake_scene, lang_code="e")
328 # gen_audio should be called without voice_sample (no audio file → standard TTS)
329 gen_audio_mock.assert_called_once()
330 assert gen_audio_mock.call_args.kwargs.get("voice_sample") is None
332 await job.close()
335@pytest.mark.asyncio
336async def test_gen_dub_scene_updates_scenes_json() -> None:
337 """gen_dub_scene must update scenes.json with transcript and translation
338 so the UI stops showing '⏳ Transcribing…' / '⏳ Translating…' during processing."""
339 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
340 with temp_sys_path("apps", "apps/streamdub"):
341 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
342 from apps.scene import SceneSegment
344 service_manager = AsyncMock()
345 job = _StreamDubJob(job_id="test_scenes_json_update", service_manager=service_manager)
346 os.makedirs(job.job_path, exist_ok=True)
348 scene = SceneSegment(
349 scene_id=2, start_frame=60, end_frame=90,
350 start_sec=2.0, end_sec=3.0,
351 audio_path="scene_002.wav",
352 )
353 job.scenes = [scene]
355 # Write initial scenes.json without transcript/translation (mirrors gen_dub behaviour)
356 scenes_path = os.path.join(job.job_path, "scenes.json")
357 with open(scenes_path, "w") as scenes_file:
358 json.dump([asdict(scene)], scenes_file)
360 with open(os.path.join(job.job_path, "scene_002.wav"), "wb") as audio_file:
361 audio_file.write(b"\x00" * 16)
363 original_transcript = "Hello, how are you?"
364 translated_text = "Hola, ¿cómo estás?"
366 # Capture scenes.json state when translate_scene is called — transcript must already be present
367 scenes_json_at_translation_time: list = []
369 async def fake_transcribe(s: object) -> str:
370 return original_transcript
372 async def fake_translate(s: object, **kwargs: object) -> str:
373 with open(scenes_path) as f:
374 scenes_json_at_translation_time.extend(json.load(f))
375 return translated_text
377 async def fake_lip_sync(s: object) -> bytes:
378 return b"video_bytes"
380 with patch.object(job, "transcribe_audio", side_effect=fake_transcribe), \
381 patch.object(job, "translate_scene", side_effect=fake_translate), \
382 patch.object(job, "save_status", new=AsyncMock()), \
383 patch.object(job, "gen_video_lip_synced", side_effect=fake_lip_sync), \
384 patch.object(job, "_add_subtitles_to_video", new=AsyncMock(return_value=b"video_bytes")), \
385 patch.object(job, "get_submission_time", return_value=0.0):
386 job.gen = MagicMock()
387 job.gen.gen_audio = AsyncMock(return_value="AAAA")
388 job.gen.stop = AsyncMock()
389 await job.gen_dub_scene(scene, lang_code="e")
391 # scenes.json must contain the transcript by the time translation is requested
392 assert len(scenes_json_at_translation_time) == 1
393 assert scenes_json_at_translation_time[0]["transcript"] == original_transcript, \
394 "scenes.json must be updated with transcript before translation begins"
396 # scenes.json must contain the translation after gen_dub_scene completes
397 with open(scenes_path) as f:
398 final_scenes = json.load(f)
399 assert final_scenes[0]["transcript"] == original_transcript
400 assert final_scenes[0]["translation"] == translated_text, \
401 "scenes.json must be updated with translation after gen_dub_scene completes"
403 await job.close()
406def test_scene_segment_has_translation_field() -> None:
407 """SceneSegment must have a 'translation' field separate from 'transcript'."""
408 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
409 with temp_sys_path("apps", "apps/streamdub"):
410 from apps.scene import SceneSegment
412 scene = SceneSegment(scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0)
413 scene.transcript = "Hello, how are you?"
414 scene.translation = "Hola, ¿cómo estás?"
416 d = asdict(scene)
417 assert d["transcript"] == "Hello, how are you?"
418 assert d["translation"] == "Hola, ¿cómo estás?"
419 # Changing translation must not affect transcript
420 scene.translation = "Bonjour"
421 assert scene.transcript == "Hello, how are you?"
424@pytest.mark.asyncio
425async def test_gen_dub_scene_stores_translation_separately() -> None:
426 """gen_dub_scene must store original transcript in scene.transcript and
427 the translation in scene.translation (not overwrite transcript)."""
428 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
429 with temp_sys_path("apps", "apps/streamdub"):
430 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
431 from apps.scene import SceneSegment
433 service_manager = AsyncMock()
434 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
436 job = _StreamDubJob(job_id="test_translation_field", service_manager=service_manager,
437 config={"add_subtitles": False})
439 # Pre-create the job directory so transcript file writes succeed
440 os.makedirs(job.job_path, exist_ok=True)
442 scene = SceneSegment(
443 scene_id=0, start_frame=0, end_frame=30,
444 start_sec=0.0, end_sec=1.0,
445 audio_path="scene_000.wav",
446 )
448 # Create a dummy audio file so voice cloning has a sample to read
449 with open(os.path.join(job.job_path, "scene_000.wav"), "wb") as f:
450 f.write(b"\x00" * 16)
452 original_transcript = "Hello, how are you?"
453 translated_text = "Hola, ¿cómo estás?"
455 async def fake_transcribe(s: object) -> str:
456 return original_transcript
458 async def fake_translate(s: object, **kwargs: object) -> str:
459 return translated_text
461 async def fake_lip_sync(s: object) -> bytes:
462 return b"video_bytes"
464 with patch.object(job, "transcribe_audio", side_effect=fake_transcribe), \
465 patch.object(job, "translate_scene", side_effect=fake_translate), \
466 patch.object(job, "save_status", new=AsyncMock()), \
467 patch.object(job, "gen_video_lip_synced", side_effect=fake_lip_sync), \
468 patch.object(job, "get_submission_time", return_value=0.0):
469 job.gen = MagicMock()
470 job.gen.gen_audio = AsyncMock(return_value="AAAA")
471 job.gen.stop = AsyncMock()
472 result = await job.gen_dub_scene(scene, lang_code="e")
474 # Transcript must remain the original transcription
475 assert scene.transcript == original_transcript, \
476 "scene.transcript must hold the original transcription, not the translation"
477 # Translation must be stored in the dedicated field
478 assert scene.translation == translated_text, \
479 "scene.translation must hold the translated text"
480 # gen_audio must have been called with the translation text, not the original
481 job.gen.gen_audio.assert_called_once()
482 call_kwargs = job.gen.gen_audio.call_args
483 tts_text = call_kwargs.kwargs["text"]
484 assert tts_text == translated_text, \
485 "gen_audio must receive the translation, not the original transcript"
486 assert result == b"video_bytes"
488 await job.close()
491@pytest.mark.asyncio
492async def test_gen_dub_scene_empty_translation_uses_original_video() -> None:
493 """gen_dub_scene must use original video (no lip sync) when translation is empty."""
494 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
495 with temp_sys_path("apps", "apps/streamdub"):
496 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
497 from apps.scene import SceneSegment
499 service_manager = AsyncMock()
500 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
502 job = _StreamDubJob(
503 job_id="test_empty_translation",
504 service_manager=service_manager,
505 config={"add_subtitles": False},
506 )
507 os.makedirs(job.job_path, exist_ok=True)
509 fake_scene = SceneSegment(
510 scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0,
511 audio_path="scene_000.wav",
512 )
514 original_video = b"original_video_bytes"
515 get_video_mock = AsyncMock(return_value=original_video)
516 gen_audio_mock = AsyncMock()
517 gen_video_mock = AsyncMock()
519 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello")), \
520 patch.object(job, "translate_scene", new=AsyncMock(return_value="")), \
521 patch.object(job, "get_video_scene", get_video_mock), \
522 patch.object(job.gen, "gen_audio", gen_audio_mock), \
523 patch.object(job, "gen_video_lip_synced", gen_video_mock):
524 result = await job.gen_dub_scene(fake_scene, lang_code="e")
526 # Original video must be returned; no audio generation or lip sync should happen
527 get_video_mock.assert_called_once()
528 gen_audio_mock.assert_not_called()
529 gen_video_mock.assert_not_called()
530 assert result == original_video
532 await job.close()
535@pytest.mark.asyncio
536async def test_gen_dub_scene_adds_subtitles() -> None:
537 """gen_dub_scene must call _add_subtitles_to_video when add_subtitles is True (default)."""
538 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
539 with temp_sys_path("apps", "apps/streamdub"):
540 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
541 from apps.scene import SceneSegment
543 service_manager = AsyncMock()
544 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
546 # Default config: add_subtitles defaults to True
547 job = _StreamDubJob(job_id="test_subtitles", service_manager=service_manager)
548 os.makedirs(job.job_path, exist_ok=True)
550 # Write a dummy audio file for voice cloning
551 with open(os.path.join(job.job_path, "scene_000.wav"), "wb") as f:
552 f.write(b"\x00" * 16)
554 fake_scene = SceneSegment(
555 scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0,
556 audio_path="scene_000.wav",
557 )
559 dubbed_video = b"dubbed_video_bytes"
560 subtitled_video = b"subtitled_video_bytes"
561 subtitles_mock = AsyncMock(return_value=subtitled_video)
563 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello")), \
564 patch.object(job, "translate_scene", new=AsyncMock(return_value="Hola")), \
565 patch.object(job.gen, "gen_audio", AsyncMock(return_value=base64.b64encode(b"audio").decode())), \
566 patch.object(job, "gen_video_lip_synced", AsyncMock(return_value=dubbed_video)), \
567 patch.object(job, "_add_subtitles_to_video", subtitles_mock):
568 result = await job.gen_dub_scene(fake_scene, lang_code="e")
570 # _add_subtitles_to_video must be called with the scene and dubbed video
571 subtitles_mock.assert_called_once()
572 call_args = subtitles_mock.call_args
573 assert call_args.args[0] is fake_scene
574 assert call_args.args[1] == dubbed_video
575 assert result == subtitled_video
577 await job.close()
580@pytest.mark.asyncio
581async def test_gen_dub_scene_voice_cloning_disabled() -> None:
582 """gen_dub_scene uses gen_audio when voice_cloning is disabled in config."""
583 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
584 with temp_sys_path("apps", "apps/streamdub"):
585 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
586 from apps.scene import SceneSegment
588 service_manager = AsyncMock()
589 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
591 job = _StreamDubJob(
592 job_id="test_no_clone",
593 service_manager=service_manager,
594 config={"add_subtitles": False, "voice_cloning": False},
595 )
596 os.makedirs(job.job_path, exist_ok=True)
598 # Write a scene audio file — voice cloning is disabled so it must NOT be used
599 with open(os.path.join(job.job_path, "scene_000.wav"), "wb") as f:
600 f.write(b"\x00" * 16)
602 fake_scene = SceneSegment(
603 scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0,
604 audio_path="scene_000.wav",
605 )
607 dubbed_audio_b64 = base64.b64encode(b"dubbed_audio").decode()
608 gen_audio_mock = AsyncMock(return_value=dubbed_audio_b64)
610 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello")), \
611 patch.object(job, "translate_scene", new=AsyncMock(return_value="Hola")), \
612 patch.object(job.gen, "gen_audio", gen_audio_mock), \
613 patch.object(job, "gen_video_lip_synced", AsyncMock(return_value=b"video")):
614 await job.gen_dub_scene(fake_scene, lang_code="e")
616 # gen_audio must be called without voice_sample (voice cloning disabled → standard TTS)
617 gen_audio_mock.assert_called_once()
618 assert gen_audio_mock.call_args.kwargs.get("voice_sample") is None
620 await job.close()
623@pytest.mark.asyncio
624async def test_gen_dub_scene_voice_cloning_falls_back_on_error() -> None:
625 """gen_dub_scene falls back to standard TTS (no voice_sample) when voice cloning fails."""
626 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
627 with temp_sys_path("apps", "apps/streamdub"):
628 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
629 from apps.scene import SceneSegment
631 service_manager = AsyncMock()
632 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
634 job = _StreamDubJob(
635 job_id="test_clone_fallback",
636 service_manager=service_manager,
637 config={"add_subtitles": False},
638 )
639 os.makedirs(job.job_path, exist_ok=True)
641 # Write a scene audio file so voice cloning is attempted
642 with open(os.path.join(job.job_path, "scene_000.wav"), "wb") as f:
643 f.write(b"\x00" * 16)
645 fake_scene = SceneSegment(
646 scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0,
647 audio_path="scene_000.wav",
648 )
650 fallback_audio_b64 = base64.b64encode(b"fallback_audio").decode()
652 # First call (with voice_sample → VibeVoice) raises; second call (no voice_sample → kokoro) succeeds
653 call_count = 0
655 async def gen_audio_side_effect(**kwargs: object) -> str:
656 nonlocal call_count
657 call_count += 1
658 if call_count == 1:
659 raise RuntimeError("VibeVoice unavailable")
660 return fallback_audio_b64
662 gen_audio_mock = AsyncMock(side_effect=gen_audio_side_effect)
664 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value="Hello")), \
665 patch.object(job, "translate_scene", new=AsyncMock(return_value="Hola")), \
666 patch.object(job.gen, "gen_audio", gen_audio_mock), \
667 patch.object(job, "gen_video_lip_synced", AsyncMock(return_value=b"video")):
668 await job.gen_dub_scene(fake_scene, lang_code="e")
670 # gen_audio called twice: first with voice_sample (→ VibeVoice, fails), then without (→ kokoro)
671 assert gen_audio_mock.call_count == 2
672 first_call_kwargs = gen_audio_mock.call_args_list[0].kwargs
673 second_call_kwargs = gen_audio_mock.call_args_list[1].kwargs
674 assert first_call_kwargs.get("voice_sample") is not None, "first call must include voice_sample"
675 assert second_call_kwargs.get("voice_sample") is None, "fallback call must NOT include voice_sample"
677 await job.close()
680# ---------------------------------------------------------------------------
681# _is_empty_transcript unit tests
682# ---------------------------------------------------------------------------
684@pytest.mark.parametrize("text", [
685 "",
686 "-",
687 "- -",
688 "♪ ♪",
689 " ",
690 "---",
691 "♪",
692 " - - ",
693])
694def test_is_empty_transcript_true(text: str) -> None:
695 """_is_empty_transcript must return True for texts with no word characters."""
696 assert _is_empty_transcript(text), f"Expected True for {text!r}"
699@pytest.mark.parametrize("text", [
700 "Hello",
701 "Hello world",
702 "- hello -",
703 "♪ La la la ♪",
704 "1",
705 "ok",
706])
707def test_is_empty_transcript_false(text: str) -> None:
708 """_is_empty_transcript must return False when the text contains word characters."""
709 assert not _is_empty_transcript(text), f"Expected False for {text!r}"
712# ---------------------------------------------------------------------------
713# gen_dub_scene skips dubbing for empty-like transcripts
714# ---------------------------------------------------------------------------
716@pytest.mark.asyncio
717@pytest.mark.parametrize("transcript", [
718 "-",
719 "- -",
720 "♪ ♪",
721 " ",
722])
723async def test_gen_dub_scene_skips_empty_like_transcript(transcript: str) -> None:
724 """gen_dub_scene must skip translation and use original video for non-word transcripts."""
725 with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
726 with temp_sys_path("apps", "apps/streamdub"):
727 from apps.streamdub.streamdub_job import StreamDubJob as _StreamDubJob
728 from apps.scene import SceneSegment
730 service_manager = AsyncMock()
731 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
733 job = _StreamDubJob(
734 job_id=f"test_empty_transcript_{hash(transcript) & 0xFFFF}",
735 service_manager=service_manager,
736 config={"add_subtitles": False},
737 )
738 os.makedirs(job.job_path, exist_ok=True)
740 fake_scene = SceneSegment(
741 scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0,
742 audio_path="scene_000.wav",
743 )
745 original_video = b"original_video_bytes"
746 get_video_mock = AsyncMock(return_value=original_video)
747 translate_mock = AsyncMock()
748 gen_audio_mock = AsyncMock()
749 gen_video_mock = AsyncMock()
751 with patch.object(job, "transcribe_audio", new=AsyncMock(return_value=transcript)), \
752 patch.object(job, "translate_scene", translate_mock), \
753 patch.object(job, "get_video_scene", get_video_mock), \
754 patch.object(job.gen, "gen_audio", gen_audio_mock), \
755 patch.object(job, "gen_video_lip_synced", gen_video_mock):
756 result = await job.gen_dub_scene(fake_scene, lang_code="e")
758 # Must return original video without calling translation, audio gen, or lip sync
759 get_video_mock.assert_called_once()
760 translate_mock.assert_not_called()
761 gen_audio_mock.assert_not_called()
762 gen_video_mock.assert_not_called()
763 assert result == original_video
765 await job.close()