From a199fb0fc4c458b2a76c04f8ae186ac2dcec53ea Mon Sep 17 00:00:00 2001 From: deploy-agent Date: Wed, 9 Sep 2026 23:19:19 +0800 Subject: [PATCH] =?UTF-8?q?test(lipsync):=20=E9=87=87=E7=94=A8=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=E8=A7=A3=E5=86=B2=E7=AA=81=E7=89=88=20test=5Flipsync?= =?UTF-8?q?=5Froutes.py=EF=BC=88v3=20submit=5Fsynthesize=5Ftask=20?= =?UTF-8?q?=E5=A5=91=E7=BA=A6=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 4efa71ec 合并时 tests/unit/test_lipsync_routes.py 误取了 develop #1809 版本 （synthesize_speech/route._resolve_voice_id/断言 audio_url 移除），与合并后 v3 #1822 双模式实现不一致，导致 7 个单测失败。改用后端解冲突分支 backend/ai-avatar-merge-1827 (0e994072) 中已适配 v3 的版本： mock submit_synthesize_task、emotion 兴奋→excited、双模式 audio_url 保留。 本地验证该文件 28 passed、lipsync/avatar/tts 相关 79 passed。 --- tests/unit/test_lipsync_routes.py | 303 +++++++++++++++++++++--------- 1 file changed, 212 insertions(+), 91 deletions(-) diff --git a/tests/unit/test_lipsync_routes.py b/tests/unit/test_lipsync_routes.py index 5b154194a..7aa92ee60 100644 --- a/tests/unit/test_lipsync_routes.py +++ b/tests/unit/test_lipsync_routes.py @@ -35,14 +35,14 @@ def mock_mediakit(): @pytest.fixture def mock_cosyvoice(): - """Mock CosyVoice 服务.""" + """Mock CosyVoice 服务（v3: service 内部走 submit_synthesize_task，返回 dict）.""" service = MagicMock() - service.synthesize_speech.return_value = MagicMock( - audio_url="https://oss.example.com/tts-output.mp3", - duration=15.0, - file_size=12345, - request_id="tts-req-789", - ) + service.submit_synthesize_task.return_value = { + "audio_url": "https://oss.example.com/tts-output.mp3", + "request_id": "tts-req-789", + } + # synthesize_speech 保留给直接同步调用场景 + service.synthesize_speech.return_value = MagicMock(audio_url="https://oss.example.com/tts-output.mp3") return service @@ -172,86 +172,120 @@ class TestSchemaValidation: ) assert "?token=" in req.video_url - def test_no_audio_url_in_request(self): - """#1809: 请求体不应包含 audio_url 字段.""" + def test_dual_mode_fields_present(self): + """v3 契约: 双模式——支持直接音频 audio_url，也支持 TTS 直生 voice_id+script_text.""" + from app.schemas.lipsync import CreateLipsyncJobRequest + + fields = CreateLipsyncJobRequest.model_fields.keys() + # 直接音频模式 + assert "audio_url" in fields + # TTS 直生模式 + assert "voice_id" in fields + assert "script_text" in fields + # 语速/情绪透传 + assert "speed" in fields + assert "emotion" in fields + + def test_direct_audio_mode_accepted(self): + """v3 契约: 只传 audio_url（直接音频模式）也合法，无需 voice_id/script_text.""" from app.schemas.lipsync import CreateLipsyncJobRequest req = CreateLipsyncJobRequest( video_url="https://example.com/video.mp4", - voice_id="longxiaochun_v3", - script_text="测试文本", + audio_url="https://example.com/audio.mp3", ) - assert not hasattr(req, "audio_url") - fields = req.model_fields.keys() - assert "audio_url" not in fields - assert "voice_id" in fields - assert "script_text" in fields + assert req.audio_url == "https://example.com/audio.mp3" + + def test_neither_mode_rejected(self): + """v3 契约: audio_url 与 voice_id+script_text 都缺时应报错.""" + from app.schemas.lipsync import CreateLipsyncJobRequest + + with pytest.raises(ValueError): + CreateLipsyncJobRequest(video_url="https://example.com/video.mp4") class TestLipsyncServiceUnit: """Service 层单元测试（纯 mock，不依赖数据库）— #1809 更新.""" def test_create_job_success(self, mock_mediakit, mock_cosyvoice): + """v3: TTS 直生——service 内部 submit_synthesize_task 合成后转存 OSS，再提交 MediaKit.""" from app.services.lipsync_service import LipsyncService mock_db = MagicMock() - mock_db.add = MagicMock() - mock_db.flush = MagicMock() - mock_db.commit = MagicMock() - mock_db.refresh = MagicMock() + mock_repo = MagicMock() + mock_repo.get.return_value = None # 预置音色，原样返回 voice_id - svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + with ( + patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch, + patch("app.services.lipsync_service.safe_download_bytes") as dl_patch, + ): + storage_patch.return_value.upload_file.return_value = "https://my-oss/tts.mp3" + dl_patch.return_value = b"audio-bytes" - job = svc.create_job( - user_id="user-1", - video_url="https://example.com/video.mp4", - voice_id="longxiaochun_v3", - script_text="大家好，欢迎来到直播间", - ) + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) + + job = svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + voice_id="longxiaochun_v3", + script_text="大家好，欢迎来到直播间", + speed=1.2, + emotion="兴奋", + ) assert job.status == "submitted" assert job.mediakit_task_id == "mk-task-123" - # TTS 应该被调用 - mock_cosyvoice.synthesize_speech.assert_called_once_with( - text="大家好，欢迎来到直播间", - voice_id="longxiaochun_v3", - ) - # MediaKit 应该用 TTS 生成的 audio_url - mock_mediakit.submit_lipsync.assert_called_once() + # TTS 直生走 submit_synthesize_task，带语速/情绪 + mock_cosyvoice.submit_synthesize_task.assert_called_once() + _, kwargs = mock_cosyvoice.submit_synthesize_task.call_args + assert kwargs["text"] == "大家好，欢迎来到直播间" + assert kwargs["voice_id"] == "longxiaochun_v3" + assert kwargs["speed"] == 1.2 + assert kwargs["emotion"] == "excited" # 兴奋→excited + # job 记录透传字段 + assert job.speed == 1.2 + assert job.emotion == "excited" + # MediaKit 用转存后的 OSS audio_url call_kwargs = mock_mediakit.submit_lipsync.call_args - assert call_kwargs.kwargs["audio_url"] == "https://oss.example.com/tts-output.mp3" + assert call_kwargs.kwargs["audio_url"] == "https://my-oss/tts.mp3" def test_create_job_tts_failure(self, mock_mediakit): - """TTS 合成失败时，应创建 failed 记录并抛出 CosyVoiceError.""" + """v3: TTS 合成失败时，CosyVoiceError 被包装为 MediaKitError(TTSSynthesisFailed)， + 在建 DB 记录之前抛出，不提交 MediaKit。""" from app.services.lipsync_service import LipsyncService - + from app.services.mediakit_client import MediaKitError from packages.application.cosyvoice_service import CosyVoiceError mock_cosyvoice = MagicMock() - mock_cosyvoice.synthesize_speech.side_effect = CosyVoiceError("TTS 服务不可用") + mock_cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("Arrearage 欠费") mock_db = MagicMock() - mock_db.add = MagicMock() - mock_db.flush = MagicMock() - mock_db.commit = MagicMock() - mock_db.refresh = MagicMock() + mock_repo = MagicMock() + mock_repo.get.return_value = None - svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) - with pytest.raises(CosyVoiceError, match="TTS 服务不可用"): + with pytest.raises(MediaKitError) as exc_info: svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", voice_id="longxiaochun_v3", script_text="测试文本", ) + assert exc_info.value.code == "TTSSynthesisFailed" # 不应提交到 MediaKit mock_mediakit.submit_lipsync.assert_not_called() - # 应该记录了失败状态 - added_job = mock_db.add.call_args[0][0] - assert added_job.status == "failed" - assert "TTS" in added_job.error_message def test_create_job_api_failure(self, mock_mediakit, mock_cosyvoice): """MediaKit 提交失败.""" @@ -399,78 +433,165 @@ class TestLipsyncServiceUnit: assert result.status == "completed" def test_create_job_stores_tts_audio_url(self, mock_mediakit, mock_cosyvoice): - """#1809: 验证 job 的 audio_url 来自 TTS 合成结果.""" + """v3: TTS 直生模式下 job.audio_url 为转存到自家 OSS 的永久地址.""" from app.services.lipsync_service import LipsyncService mock_db = MagicMock() - mock_db.add = MagicMock() - mock_db.flush = MagicMock() - mock_db.commit = MagicMock() - mock_db.refresh = MagicMock() + mock_repo = MagicMock() + mock_repo.get.return_value = None - svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + with ( + patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch, + patch("app.services.lipsync_service.safe_download_bytes") as dl_patch, + ): + storage_patch.return_value.upload_file.return_value = "https://my-oss/permanent.mp3" + dl_patch.return_value = b"audio-bytes" + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) + job = svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + voice_id="my-clone-voice", + script_text="这是一段测试文本", + ) + + # job.audio_url 是转存 OSS 后的永久地址 + assert job.audio_url == "https://my-oss/permanent.mp3" + + def test_create_job_direct_audio_skips_tts(self, mock_mediakit, mock_cosyvoice): + """v3: 直接音频模式（传 audio_url）不触发 TTS，原样把 audio_url 提交 MediaKit.""" + from app.services.lipsync_service import LipsyncService + + mock_db = MagicMock() + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=MagicMock(), + ) job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", - voice_id="my-clone-voice", - script_text="这是一段测试文本", + audio_url="https://example.com/direct-audio.mp3", ) - # job.audio_url 应该是 TTS 返回的 URL - assert job.audio_url == "https://oss.example.com/tts-output.mp3" + mock_cosyvoice.submit_synthesize_task.assert_not_called() + call_kwargs = mock_mediakit.submit_lipsync.call_args + assert call_kwargs.kwargs["audio_url"] == "https://example.com/direct-audio.mp3" + assert job.audio_url == "https://example.com/direct-audio.mp3" class TestErrorHandling: - """#1809 补充：错误返回 400 而非 500.""" + """v3: 音色解析与错误码在 service 层处理，路由层做 HTTP 状态码映射.""" - def test_voice_id_resolve_failure_returns_400(self, mock_mediakit, mock_cosyvoice): - """voice_clone_repo 查询异常时返回 400 而非 500.""" - from app.api.routes.lipsync import _resolve_voice_id - from fastapi import HTTPException + def test_voice_id_resolve_forbidden(self, mock_mediakit, mock_cosyvoice): + """v3: 克隆音色属于他人时 service._resolve_voice_id 抛 VoiceForbidden（路由映射 403）.""" + from app.services.lipsync_service import LipsyncService + from app.services.mediakit_client import MediaKitError + mock_db = MagicMock() mock_repo = MagicMock() - mock_repo.get.side_effect = Exception("DB connection error") + other_profile = MagicMock() + other_profile.user_id = "user-other" + other_profile.voice_id = "cv-voice-1" + mock_repo.get.return_value = other_profile - with pytest.raises(HTTPException) as exc_info: - _resolve_voice_id("bad-voice-id", "user-1", mock_repo) - assert exc_info.value.status_code == 400 - assert "voice_id" in str(exc_info.value.detail) + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) - def test_create_job_value_error_returns_400(self, mock_mediakit): - """ValueError（参数无效）返回 400 而非 500.""" - from app.services.lipsync_service import LipsyncService - - mock_cosyvoice = MagicMock() - mock_cosyvoice.synthesize_speech.side_effect = ValueError("voice_id 为空") - - mock_db = MagicMock() - svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) - - # Service 层会 catch CosyVoiceError 但 ValueError 会穿透 - # 路由层 catch ValueError → 400 - with pytest.raises(ValueError): + with pytest.raises(MediaKitError) as exc_info: svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", - voice_id="", - script_text="test", + voice_id="clone-profile-id", + script_text="测试", ) + assert exc_info.value.code == "VoiceForbidden" + mock_mediakit.submit_lipsync.assert_not_called() - def test_create_job_unexpected_exception_returns_400(self, mock_mediakit): - """未预期的异常应被路由层捕获返回 400 而非 500.""" + def test_voice_id_resolve_not_ready(self, mock_mediakit, mock_cosyvoice): + """v3: 克隆音色尚未生成 voice_id 时抛 VoiceNotReady（路由映射 400）.""" from app.services.lipsync_service import LipsyncService - - mock_cosyvoice = MagicMock() - mock_cosyvoice.synthesize_speech.side_effect = RuntimeError("unexpected") + from app.services.mediakit_client import MediaKitError mock_db = MagicMock() - svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + mock_repo = MagicMock() + profile = MagicMock() + profile.user_id = "user-1" + profile.voice_id = "" # 克隆未完成 + mock_repo.get.return_value = profile - with pytest.raises(RuntimeError): + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) + + with pytest.raises(MediaKitError) as exc_info: svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", - voice_id="test-voice", + voice_id="clone-profile-id", + script_text="测试", + ) + assert exc_info.value.code == "VoiceNotReady" + + def test_tts_value_error_mapped_to_invalid_param(self, mock_mediakit): + """v3: CosyVoice 抛 ValueError（参数无效）被包装为 TTSInvalidParam（路由映射 400）.""" + from app.services.lipsync_service import LipsyncService + from app.services.mediakit_client import MediaKitError + + mock_cosyvoice = MagicMock() + mock_cosyvoice.submit_synthesize_task.side_effect = ValueError("voice_id 为空") + + mock_db = MagicMock() + mock_repo = MagicMock() + mock_repo.get.return_value = None + + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=mock_repo, + ) + + with pytest.raises(MediaKitError) as exc_info: + svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + voice_id="some-voice", script_text="test", ) + assert exc_info.value.code == "TTSInvalidParam" + + def test_missing_both_inputs_raises_invalid_input(self, mock_mediakit, mock_cosyvoice): + """v3: 既无 audio_url 又无 voice_id+script_text 时抛 InvalidInput（路由映射 400）.""" + from app.services.lipsync_service import LipsyncService + from app.services.mediakit_client import MediaKitError + + mock_db = MagicMock() + svc = LipsyncService( + mock_db, + client=mock_mediakit, + cosyvoice_service=mock_cosyvoice, + voice_clone_repo=MagicMock(), + ) + + with pytest.raises(MediaKitError) as exc_info: + svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + ) + assert exc_info.value.code == "InvalidInput" + mock_cosyvoice.submit_synthesize_task.assert_not_called() + mock_mediakit.submit_lipsync.assert_not_called()