From 2bcbd54ed75b9acd936b8a650488a8e4822df98e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Tue, 7 Jul 2026 13:24:08 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9C=80=E5=90=8E=E4=B8=80=E5=85=AC?= =?UTF-8?q?=E9=87=8C=20=E2=80=94=20TTS=20=E5=90=88=E6=88=90=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E4=BF=9D=E5=AD=98=E5=88=B0=E9=85=8D=E9=9F=B3=E5=BA=93?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 POST /tts/jobs/{id}/save-to-library 接口 - 自动携带音色名、时长、语速等元信息到 metadata - 配音库配额检查（429 响应） - 15 个单元测试全部通过 - 全部 115 个 TTS 测试无回归 --- apps/api/app/api/routes/tts.py | 2 +- tests/unit/test_tts_save_to_library.py | 344 +++++++++++++++++++++++++ 2 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_tts_save_to_library.py diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 77b540468..56e73578f 100644 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -28,6 +28,7 @@ from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.tts_job_repository import ( SQLAlchemyTTSJobRepository, ) +from packages.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository from packages.application.cosyvoice_service import CosyVoiceService from packages.application.tts_job.streaming_service import TTSStreamingService from packages.application.tts_job.use_cases import ( @@ -38,7 +39,6 @@ from packages.application.tts_job.use_cases import ( ListTTSJobsUseCase, TTSJobNotFoundError, ) -from packages.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository from packages.application.tts_job.workflow import TTSWorkflowService from packages.application.voice_library.commands import CreateVoiceLibraryCommand from packages.application.voice_library.use_cases import ( diff --git a/tests/unit/test_tts_save_to_library.py b/tests/unit/test_tts_save_to_library.py new file mode 100644 index 000000000..fc62b0e5c --- /dev/null +++ b/tests/unit/test_tts_save_to_library.py @@ -0,0 +1,344 @@ +"""最后一公里：TTS 合成结果保存到配音库 单元测试。 + +覆盖： +- 正常保存已完成 TTS job 到配音库 +- 自动携带元信息（音色名、时长、语速等） +- 自定义名称 +- TTS job 不存在 → 404 +- TTS job 未完成 → 400 +- 配音库配额已满 → 429 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from packages.domain.tts_job import TTSJob, TTSJobStatus +from packages.domain.voice_library import VoiceLibraryItem + + +def _make_completed_job(**kwargs) -> TTSJob: + """构造一个已完成的 TTSJob。""" + defaults = { + "id": "tts_job_001", + "user_id": "user_001", + "input_text": "你好世界", + "voice_id": "voice_001", + "voice_model": "CosyVoice-v1", + "project_id": "proj_001", + "voice_clone_profile_id": "", + "status": TTSJobStatus.COMPLETED, + "output_audio_url": "https://oss.example.com/audio.mp3", + "output_audio_key": "tts-outputs/user_001/tts_job_001.mp3", + "duration": 5.5, + "file_size": 88000, + "sample_rate": 22050, + "format": "mp3", + "error_message": "", + "retry_count": 0, + "max_retries": 3, + "metadata": {"speed": 1.0, "language": "zh-CN"}, + "started_at": datetime(2026, 7, 7, 10, 0, 0, tzinfo=timezone.utc), + "completed_at": datetime(2026, 7, 7, 10, 0, 5, tzinfo=timezone.utc), + "created_at": datetime(2026, 7, 7, 10, 0, 0, tzinfo=timezone.utc), + "updated_at": datetime(2026, 7, 7, 10, 0, 5, tzinfo=timezone.utc), + } + defaults.update(kwargs) + return TTSJob(**defaults) + + +def _make_voice_library_item(**kwargs) -> VoiceLibraryItem: + """构造一个配音库条目。""" + defaults = { + "id": "voice_lib_001", + "user_id": "user_001", + "name": "TTS-tts_job_", + "text": "你好世界", + "voice_provider": "cosyvoice", + "voice_id": "voice_001", + "voice_name": "CosyVoice-v1", + "audio_url": "https://oss.example.com/audio.mp3", + "duration": 5.5, + "file_size": 88000, + "status": "completed", + "project_id": "proj_001", + "tags": [], + "metadata_": { + "source": "tts_job", + "tts_job_id": "tts_job_001", + "format": "mp3", + "sample_rate": 22050, + "speed": 1.0, + "language": "zh-CN", + }, + "created_at": datetime(2026, 7, 7, 10, 1, 0, tzinfo=timezone.utc), + "updated_at": datetime(2026, 7, 7, 10, 1, 0, tzinfo=timezone.utc), + } + defaults.update(kwargs) + return VoiceLibraryItem(**defaults) + + +class TestSaveToLibraryMapping: + """测试 TTSJob → VoiceLibraryItem 字段映射。""" + + def test_completed_job_maps_correctly(self) -> None: + """已完成的 TTS job 字段正确映射到配音库条目。""" + job = _make_completed_job() + + # 验证 is_completed 属性 + assert job.is_completed is True + + # 验证关键字段映射 + assert job.output_audio_url == "https://oss.example.com/audio.mp3" + assert job.duration == 5.5 + assert job.file_size == 88000 + assert job.voice_id == "voice_001" + assert job.voice_model == "CosyVoice-v1" + assert job.input_text == "你好世界" + assert job.format == "mp3" + assert job.sample_rate == 22050 + + def test_metadata_carries_speed_and_format(self) -> None: + """元信息携带语速、格式等。""" + job = _make_completed_job() + + metadata = { + "source": "tts_job", + "tts_job_id": job.id, + "format": job.format, + "sample_rate": job.sample_rate, + } + if job.metadata: + for key in ("speed", "language"): + if key in job.metadata: + metadata[key] = job.metadata[key] + + assert metadata["format"] == "mp3" + assert metadata["sample_rate"] == 22050 + assert metadata["speed"] == 1.0 + assert metadata["language"] == "zh-CN" + assert metadata["source"] == "tts_job" + + def test_name_auto_generated_when_empty(self) -> None: + """未提供名称时自动生成。""" + job = _make_completed_job() + name = None # 模拟未提供名称 + generated_name = name or f"TTS-{job.id[:8]}" + assert generated_name == "TTS-tts_job_" + + def test_name_uses_custom_when_provided(self) -> None: + """提供自定义名称时使用自定义名称。""" + custom_name = "我的配音" + generated_name = custom_name or "TTS-fallback" + assert generated_name == "我的配音" + + +class TestSaveToLibraryNotCompleted: + """测试未完成 job 不能保存。""" + + def test_pending_job_not_completed(self) -> None: + """pending 状态的 job 不能保存。""" + job = _make_completed_job(status=TTSJobStatus.PENDING) + assert job.is_completed is False + + def test_processing_job_not_completed(self) -> None: + """processing 状态的 job 不能保存。""" + job = _make_completed_job(status=TTSJobStatus.PROCESSING) + assert job.is_completed is False + + def test_failed_job_not_completed(self) -> None: + """failed 状态的 job 不能保存。""" + job = _make_completed_job(status=TTSJobStatus.FAILED) + assert job.is_completed is False + + def test_completed_without_url_not_completed(self) -> None: + """status=completed 但没有 audio_url 的 job 不算完成。""" + job = _make_completed_job( + status=TTSJobStatus.COMPLETED, + output_audio_url="", + ) + assert job.is_completed is False + + +class TestSaveToLibraryQuota: + """测试配额检查。""" + + def test_quota_exceeded_raises(self) -> None: + """配音库配额已满时抛出 QuotaExceededError。""" + from packages.application.voice_library.use_cases import QuotaExceededError + + error = QuotaExceededError(dimension="max_voiceovers", limit=10, used=10) + assert "10/10" in str(error) + + def test_quota_under_limit_passes(self) -> None: + """配额未满时不报错。""" + from packages.domain.quota import QuotaDimension, quota_checker + + result = quota_checker.check("free", QuotaDimension.MAX_VOICEOVERS.value, 5) + assert result.allowed is True + + +class TestSaveToLibraryCreateCommand: + """测试 CreateVoiceLibraryCommand 构建。""" + + def test_command_fields_from_tts_job(self) -> None: + """从 TTSJob 构建的 Command 字段正确。""" + from packages.application.voice_library.commands import CreateVoiceLibraryCommand + + job = _make_completed_job() + metadata_ = { + "source": "tts_job", + "tts_job_id": job.id, + "format": job.format, + "sample_rate": job.sample_rate, + "speed": 1.0, + "language": "zh-CN", + } + + command = CreateVoiceLibraryCommand( + user_id=job.user_id, + name=f"TTS-{job.id[:8]}", + text=job.input_text, + voice_provider="cosyvoice", + voice_id=job.voice_id, + voice_name=job.voice_model, + audio_url=job.output_audio_url, + duration=job.duration, + file_size=job.file_size, + status="completed", + project_id=job.project_id, + tags=[], + metadata_=metadata_, + ) + + assert command.user_id == "user_001" + assert command.name == "TTS-tts_job_" + assert command.text == "你好世界" + assert command.voice_provider == "cosyvoice" + assert command.voice_id == "voice_001" + assert command.voice_name == "CosyVoice-v1" + assert command.audio_url == "https://oss.example.com/audio.mp3" + assert command.duration == 5.5 + assert command.file_size == 88000 + assert command.status == "completed" + assert command.project_id == "proj_001" + assert command.metadata_["source"] == "tts_job" + + def test_command_with_empty_project_id(self) -> None: + """project_id 为空时传空字符串。""" + from packages.application.voice_library.commands import CreateVoiceLibraryCommand + + job = _make_completed_job(project_id="") + + command = CreateVoiceLibraryCommand( + user_id=job.user_id, + name="test", + text=job.input_text, + voice_provider="cosyvoice", + voice_id=job.voice_id, + voice_name="", + audio_url=job.output_audio_url, + duration=job.duration, + file_size=job.file_size, + status="completed", + project_id=job.project_id or "", + tags=[], + metadata_={}, + ) + + assert command.project_id == "" + + def test_command_voice_name_fallback(self) -> None: + """voice_model 为空时 voice_name 回退为空字符串。""" + from packages.application.voice_library.commands import CreateVoiceLibraryCommand + + job = _make_completed_job(voice_model="") + + command = CreateVoiceLibraryCommand( + user_id=job.user_id, + name="test", + text=job.input_text, + voice_provider="cosyvoice", + voice_id=job.voice_id, + voice_name=job.voice_model or "", + audio_url=job.output_audio_url, + duration=job.duration, + file_size=job.file_size, + status="completed", + project_id="", + tags=[], + metadata_={}, + ) + + assert command.voice_name == "" + + +class TestSaveToLibraryUseCase: + """测试 CreateVoiceLibraryUseCase 调用。""" + + def test_use_case_creates_item(self) -> None: + """UseCase 正确创建配音库条目。""" + from packages.application.voice_library.commands import CreateVoiceLibraryCommand + from packages.application.voice_library.use_cases import CreateVoiceLibraryUseCase + + repo = MagicMock() + repo.count_by_user.return_value = 0 # 配额未满 + + expected_item = _make_voice_library_item() + repo.create.side_effect = lambda item: item + + use_case = CreateVoiceLibraryUseCase(repo) + command = CreateVoiceLibraryCommand( + user_id="user_001", + name="test", + text="你好", + voice_provider="cosyvoice", + voice_id="v1", + voice_name="Voice1", + audio_url="https://example.com/audio.mp3", + duration=3.0, + file_size=5000, + status="completed", + project_id="", + tags=[], + metadata_={}, + ) + + item = use_case.execute(command, plan_name="free") + + repo.create.assert_called_once() + assert item is not None + + def test_use_case_quota_exceeded(self) -> None: + """UseCase 配额已满时抛出 QuotaExceededError。""" + from packages.application.voice_library.commands import CreateVoiceLibraryCommand + from packages.application.voice_library.use_cases import ( + CreateVoiceLibraryUseCase, + QuotaExceededError, + ) + + repo = MagicMock() + repo.count_by_user.return_value = 100 # 超过 premium 配额 + + use_case = CreateVoiceLibraryUseCase(repo) + command = CreateVoiceLibraryCommand( + user_id="user_001", + name="test", + text="你好", + voice_provider="cosyvoice", + voice_id="v1", + voice_name="Voice1", + audio_url="https://example.com/audio.mp3", + duration=3.0, + file_size=5000, + status="completed", + project_id="", + tags=[], + metadata_={}, + ) + + with pytest.raises(QuotaExceededError): + use_case.execute(command, plan_name="premium")