"""AI 数字人口型视频生成速度优化 — 单元测试. 验证两个优化点： 1. FFmpeg 编码 preset 从 fast 改为 veryfast（提速 30~50%） 2. TTS 合成从同步改为 Celery 异步任务（API 响应从 6~35s 降到 <1s） Issue: lipsync-speed-optimization """ import os from unittest.mock import MagicMock, patch import pytest os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing") # ═══════════════════════════════════════════════════════════════════════════════ # 优化1: FFmpeg 编码提速 — preset veryfast # ═══════════════════════════════════════════════════════════════════════════════ class TestFFmpegPresetOptimization: """验证 FFmpeg 编码命令从 -preset fast 改为 -preset veryfast.""" def test_preset_is_veryfast(self): """_build_ffmpeg_command 输出必须包含 -preset veryfast.""" from app.services.ai_avatar_render_service import AiAvatarRenderService svc = AiAvatarRenderService.__new__(AiAvatarRenderService) cmd = svc._build_ffmpeg_command( input_video="https://example.com/video.mp4", b_roll_segments=[], filter_complex="", final_label=None, output_path="/tmp/output.mp4", ) assert "-preset veryfast" in cmd, f"期望 -preset veryfast，实际命令: {cmd}" def test_preset_veryfast_with_filter(self): """带滤镜场景下也必须使用 veryfast.""" from app.services.ai_avatar_render_service import AiAvatarRenderService svc = AiAvatarRenderService.__new__(AiAvatarRenderService) cmd = svc._build_ffmpeg_command( input_video="https://example.com/video.mp4", b_roll_segments=[], filter_complex="overlay=0:0", final_label="[v]", output_path="/tmp/output.mp4", ) assert "-preset veryfast" in cmd assert "-filter_complex" in cmd def test_preset_not_fast(self): """确保不再使用旧的 -preset fast.""" from app.services.ai_avatar_render_service import AiAvatarRenderService svc = AiAvatarRenderService.__new__(AiAvatarRenderService) cmd = svc._build_ffmpeg_command( input_video="https://example.com/video.mp4", b_roll_segments=[], filter_complex="", final_label=None, output_path="/tmp/output.mp4", ) # 确保是 veryfast 而不是 fast assert "-preset veryfast" in cmd # 排除 "fast" 单独出现（veryfast 包含 fast 子串，需精确判断） parts = cmd.split() preset_idx = parts.index("-preset") assert parts[preset_idx + 1] == "veryfast" # ═══════════════════════════════════════════════════════════════════════════════ # 优化2: TTS 合成 Celery 异步化 # ═══════════════════════════════════════════════════════════════════════════════ def _make_service_with_mocks(): """构造 LipsyncService 测试实例及 mock 依赖.""" from app.services.lipsync_service import LipsyncService db = MagicMock() client = MagicMock() client.is_available = True client.submit_lipsync.return_value = { "success": True, "task_id": "mk-1", "request_id": "req-1", } cosy = MagicMock() cosy.submit_synthesize_task.return_value = { "audio_url": "https://tts/raw.mp3", "request_id": "tts-req", "audio_duration": 3.0, } svc = LipsyncService(db, client=client, cosyvoice_service=cosy, voice_clone_repo=MagicMock()) # _resolve_voice_id 默认原样返回（repo.get 返回 None） svc._voice_clone_repo.get.return_value = None return svc, client, cosy class TestCreateJobAsyncTTS: """验证 TTS 模式改为 Celery 异步后的行为.""" def test_tts_mode_returns_tts_processing_status(self): """TTS 模式下 create_job 立即返回，状态为 tts_processing.""" svc, client, cosy = _make_service_with_mocks() with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async = MagicMock() job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", voice_id="longxiaochun_v3", script_text="大家好", speed=1.0, emotion="", ) assert job.status == "tts_processing" def test_tts_mode_dispatches_celery_task(self): """TTS 模式必须 dispatch Celery 异步任务.""" svc, client, cosy = _make_service_with_mocks() with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async = MagicMock() svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", voice_id="v-1", script_text="测试文本", ) mock_task.apply_async.assert_called_once() call_kwargs = mock_task.apply_async.call_args args = call_kwargs.kwargs.get("args") or call_kwargs[1].get("args", call_kwargs[0][0] if call_kwargs[0] else ()) assert args[1] == "user-1" # user_id assert args[2] == "v-1" # voice_id assert args[3] == "测试文本" # script_text def test_tts_mode_celery_dispatch_failure_still_creates_job(self): """Celery dispatch 失败时，job 记录已创建，状态保持 tts_processing.""" svc, client, cosy = _make_service_with_mocks() with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async = MagicMock(side_effect=Exception("Celery broker down")) job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", voice_id="v-1", script_text="测试文本", ) # job 已创建 assert job is not None assert job.status == "tts_processing" # MediaKit 未被调用 client.submit_lipsync.assert_not_called() def test_tts_mode_voice_validation_still_sync(self): """TTS 模式下音色校验仍在 HTTP 请求中同步执行.""" from app.services.mediakit_client import MediaKitError svc, client, cosy = _make_service_with_mocks() # 模拟音色属于其他用户 other_profile = MagicMock() other_profile.user_id = "user-other" svc._voice_clone_repo.get.return_value = other_profile with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit"): with pytest.raises(MediaKitError) as exc: svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", voice_id="clone-profile-id", script_text="测试", ) assert exc.value.code == "VoiceForbidden" def test_tts_mode_missing_input_raises_immediately(self): """缺少 voice_id 或 script_text 时立即报错，不 dispatch Celery 任务.""" from app.services.mediakit_client import MediaKitError svc, client, cosy = _make_service_with_mocks() with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task: mock_task.delay = MagicMock() with pytest.raises(MediaKitError) as exc: svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", # 缺少 voice_id 和 script_text ) assert exc.value.code == "InvalidInput" # Celery 任务未被 dispatch mock_task.delay.assert_not_called() # TTS 和 MediaKit 均未调用 cosy.submit_synthesize_task.assert_not_called() client.submit_lipsync.assert_not_called() class TestCreateJobDirectAudio: """验证直接音频模式不受异步化影响.""" def test_direct_audio_still_submits_synchronously(self): """直接音频模式仍然同步提交 MediaKit，状态为 submitted.""" svc, client, cosy = _make_service_with_mocks() with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task: mock_task.delay = MagicMock() job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", audio_url="https://example.com/audio.mp3", ) assert job.status == "submitted" assert job.mediakit_task_id == "mk-1" client.submit_lipsync.assert_called_once() # TTS Celery 任务不应被调用 mock_task.delay.assert_not_called() def test_direct_audio_skips_tts(self): """直接音频模式不调用 CosyVoice TTS.""" svc, client, cosy = _make_service_with_mocks() job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", audio_url="https://example.com/audio.mp3", ) cosy.submit_synthesize_task.assert_not_called() call_kwargs = client.submit_lipsync.call_args assert call_kwargs.kwargs["audio_url"] == "https://example.com/audio.mp3" class TestCancelJobTtsProcessing: """验证 tts_processing 状态的任务可以被取消.""" def test_cancel_tts_processing(self): """tts_processing 状态的任务可以成功取消.""" svc, client, cosy = _make_service_with_mocks() mock_job = MagicMock() mock_job.status = "tts_processing" mock_job.id = "job-1" svc.get_job = MagicMock(return_value=mock_job) result = svc.cancel_job("job-1", "user-1") assert result.status == "cancelled" def test_cancel_pending_still_works(self): """pending 状态仍可取消.""" svc, client, cosy = _make_service_with_mocks() mock_job = MagicMock() mock_job.status = "pending" mock_job.id = "job-1" svc.get_job = MagicMock(return_value=mock_job) result = svc.cancel_job("job-1", "user-1") assert result.status == "cancelled" def test_cancel_submitted_still_works(self): """submitted 状态仍可取消.""" svc, client, cosy = _make_service_with_mocks() mock_job = MagicMock() mock_job.status = "submitted" mock_job.id = "job-1" svc.get_job = MagicMock(return_value=mock_job) result = svc.cancel_job("job-1", "user-1") assert result.status == "cancelled"