From 1f8bbe37dd39e978accdb762d41f82127146a42e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Sun, 5 Jul 2026 17:05:17 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20TTS=20=E5=90=88=E6=88=90=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=A4=B1=E8=B4=A5=E6=97=B6=E8=BF=94=E5=9B=9E=20201=20?= =?UTF-8?q?+=20failed=20=E7=8A=B6=E6=80=81=EF=BC=8C=E4=B8=8D=E6=8A=9B=2050?= =?UTF-8?q?0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与音色克隆接口保持一致： - 路由层增加 try/except 兜底，捕获 workflow 未处理的意外异常（DB错误等） - CosyVoice 失败时标记 job 为 failed，返回 201 + failed 状态任务记录 - Celery 调度失败时也标记 job 为 failed，不抛 500 - 新增 10 个单元测试覆盖各种失败场景 Co-Authored-By: Claude Fable 5 --- apps/api/app/api/routes/tts.py | 22 +- tests/unit/test_tts_synthesis_failure.py | 258 +++++++++++++++++++++++ 2 files changed, 278 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_tts_synthesis_failure.py diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 82bd42927..34b6f84dd 100644 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from typing import Optional from app.auth import AuthenticatedUser, get_current_user @@ -30,6 +31,8 @@ from packages.application.tts_job.use_cases import ( ) from packages.application.tts_job.workflow import TTSWorkflowService +logger = logging.getLogger(__name__) + router = APIRouter() @@ -74,6 +77,7 @@ def synthesize( """发起 TTS 合成任务。 创建 TTS 任务 → 提交 CosyVoice 合成 → 触发 Celery 异步轮询。 + 与音色克隆接口保持一致：CosyVoice 失败时不抛 500，而是返回 201 + failed 状态任务记录。 """ user_id = authenticated_user.user.id use_case = CreateTTSJobUseCase(repository) @@ -91,7 +95,18 @@ def synthesize( repository=repository, cosyvoice_service=cosyvoice_service, ) - job = workflow.start_synthesis(job.id) + + try: + job = workflow.start_synthesis(job.id) + except Exception as e: + # 兜底：workflow 内部已捕获 CosyVoiceError / ValueError， + # 但 DB 异常、网络异常等意外错误可能逃逸。 + # 与音色克隆接口保持一致：标记 failed，返回 201，不抛 500。 + logger.error(f"TTS 合成异常: job_id={job.id}, error={e}", exc_info=True) + try: + job = workflow.process_synthesis_failure(job.id, str(e)) + except Exception as inner_e: + logger.error(f"标记 TTS job 失败时出错: job_id={job.id}, error={inner_e}") # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询 if job.status.value == "processing": @@ -103,7 +118,10 @@ def synthesize( process_tts_synthesis.delay(job.id) except Exception as e: # Celery 调度失败，标记 job 为 failed - workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}") + try: + workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}") + except Exception as inner_e: + logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}") return TTSSynthesizeResponse( job_id=job.id, diff --git a/tests/unit/test_tts_synthesis_failure.py b/tests/unit/test_tts_synthesis_failure.py new file mode 100644 index 000000000..a21153558 --- /dev/null +++ b/tests/unit/test_tts_synthesis_failure.py @@ -0,0 +1,258 @@ +"""TTS 合成失败场景单元测试。 + +验证 TTS 合成接口在 CosyVoice 失败时返回 201 + failed 状态（与音色克隆一致）， +而不是抛 500 错误。 +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from packages.application.cosyvoice_service import ( + CosyVoiceAuthError, + CosyVoiceError, + CosyVoiceService, + CosyVoiceTimeoutError, +) +from packages.application.tts_job.workflow import TTSWorkflowService +from packages.domain.tts_job import TTSJob, TTSJobStatus + + +def _make_job(**kwargs) -> TTSJob: + defaults = { + "id": "test_job_001", + "user_id": "user_001", + "input_text": "测试文本", + "voice_id": "voice_001", + "voice_model": "", + "project_id": "", + "voice_clone_profile_id": "", + "status": TTSJobStatus.PENDING, + "output_audio_url": "", + "output_audio_key": "", + "duration": 0.0, + "file_size": 0, + "sample_rate": 22050, + "format": "mp3", + "error_message": "", + "retry_count": 0, + "max_retries": 3, + "metadata": {}, + "started_at": None, + "completed_at": None, + "created_at": datetime.now(timezone.utc), + "updated_at": datetime.now(timezone.utc), + } + defaults.update(kwargs) + return TTSJob(**defaults) + + +class TestTTSWorkflowFailureScenarios: + """测试 TTSWorkflowService 在各种失败场景下的行为。 + + 核心要求：CosyVoice 失败时，job 应被标记为 failed，而不是抛异常。 + """ + + def _make_workflow(self, cosyvoice_service: MagicMock, repo: MagicMock | None = None) -> TTSWorkflowService: + if repo is None: + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + return TTSWorkflowService(repository=repo, cosyvoice_service=cosyvoice_service) + + def test_auth_error_marks_job_failed(self) -> None: + """CosyVoiceAuthError（API Key 未配置）→ job 标记为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = CosyVoiceAuthError("CosyVoice API Key 未配置") + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.FAILED + assert "API Key 未配置" in job.error_message + + def test_api_error_marks_job_failed(self) -> None: + """CosyVoiceError（API 调用失败）→ job 标记为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = CosyVoiceError("CosyVoice API 服务端错误: HTTP 500") + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.FAILED + assert "服务端错误" in job.error_message + + def test_timeout_error_marks_job_failed(self) -> None: + """CosyVoiceTimeoutError（超时）→ job 标记为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = CosyVoiceTimeoutError("请求超时") + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.FAILED + assert "超时" in job.error_message + + def test_value_error_marks_job_failed(self) -> None: + """ValueError（参数错误）→ job 标记为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = ValueError("voice_id 不能为空") + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.FAILED + assert "voice_id" in job.error_message + + def test_unexpected_exception_propagates(self) -> None: + """非预期异常（如 DB 错误）→ workflow 不捕获，由路由层兜底。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = RuntimeError("unexpected DB error") + + workflow = self._make_workflow(service) + + with pytest.raises(RuntimeError, match="unexpected DB error"): + workflow.start_synthesis("test_job_001") + + def test_sync_success_marks_job_completed(self) -> None: + """CosyVoice 同步返回 audio_url → job 标记为 completed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "", + "audio_url": "https://example.com/audio.mp3", + "duration": 3.5, + "file_size": 12345, + "request_id": "req_001", + } + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "https://example.com/audio.mp3" + assert job.duration == 3.5 + + def test_async_success_marks_job_processing(self) -> None: + """CosyVoice 异步模式（返回 task_id）→ job 标记为 processing。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "cosy_task_001", + "audio_url": "", + "duration": 0.0, + "file_size": 0, + "request_id": "req_002", + } + + workflow = self._make_workflow(service) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.PROCESSING + assert job.metadata.get("cosyvoice_task_id") == "cosy_task_001" + + def test_process_synthesis_failure(self) -> None: + """process_synthesis_failure 将 job 标记为 failed。""" + repo = MagicMock() + job = _make_job(status=TTSJobStatus.PROCESSING) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + service = MagicMock(spec=CosyVoiceService) + workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service) + + result = workflow.process_synthesis_failure("test_job_001", "Celery 任务调度失败") + + assert result.status == TTSJobStatus.FAILED + assert "Celery 任务调度失败" in result.error_message + + +class TestTTSRouteFailureScenarios: + """测试 TTS 路由层在各种失败场景下的行为。 + + 核心要求：即使 workflow 抛出意外异常，路由也应返回 201 + failed，不抛 500。 + """ + + def test_route_catches_unexpected_exception_from_workflow(self) -> None: + """路由层兜底：workflow 抛出非预期异常时，job 仍标记为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = RuntimeError("unexpected") + + repo = MagicMock() + job = _make_job() + repo.get.return_value = job + repo.update.side_effect = lambda j: j + repo.create.side_effect = lambda j: j + + # 模拟路由逻辑 + from packages.application.tts_job.use_cases import CreateTTSJobUseCase + from packages.application.tts_job.workflow import TTSWorkflowService + + use_case = CreateTTSJobUseCase(repo) + created_job = use_case.execute( + user_id="user_001", + input_text="测试文本", + voice_id="voice_001", + ) + + workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service) + + # 模拟路由的 try/except 兜底逻辑 + try: + result_job = workflow.start_synthesis(created_job.id) + except Exception as e: + result_job = workflow.process_synthesis_failure(created_job.id, str(e)) + + assert result_job.status == TTSJobStatus.FAILED + assert "unexpected" in result_job.error_message + + def test_route_handles_double_failure_gracefully(self) -> None: + """路由层兜底：即使 process_synthesis_failure 也失败，不崩溃。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = RuntimeError("unexpected") + + repo = MagicMock() + job = _make_job() + repo.get.return_value = job + repo.update.side_effect = lambda j: j + repo.create.side_effect = lambda j: j + + from packages.application.tts_job.use_cases import CreateTTSJobUseCase + from packages.application.tts_job.workflow import TTSWorkflowService + + use_case = CreateTTSJobUseCase(repo) + created_job = use_case.execute( + user_id="user_001", + input_text="测试文本", + voice_id="voice_001", + ) + + # 让 process_synthesis_failure 也失败 + call_count = 0 + original_update = repo.update + + def failing_update(j): + nonlocal call_count + call_count += 1 + if call_count > 2: # 第3次调用时失败（mark_failed 时的 update） + raise RuntimeError("DB connection lost") + return original_update(j) + + repo.update.side_effect = failing_update + + workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service) + + # 模拟路由的 try/except 兜底逻辑（包含内层 try/except） + try: + result_job = workflow.start_synthesis(created_job.id) + except Exception as e: + try: + result_job = workflow.process_synthesis_failure(created_job.id, str(e)) + except Exception: + # 路由层日志记录后，使用原始 job 对象 + result_job = created_job + + # 不应崩溃，即使状态可能不是 failed + assert result_job is not None