diff --git a/tests/unit/test_tts_job_use_cases.py b/tests/unit/test_tts_job_use_cases.py new file mode 100755 index 000000000..5a2d8403f --- /dev/null +++ b/tests/unit/test_tts_job_use_cases.py @@ -0,0 +1,270 @@ +"""TTS Job UseCase 单元测试.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from packages.application.tts_job.exceptions import TTSJobNotFoundError +from packages.application.tts_job.use_cases import ( + CreateTTSJobUseCase, + DeleteTTSJobUseCase, + GetTTSJobStatusUseCase, + GetTTSJobUseCase, + ListTTSJobsUseCase, +) +from packages.domain.tts_job import TTSJob + + +def _make_job( + id: str = "job1", + user_id: str = "user_1", + status: str = "pending", +) -> TTSJob: + j = TTSJob.create( + user_id=user_id, + input_text="你好世界", + voice_id="voice_001", + voice_model="cosyvoice", + project_id="proj_1", + voice_clone_profile_id="", + sample_rate=22050, + format="mp3", + max_retries=3, + ) + j.id = id + object.__setattr__(j, "status", status) + return j + + +@pytest.fixture +def mock_repo(): + return MagicMock() + + +@pytest.fixture +def sample_job(): + return _make_job() + + +class TestCreateTTSJobUseCase: + """CreateTTSJobUseCase 测试""" + + def test_create_success(self, mock_repo, sample_job): + """创建成功""" + mock_repo.create.return_value = sample_job + use_case = CreateTTSJobUseCase(mock_repo) + + result = use_case.execute( + "user_1", + "你好世界", + voice_id="voice_001", + voice_model="cosyvoice", + project_id="proj_1", + ) + + assert result.id == "job1" + assert result.status == "pending" + mock_repo.create.assert_called_once() + created = mock_repo.create.call_args[0][0] + assert isinstance(created, TTSJob) + assert created.input_text == "你好世界" + + def test_create_with_defaults(self, mock_repo): + """使用默认参数创建""" + mock_repo.create.side_effect = lambda x: x + use_case = CreateTTSJobUseCase(mock_repo) + + result = use_case.execute("user_1", "测试文本") + + assert result.user_id == "user_1" + assert result.input_text == "测试文本" + assert result.sample_rate == 22050 + assert result.format == "mp3" + assert result.max_retries == 3 + + def test_create_with_clone_profile(self, mock_repo): + """创建时带音色克隆ID""" + mock_repo.create.side_effect = lambda x: x + use_case = CreateTTSJobUseCase(mock_repo) + + result = use_case.execute( + "user_1", "文本", voice_clone_profile_id="vc_123" + ) + + assert result.voice_clone_profile_id == "vc_123" + + def test_create_with_metadata(self, mock_repo): + """创建时带metadata""" + mock_repo.create.side_effect = lambda x: x + use_case = CreateTTSJobUseCase(mock_repo) + + result = use_case.execute( + "user_1", "text", metadata={"source": "api", "priority": "high"} + ) + + assert result.metadata == {"source": "api", "priority": "high"} + + +class TestListTTSJobsUseCase: + """ListTTSJobsUseCase 测试""" + + def test_list_returns_items_and_total(self, mock_repo, sample_job): + """返回 (items, total) 元组""" + mock_repo.list_by_user.return_value = [sample_job] + mock_repo.count_by_user.return_value = 1 + use_case = ListTTSJobsUseCase(mock_repo) + + items, total = use_case.execute("user_1") + + assert len(items) == 1 + assert total == 1 + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, limit=50, offset=0 + ) + mock_repo.count_by_user.assert_called_once_with("user_1", status=None) + + def test_list_with_status(self, mock_repo, sample_job): + """按状态过滤""" + mock_repo.list_by_user.return_value = [sample_job] + mock_repo.count_by_user.return_value = 5 + use_case = ListTTSJobsUseCase(mock_repo) + + use_case.execute("user_1", status="completed") + + mock_repo.list_by_user.assert_called_once_with( + "user_1", status="completed", limit=50, offset=0 + ) + mock_repo.count_by_user.assert_called_once_with("user_1", status="completed") + + def test_list_with_pagination(self, mock_repo, sample_job): + """带分页参数""" + mock_repo.list_by_user.return_value = [sample_job] + mock_repo.count_by_user.return_value = 100 + use_case = ListTTSJobsUseCase(mock_repo) + + use_case.execute("user_1", skip=20, limit=10) + + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, limit=10, offset=20 + ) + + def test_empty_list(self, mock_repo): + """空列表""" + mock_repo.list_by_user.return_value = [] + mock_repo.count_by_user.return_value = 0 + use_case = ListTTSJobsUseCase(mock_repo) + + items, total = use_case.execute("user_1") + + assert items == [] + assert total == 0 + + +class TestGetTTSJobUseCase: + """GetTTSJobUseCase 测试""" + + def test_get_existing(self, mock_repo, sample_job): + """获取存在的任务""" + mock_repo.get.return_value = sample_job + use_case = GetTTSJobUseCase(mock_repo) + + result = use_case.execute("job1", "user_1") + + assert result.id == "job1" + mock_repo.get.assert_called_once_with("job1") + + def test_get_nonexistent_raises(self, mock_repo): + """不存在抛出 TTSJobNotFoundError""" + mock_repo.get.return_value = None + use_case = GetTTSJobUseCase(mock_repo) + + with pytest.raises(TTSJobNotFoundError, match="not found"): + use_case.execute("noexist", "user_1") + + def test_get_wrong_user_raises(self, mock_repo, sample_job): + """非本人任务抛出""" + sample_job.user_id = "other_user" + mock_repo.get.return_value = sample_job + use_case = GetTTSJobUseCase(mock_repo) + + with pytest.raises(TTSJobNotFoundError): + use_case.execute("job1", "user_1") + + +class TestGetTTSJobStatusUseCase: + """GetTTSJobStatusUseCase 测试""" + + def test_get_status_pending(self, mock_repo, sample_job): + """获取 pending 状态""" + mock_repo.get.return_value = sample_job + use_case = GetTTSJobStatusUseCase(mock_repo) + + result = use_case.execute("job1", "user_1") + + assert result.status == "pending" + + def test_get_status_completed(self, mock_repo): + """获取 completed 状态""" + job = _make_job(status="completed") + mock_repo.get.return_value = job + use_case = GetTTSJobStatusUseCase(mock_repo) + + result = use_case.execute("job1", "user_1") + + assert result.status == "completed" + + def test_get_status_not_found(self, mock_repo): + """不存在抛出""" + mock_repo.get.return_value = None + use_case = GetTTSJobStatusUseCase(mock_repo) + + with pytest.raises(TTSJobNotFoundError): + use_case.execute("noexist", "user_1") + + def test_get_status_wrong_user(self, mock_repo, sample_job): + """非本人抛出""" + sample_job.user_id = "other" + mock_repo.get.return_value = sample_job + use_case = GetTTSJobStatusUseCase(mock_repo) + + with pytest.raises(TTSJobNotFoundError): + use_case.execute("job1", "user_1") + + +class TestDeleteTTSJobUseCase: + """DeleteTTSJobUseCase 测试""" + + def test_delete_success(self, mock_repo, sample_job): + """删除成功""" + mock_repo.get.return_value = sample_job + mock_repo.delete.return_value = True + use_case = DeleteTTSJobUseCase(mock_repo) + + result = use_case.execute("job1", "user_1") + + assert result is True + mock_repo.get.assert_called_once_with("job1") + mock_repo.delete.assert_called_once_with("job1") + + def test_delete_nonexistent_returns_false(self, mock_repo): + """不存在返回 False""" + mock_repo.get.return_value = None + use_case = DeleteTTSJobUseCase(mock_repo) + + result = use_case.execute("noexist", "user_1") + + assert result is False + mock_repo.delete.assert_not_called() + + def test_delete_wrong_user_returns_false(self, mock_repo, sample_job): + """非本人返回 False""" + sample_job.user_id = "other" + mock_repo.get.return_value = sample_job + use_case = DeleteTTSJobUseCase(mock_repo) + + result = use_case.execute("job1", "user_1") + + assert result is False + mock_repo.delete.assert_not_called() diff --git a/tests/unit/test_voice_clone_use_cases.py b/tests/unit/test_voice_clone_use_cases.py new file mode 100755 index 000000000..63cac0832 --- /dev/null +++ b/tests/unit/test_voice_clone_use_cases.py @@ -0,0 +1,300 @@ +"""音色克隆 Voice Clone UseCase 单元测试.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from packages.application.voice_clone.use_cases import ( + CreateVoiceCloneUseCase, + DeleteVoiceCloneUseCase, + GetVoiceCloneStatusUseCase, + GetVoiceCloneUseCase, + ListVoiceClonesUseCase, + RetryVoiceCloneUseCase, + VoiceCloneNotFoundError, + VoiceCloneNotRetryableError, +) +from packages.domain.voice_clone_profile import VoiceCloneProfile + + +def _make_profile( + id: str = "vc1", + user_id: str = "user_1", + name: str = "我的音色", + status: str = "completed", +) -> VoiceCloneProfile: + p = VoiceCloneProfile.create( + user_id=user_id, + name=name, + description="测试音色", + source_audio_url="https://oss.example.com/source.wav", + voice_model="cosyvoice", + language="zh-CN", + gender="female", + max_retries=3, + ) + p.id = id + # 直接设状态绕过状态机 + object.__setattr__(p, "status", status) + return p + + +@pytest.fixture +def mock_repo(): + return MagicMock() + + +@pytest.fixture +def sample_profile(): + return _make_profile() + + +class TestCreateVoiceCloneUseCase: + """CreateVoiceCloneUseCase 测试""" + + def test_create_success(self, mock_repo, sample_profile): + """创建成功""" + mock_repo.create.return_value = sample_profile + use_case = CreateVoiceCloneUseCase(mock_repo) + + result = use_case.execute( + "user_1", + "新音色", + description="自定义音色", + source_audio_url="https://oss.example.com/src.wav", + voice_model="cosyvoice", + language="zh-CN", + gender="female", + ) + + assert result.id == "vc1" + assert result.name == "我的音色" + mock_repo.create.assert_called_once() + created = mock_repo.create.call_args[0][0] + assert isinstance(created, VoiceCloneProfile) + + def test_create_with_defaults(self, mock_repo): + """使用默认参数创建""" + mock_repo.create.side_effect = lambda x: x + use_case = CreateVoiceCloneUseCase(mock_repo) + + result = use_case.execute("user_1", "极简音色") + + assert result.user_id == "user_1" + assert result.name == "极简音色" + assert result.language == "zh-CN" + assert result.gender == "unknown" + assert result.max_retries == 3 + + +class TestListVoiceClonesUseCase: + """ListVoiceClonesUseCase 测试""" + + def test_list_returns_items_and_total(self, mock_repo, sample_profile): + """返回 (items, total) 元组""" + mock_repo.list_by_user.return_value = [sample_profile] + mock_repo.count_by_user.return_value = 1 + use_case = ListVoiceClonesUseCase(mock_repo) + + items, total = use_case.execute("user_1") + + assert len(items) == 1 + assert total == 1 + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, limit=50, offset=0 + ) + mock_repo.count_by_user.assert_called_once_with("user_1", status=None) + + def test_list_with_status(self, mock_repo, sample_profile): + """按状态过滤""" + mock_repo.list_by_user.return_value = [sample_profile] + mock_repo.count_by_user.return_value = 1 + use_case = ListVoiceClonesUseCase(mock_repo) + + use_case.execute("user_1", status="processing") + + mock_repo.list_by_user.assert_called_once_with( + "user_1", status="processing", limit=50, offset=0 + ) + mock_repo.count_by_user.assert_called_once_with("user_1", status="processing") + + def test_list_with_pagination(self, mock_repo, sample_profile): + """带分页参数""" + mock_repo.list_by_user.return_value = [sample_profile] + mock_repo.count_by_user.return_value = 10 + use_case = ListVoiceClonesUseCase(mock_repo) + + use_case.execute("user_1", skip=5, limit=10) + + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, limit=10, offset=5 + ) + + def test_empty_list(self, mock_repo): + """空列表""" + mock_repo.list_by_user.return_value = [] + mock_repo.count_by_user.return_value = 0 + use_case = ListVoiceClonesUseCase(mock_repo) + + items, total = use_case.execute("user_1") + + assert items == [] + assert total == 0 + + +class TestGetVoiceCloneUseCase: + """GetVoiceCloneUseCase 测试""" + + def test_get_existing(self, mock_repo, sample_profile): + """获取存在的音色克隆""" + mock_repo.get.return_value = sample_profile + use_case = GetVoiceCloneUseCase(mock_repo) + + result = use_case.execute("vc1", "user_1") + + assert result.id == "vc1" + mock_repo.get.assert_called_once_with("vc1") + + def test_get_nonexistent_raises(self, mock_repo): + """获取不存在的抛出 VoiceCloneNotFoundError""" + mock_repo.get.return_value = None + use_case = GetVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError, match="not found"): + use_case.execute("noexist", "user_1") + + def test_get_wrong_user_raises(self, mock_repo, sample_profile): + """非本人的音色克隆抛出 NotFoundError""" + sample_profile.user_id = "other_user" + mock_repo.get.return_value = sample_profile + use_case = GetVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("vc1", "user_1") + + +class TestGetVoiceCloneStatusUseCase: + """GetVoiceCloneStatusUseCase 测试""" + + def test_get_status_completed(self, mock_repo, sample_profile): + """获取 completed 状态""" + mock_repo.get.return_value = sample_profile + use_case = GetVoiceCloneStatusUseCase(mock_repo) + + result = use_case.execute("vc1", "user_1") + + assert result.status == "completed" + + def test_get_status_not_found(self, mock_repo): + """不存在抛出""" + mock_repo.get.return_value = None + use_case = GetVoiceCloneStatusUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("noexist", "user_1") + + def test_get_status_wrong_user(self, mock_repo, sample_profile): + """非本人抛出""" + sample_profile.user_id = "other" + mock_repo.get.return_value = sample_profile + use_case = GetVoiceCloneStatusUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("vc1", "user_1") + + +class TestDeleteVoiceCloneUseCase: + """DeleteVoiceCloneUseCase 测试""" + + def test_delete_success(self, mock_repo, sample_profile): + """删除成功""" + mock_repo.get.return_value = sample_profile + mock_repo.delete.return_value = True + use_case = DeleteVoiceCloneUseCase(mock_repo) + + result = use_case.execute("vc1", "user_1") + + assert result is True + mock_repo.get.assert_called_once_with("vc1") + mock_repo.delete.assert_called_once_with("vc1") + + def test_delete_nonexistent_returns_false(self, mock_repo): + """不存在返回 False""" + mock_repo.get.return_value = None + use_case = DeleteVoiceCloneUseCase(mock_repo) + + result = use_case.execute("noexist", "user_1") + + assert result is False + mock_repo.delete.assert_not_called() + + def test_delete_wrong_user_returns_false(self, mock_repo, sample_profile): + """非本人返回 False""" + sample_profile.user_id = "other" + mock_repo.get.return_value = sample_profile + use_case = DeleteVoiceCloneUseCase(mock_repo) + + result = use_case.execute("vc1", "user_1") + + assert result is False + mock_repo.delete.assert_not_called() + + +class TestRetryVoiceCloneUseCase: + """RetryVoiceCloneUseCase 测试""" + + def test_retry_failed_profile(self, mock_repo): + """失败的音色克隆可以重试""" + profile = _make_profile(status="failed") + mock_repo.get.return_value = profile + mock_repo.update.side_effect = lambda x: x + use_case = RetryVoiceCloneUseCase(mock_repo) + + result = use_case.execute("vc1", "user_1") + + assert result is not None + assert result.status == "pending" + assert result.retry_count >= 1 + mock_repo.update.assert_called_once() + + def test_retry_not_found(self, mock_repo): + """不存在抛出""" + mock_repo.get.return_value = None + use_case = RetryVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("noexist", "user_1") + + mock_repo.update.assert_not_called() + + def test_retry_wrong_user(self, mock_repo, sample_profile): + """非本人抛出""" + sample_profile.user_id = "other" + mock_repo.get.return_value = sample_profile + use_case = RetryVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotFoundError): + use_case.execute("vc1", "user_1") + + mock_repo.update.assert_not_called() + + def test_retry_completed_not_retryable(self, mock_repo, sample_profile): + """completed 状态不可重试""" + mock_repo.get.return_value = sample_profile + use_case = RetryVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotRetryableError): + use_case.execute("vc1", "user_1") + + mock_repo.update.assert_not_called() + + def test_retry_processing_not_retryable(self, mock_repo): + """processing 状态不可重试""" + profile = _make_profile(status="processing") + mock_repo.get.return_value = profile + use_case = RetryVoiceCloneUseCase(mock_repo) + + with pytest.raises(VoiceCloneNotRetryableError): + use_case.execute("vc1", "user_1") diff --git a/tests/unit/test_voice_library_use_cases.py b/tests/unit/test_voice_library_use_cases.py old mode 100644 new mode 100755 index 87cf3a6a4..cb012e97c --- a/tests/unit/test_voice_library_use_cases.py +++ b/tests/unit/test_voice_library_use_cases.py @@ -1,14 +1,8 @@ -""" -配音库（Voice Library）Use Case 回归测试 +"""音色库 UseCase 单元测试.""" -测试目标： -1. CreateVoiceLibraryUseCase - 创建配音库条目，验证 voice_id 字段映射正确（PR#74 P0 bug 修复） -2. UpdateVoiceLibraryUseCase - 更新配音库条目，验证 voice_id 字段映射正确 -3. 配额逻辑覆盖 - free=10, basic=100, premium=100 -4. 边界条件与异常场景 -""" +from __future__ import annotations -from unittest.mock import Mock +from unittest.mock import MagicMock import pytest @@ -21,671 +15,259 @@ from packages.application.voice_library.use_cases import ( DeleteVoiceLibraryUseCase, GetVoiceLibraryUseCase, ListVoiceLibraryUseCase, - NotFoundError, - QuotaExceededError, UpdateVoiceLibraryUseCase, ) +from packages.domain.exceptions import NotFoundError, QuotaExceededError from packages.domain.voice_library import VoiceLibraryItem -# --------------------------------------------------------------------------- -# Fixtures -# --------------------------------------------------------------------------- - -@pytest.fixture -def mock_repo(): - """创建 Mock 仓储""" - repo = Mock() - repo.count_by_user = Mock(return_value=0) - repo.create = Mock(side_effect=lambda item: item) - repo.update = Mock(side_effect=lambda item: item) - repo.get = Mock(return_value=None) - repo.delete = Mock(return_value=True) - repo.list_by_user = Mock(return_value=[]) - return repo - - -@pytest.fixture -def create_use_case(mock_repo): - return CreateVoiceLibraryUseCase(repository=mock_repo) - - -@pytest.fixture -def update_use_case(mock_repo): - return UpdateVoiceLibraryUseCase(repository=mock_repo) - - -@pytest.fixture -def sample_create_command(): - """标准创建命令""" - return CreateVoiceLibraryCommand( - user_id="user-001", - name="测试配音", - text="你好世界", - voice_provider="aliyun", - voice_id="voice-abc-123", - voice_name="小云", - audio_url="https://oss.example.com/audio/abc.wav", - duration=3.5, - file_size=56000, - status="completed", - project_id="proj-001", - tags=["测试", "中文"], - metadata_={"source": "unit_test"}, - ) - - -@pytest.fixture -def existing_voice_item(): - """模拟已存在的配音条目""" +def _make_item(id: str = "v1", name: str = "测试音色", user_id: str = "user_1") -> VoiceLibraryItem: return VoiceLibraryItem( - id="existing-voice-001", - user_id="user-001", - name="旧配音", - text="旧文本", - voice_provider="old_provider", - voice_id="old-voice-id", - voice_name="旧声音", - audio_url="https://oss.example.com/old.wav", - duration=1.0, - file_size=16000, - status="completed", - project_id="proj-001", - tags=["旧"], + id=id, + user_id=user_id, + name=name, + text="示例文本", + voice_provider="cosyvoice", + voice_id="voice_001", + voice_name="温柔女声", + audio_url="https://oss.example.com/voice.mp3", + duration=10.5, + file_size=102400, + status="ready", + project_id="", + tags=["温柔", "女声"], metadata_={}, ) -# =========================================================================== -# 1. CreateVoiceLibraryUseCase 测试 -# =========================================================================== +@pytest.fixture +def mock_repo(): + return MagicMock() -class TestCreateVoiceLibraryUseCase: - """配音库创建 UseCase 测试""" - - def test_create_success_all_fields(self, create_use_case, mock_repo, sample_create_command): - """测试创建成功 - 所有字段完整传入""" - result = create_use_case.execute(sample_create_command, plan_name="free") - - assert result is not None - assert result.user_id == "user-001" - assert result.name == "测试配音" - assert result.text == "你好世界" - assert result.voice_provider == "aliyun" - assert result.voice_name == "小云" - assert result.audio_url == "https://oss.example.com/audio/abc.wav" - assert result.duration == 3.5 - assert result.file_size == 56000 - assert result.status == "completed" - assert result.project_id == "proj-001" - assert result.tags == ["测试", "中文"] - assert result.metadata_ == {"source": "unit_test"} - - mock_repo.count_by_user.assert_called_once_with("user-001") - mock_repo.create.assert_called_once() - - def test_create_voice_id_field_mapping(self, create_use_case, mock_repo): - """ - 【P0 回归】验证 voice_id 字段映射正确 - - PR#74 修复了 command.id 被错误使用的问题。 - 此测试确保 CreateVoiceLibraryCommand 中的 voice_id 字段 - 被正确传递到 VoiceLibraryItem 的 voice_id 属性上， - 而非被其他字段（如 item 自身的 id）覆盖。 - """ - command = CreateVoiceLibraryCommand( - user_id="user-001", - name="voice_id 回归测试", - voice_id="specific-voice-id-xyz", - voice_provider="azure", - voice_name="Azure Xiaoxiao", - ) - - result = create_use_case.execute(command, plan_name="free") - - # 核心断言：voice_id 必须来自 command.voice_id - assert result.voice_id == "specific-voice-id-xyz", "voice_id 应来自 command.voice_id，而非其他字段" - # 同时确保 item 自身生成的 id 与 voice_id 不同 - assert result.id != "specific-voice-id-xyz", "item.id（UUID）不应与 voice_id 混淆" - - def test_create_voice_id_empty_string(self, create_use_case, mock_repo): - """测试 voice_id 为空字符串的合法场景""" - command = CreateVoiceLibraryCommand( - user_id="user-001", - name="无 voice_id 配音", - voice_id="", - voice_provider="custom", - ) - - result = create_use_case.execute(command, plan_name="free") - - assert result.voice_id == "" - - def test_create_default_values(self, create_use_case, mock_repo): - """测试默认值填充""" - command = CreateVoiceLibraryCommand( - user_id="user-001", - name="最小化创建", - ) - - result = create_use_case.execute(command, plan_name="free") - - assert result.text == "" - assert result.voice_provider == "" - assert result.voice_id == "" - assert result.voice_name == "" - assert result.audio_url == "" - assert result.duration == 0 - assert result.file_size == 0 - assert result.status == "completed" - assert result.project_id is None - assert result.tags == [] - assert result.metadata_ == {} - - def test_create_generates_uuid(self, create_use_case, mock_repo): - """测试创建时自动生成 UUID 作为 id""" - command = CreateVoiceLibraryCommand( - user_id="user-001", - name="UUID 测试", - ) - - result = create_use_case.execute(command, plan_name="free") - - assert result.id is not None - assert len(result.id) == 32 # uuid4().hex 长度为 32 - assert result.id.isalnum() - - -# =========================================================================== -# 2. 配额逻辑测试（Create 时的配额检查） -# =========================================================================== - - -class TestCreateVoiceLibraryQuota: - """配音库创建配额检查测试""" - - def test_quota_free_plan_under_limit(self, create_use_case, mock_repo, sample_create_command): - """free 套餐（上限10），当前 5 个，允许创建""" - mock_repo.count_by_user.return_value = 5 - - result = create_use_case.execute(sample_create_command, plan_name="free") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_free_plan_at_limit(self, create_use_case, mock_repo, sample_create_command): - """free 套餐（上限10），当前 10 个，拒绝创建""" - mock_repo.count_by_user.return_value = 10 - - with pytest.raises(QuotaExceededError) as exc_info: - create_use_case.execute(sample_create_command, plan_name="free") - - assert exc_info.value.dimension == "max_voiceovers" - assert exc_info.value.limit == 10 - assert exc_info.value.used == 10 - mock_repo.create.assert_not_called() - - def test_quota_free_plan_over_limit(self, create_use_case, mock_repo, sample_create_command): - """free 套餐（上限10），当前 15 个，拒绝创建""" - mock_repo.count_by_user.return_value = 15 - - with pytest.raises(QuotaExceededError) as exc_info: - create_use_case.execute(sample_create_command, plan_name="free") - - assert exc_info.value.dimension == "max_voiceovers" - assert exc_info.value.limit == 10 - assert exc_info.value.used == 15 - - def test_quota_free_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command): - """free 套餐（上限10），当前 9 个，允许创建（边界）""" - mock_repo.count_by_user.return_value = 9 - - result = create_use_case.execute(sample_create_command, plan_name="free") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_basic_plan_under_limit(self, create_use_case, mock_repo, sample_create_command): - """basic 套餐（上限100），当前 50 个，允许创建""" - mock_repo.count_by_user.return_value = 50 - - result = create_use_case.execute(sample_create_command, plan_name="basic") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_basic_plan_at_limit(self, create_use_case, mock_repo, sample_create_command): - """basic 套餐（上限100），当前 100 个，拒绝创建""" - mock_repo.count_by_user.return_value = 100 - - with pytest.raises(QuotaExceededError) as exc_info: - create_use_case.execute(sample_create_command, plan_name="basic") - - assert exc_info.value.dimension == "max_voiceovers" - assert exc_info.value.limit == 100 - assert exc_info.value.used == 100 - - def test_quota_basic_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command): - """basic 套餐（上限100），当前 99 个，允许创建（边界）""" - mock_repo.count_by_user.return_value = 99 - - result = create_use_case.execute(sample_create_command, plan_name="basic") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_premium_plan_under_limit(self, create_use_case, mock_repo, sample_create_command): - """premium 套餐（上限100），当前 50 个，允许创建""" - mock_repo.count_by_user.return_value = 50 - - result = create_use_case.execute(sample_create_command, plan_name="premium") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_premium_plan_at_limit(self, create_use_case, mock_repo, sample_create_command): - """premium 套餐（上限100），当前 100 个，拒绝创建""" - mock_repo.count_by_user.return_value = 100 - - with pytest.raises(QuotaExceededError) as exc_info: - create_use_case.execute(sample_create_command, plan_name="premium") - - assert exc_info.value.dimension == "max_voiceovers" - assert exc_info.value.limit == 100 - assert exc_info.value.used == 100 - - def test_quota_premium_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command): - """premium 套餐（上限100），当前 99 个，允许创建（边界）""" - mock_repo.count_by_user.return_value = 99 - - result = create_use_case.execute(sample_create_command, plan_name="premium") - - assert result is not None - mock_repo.create.assert_called_once() - - def test_quota_zero_usage(self, create_use_case, mock_repo, sample_create_command): - """新用户零使用量，所有套餐均可创建""" - mock_repo.count_by_user.return_value = 0 - - for plan in ["free", "basic", "premium"]: - mock_repo.create.reset_mock() - mock_repo.count_by_user.reset_mock() - mock_repo.count_by_user.return_value = 0 - - result = create_use_case.execute(sample_create_command, plan_name=plan) - assert result is not None, f"{plan} 套餐零使用量应允许创建" - - def test_quota_unknown_plan_defaults_to_zero(self, create_use_case, mock_repo, sample_create_command): - """未知套餐名默认配额为 0，即使 0 使用量也无法创建""" - mock_repo.count_by_user.return_value = 0 - - with pytest.raises(QuotaExceededError): - create_use_case.execute(sample_create_command, plan_name="unknown_plan") - - def test_quota_exceeded_error_attributes(self, create_use_case, mock_repo, sample_create_command): - """QuotaExceededError 异常属性完整性""" - mock_repo.count_by_user.return_value = 10 - - with pytest.raises(QuotaExceededError) as exc_info: - create_use_case.execute(sample_create_command, plan_name="free") - - err = exc_info.value - assert hasattr(err, "dimension") - assert hasattr(err, "limit") - assert hasattr(err, "used") - assert "max_voiceovers" in str(err) - assert "10" in str(err) - - -# =========================================================================== -# 3. UpdateVoiceLibraryUseCase 测试 -# =========================================================================== - - -class TestUpdateVoiceLibraryUseCase: - """配音库更新 UseCase 测试""" - - def test_update_success_all_fields(self, update_use_case, mock_repo, existing_voice_item): - """测试全字段更新成功""" - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - name="更新后的名称", - text="更新后的文本", - voice_provider="new_provider", - voice_id="new-voice-id-456", - voice_name="新声音", - audio_url="https://oss.example.com/new.wav", - duration=5.0, - file_size=80000, - status="processing", - tags=["新标签"], - metadata_={"updated": True}, - ) - - result = update_use_case.execute(command) - - assert result.name == "更新后的名称" - assert result.text == "更新后的文本" - assert result.voice_provider == "new_provider" - assert result.voice_name == "新声音" - assert result.audio_url == "https://oss.example.com/new.wav" - assert result.duration == 5.0 - assert result.file_size == 80000 - assert result.status == "processing" - assert result.tags == ["新标签"] - assert result.metadata_ == {"updated": True} - - mock_repo.update.assert_called_once() - - def test_update_voice_id_field_mapping(self, update_use_case, mock_repo, existing_voice_item): - """ - 【P0 回归】验证 update 时 voice_id 字段映射正确 - - PR#74 修复了 API 路由层将 command.id 错误传给 voice_id 的 bug。 - 此测试确保 UpdateVoiceLibraryCommand 中 voice_id 字段 - 被正确写入 VoiceLibraryItem.voice_id，而非被 item.id 覆盖。 - """ - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - voice_id="completely-different-voice-id", - ) - - result = update_use_case.execute(command) - - # 核心断言：voice_id 应被更新为新值 - assert result.voice_id == "completely-different-voice-id", "voice_id 应被更新为 command.voice_id 的值" - # item 自身的 id 保持不变 - assert result.id == "existing-voice-001" - - def test_update_partial_only_voice_id(self, update_use_case, mock_repo, existing_voice_item): - """测试仅更新 voice_id 一个字段""" - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - voice_id="only-voice-id-changed", - ) - - result = update_use_case.execute(command) - - assert result.voice_id == "only-voice-id-changed" - # 其他字段保持不变 - assert result.name == "旧配音" - assert result.text == "旧文本" - assert result.voice_provider == "old_provider" - assert result.voice_name == "旧声音" - assert result.audio_url == "https://oss.example.com/old.wav" - assert result.duration == 1.0 - assert result.file_size == 16000 - - def test_update_partial_only_name(self, update_use_case, mock_repo, existing_voice_item): - """测试仅更新 name""" - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - name="仅改名", - ) - - result = update_use_case.execute(command) - - assert result.name == "仅改名" - assert result.voice_id == "old-voice-id" # voice_id 不变 - - def test_update_not_found(self, update_use_case, mock_repo): - """测试更新不存在的条目""" - mock_repo.get.return_value = None - - command = UpdateVoiceLibraryCommand( - id="nonexistent-id", - user_id="user-001", - name="不存在", - ) - - with pytest.raises(NotFoundError, match="nonexistent-id"): - update_use_case.execute(command) - - mock_repo.update.assert_not_called() - - def test_update_wrong_user(self, update_use_case, mock_repo): - """测试用户隔离 - 不能更新其他用户的条目""" - mock_repo.get.return_value = None # repo 返回 None 表示找不到（不同 user_id） - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="other-user-999", - name="恶意修改", - ) - - with pytest.raises(NotFoundError): - update_use_case.execute(command) - - def test_update_none_fields_not_changed(self, update_use_case, mock_repo, existing_voice_item): - """测试 None 字段不覆盖原有值""" - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - # 所有可选字段保持 None - ) - - result = update_use_case.execute(command) - - # 所有字段应保持不变 - assert result.name == "旧配音" - assert result.text == "旧文本" - assert result.voice_id == "old-voice-id" - assert result.voice_provider == "old_provider" - assert result.voice_name == "旧声音" - assert result.audio_url == "https://oss.example.com/old.wav" - assert result.duration == 1.0 - assert result.file_size == 16000 - assert result.status == "completed" - - def test_update_voice_id_empty_string(self, update_use_case, mock_repo, existing_voice_item): - """测试 voice_id 更新为空字符串（合法场景：清除 voice_id）""" - mock_repo.get.return_value = existing_voice_item - - command = UpdateVoiceLibraryCommand( - id="existing-voice-001", - user_id="user-001", - voice_id="", - ) - - result = update_use_case.execute(command) - - assert result.voice_id == "" - - -# =========================================================================== -# 4. DeleteVoiceLibraryUseCase 测试 -# =========================================================================== - - -class TestDeleteVoiceLibraryUseCase: - """配音库删除 UseCase 测试""" - - def test_delete_success(self, mock_repo): - """测试删除成功""" - mock_repo.delete.return_value = True - use_case = DeleteVoiceLibraryUseCase(repository=mock_repo) - - result = use_case.execute("voice-001", "user-001") - - assert result is True - mock_repo.delete.assert_called_once_with("voice-001", "user-001") - - def test_delete_not_found(self, mock_repo): - """测试删除不存在的条目""" - mock_repo.delete.return_value = False - use_case = DeleteVoiceLibraryUseCase(repository=mock_repo) - - result = use_case.execute("nonexistent", "user-001") - - assert result is False - - -# =========================================================================== -# 5. GetVoiceLibraryUseCase 测试 -# =========================================================================== - - -class TestGetVoiceLibraryUseCase: - """配音库查询 UseCase 测试""" - - def test_get_existing(self, mock_repo): - """测试查询存在的条目""" - expected = VoiceLibraryItem( - id="v-001", - user_id="user-001", - name="测试", - voice_id="voice-xyz", - ) - mock_repo.get.return_value = expected - use_case = GetVoiceLibraryUseCase(repository=mock_repo) - - result = use_case.execute("v-001", "user-001") - - assert result is not None - assert result.id == "v-001" - assert result.voice_id == "voice-xyz" - mock_repo.get.assert_called_once_with("v-001", "user-001") - - def test_get_not_found(self, mock_repo): - """测试查询不存在的条目""" - mock_repo.get.return_value = None - use_case = GetVoiceLibraryUseCase(repository=mock_repo) - - result = use_case.execute("nonexistent", "user-001") - - assert result is None - - -# =========================================================================== -# 6. ListVoiceLibraryUseCase 测试 -# =========================================================================== +@pytest.fixture +def sample_item(): + return _make_item() class TestListVoiceLibraryUseCase: - """配音库列表 UseCase 测试""" + """ListVoiceLibraryUseCase 测试""" - def test_list_default(self, mock_repo): - """测试默认列表查询""" - items = [ - VoiceLibraryItem(id="v1", user_id="user-001", name="A"), - VoiceLibraryItem(id="v2", user_id="user-001", name="B"), - ] - mock_repo.list_by_user.return_value = items - mock_repo.count_by_user.return_value = 2 - use_case = ListVoiceLibraryUseCase(repository=mock_repo) + def test_list_returns_items_and_total(self, mock_repo, sample_item): + """返回 (items, total_count) 元组""" + mock_repo.list_by_user.return_value = [sample_item] + mock_repo.count_by_user.return_value = 1 + use_case = ListVoiceLibraryUseCase(mock_repo) - result_items, total = use_case.execute("user-001") + items, total = use_case.execute("user_1") - assert len(result_items) == 2 - assert total == 2 - mock_repo.list_by_user.assert_called_once_with("user-001", status=None, skip=0, limit=50) + assert len(items) == 1 + assert items[0].id == "v1" + assert total == 1 + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, skip=0, limit=50 + ) + mock_repo.count_by_user.assert_called_once_with("user_1") - def test_list_with_status_filter(self, mock_repo): - """测试按状态筛选""" - mock_repo.list_by_user.return_value = [] - use_case = ListVoiceLibraryUseCase(repository=mock_repo) + def test_list_with_status_filter(self, mock_repo, sample_item): + """按状态过滤""" + mock_repo.list_by_user.return_value = [sample_item] + mock_repo.count_by_user.return_value = 1 + use_case = ListVoiceLibraryUseCase(mock_repo) - use_case.execute("user-001", status="completed", skip=10, limit=20) + items, total = use_case.execute("user_1", status="ready") - mock_repo.list_by_user.assert_called_once_with("user-001", status="completed", skip=10, limit=20) + assert total == 1 + mock_repo.list_by_user.assert_called_once_with( + "user_1", status="ready", skip=0, limit=50 + ) + mock_repo.count_by_user.assert_called_once_with("user_1", status="ready") - def test_list_empty(self, mock_repo): - """测试空列表""" + def test_list_with_pagination(self, mock_repo, sample_item): + """带分页参数""" + mock_repo.list_by_user.return_value = [sample_item] + mock_repo.count_by_user.return_value = 10 + use_case = ListVoiceLibraryUseCase(mock_repo) + + use_case.execute("user_1", skip=10, limit=20) + + mock_repo.list_by_user.assert_called_once_with( + "user_1", status=None, skip=10, limit=20 + ) + + def test_empty_list(self, mock_repo): + """空列表""" mock_repo.list_by_user.return_value = [] mock_repo.count_by_user.return_value = 0 - use_case = ListVoiceLibraryUseCase(repository=mock_repo) + use_case = ListVoiceLibraryUseCase(mock_repo) - items, total = use_case.execute("user-001") + items, total = use_case.execute("user_1") assert items == [] assert total == 0 -# =========================================================================== -# 7. voice_id 与 id 字段隔离专项回归测试 -# =========================================================================== +class TestGetVoiceLibraryUseCase: + """GetVoiceLibraryUseCase 测试""" + + def test_get_existing(self, mock_repo, sample_item): + """获取存在的音色""" + mock_repo.get.return_value = sample_item + use_case = GetVoiceLibraryUseCase(mock_repo) + + result = use_case.execute("v1", "user_1") + + assert result is not None + assert result.id == "v1" + mock_repo.get.assert_called_once_with("v1", "user_1") + + def test_get_nonexistent_returns_none(self, mock_repo): + """获取不存在的返回 None""" + mock_repo.get.return_value = None + use_case = GetVoiceLibraryUseCase(mock_repo) + + result = use_case.execute("noexist", "user_1") + + assert result is None -class TestVoiceIdFieldIsolation: - """ - PR#74 P0 Bug 回归：voice_id 与 item.id 字段隔离 +class TestCreateVoiceLibraryUseCase: + """CreateVoiceLibraryUseCase 测试""" - 原 bug：API 路由层误将 command.id（item 主键）用作 voice_id， - 导致 voice_id 字段值错误。本测试类从 UseCase 层验证 - 这两个字段在整个 CRUD 生命周期中互不干扰。 - """ + def test_create_success(self, mock_repo, sample_item): + """创建成功""" + mock_repo.count_by_user.return_value = 0 + mock_repo.create.return_value = sample_item + use_case = CreateVoiceLibraryUseCase(mock_repo) - def test_create_id_and_voice_id_are_independent(self, create_use_case, mock_repo): - """创建时 id 自动生成，voice_id 来自 command""" command = CreateVoiceLibraryCommand( - user_id="user-001", - name="隔离测试", - voice_id="tts-voice-001", - voice_provider="openai", + user_id="user_1", + name="新音色", + text="你好", + voice_provider="cosyvoice", + voice_id="v_new", + voice_name="新音色名", + audio_url="https://oss.example.com/new.mp3", + duration=5.0, + file_size=51200, + status="processing", + project_id="", + tags=[], + metadata_={}, ) + result = use_case.execute(command, plan_name="free") - result = create_use_case.execute(command, plan_name="free") + assert result.id == "v1" + mock_repo.count_by_user.assert_called_once_with("user_1") + mock_repo.create.assert_called_once() + created = mock_repo.create.call_args[0][0] + assert isinstance(created, VoiceLibraryItem) + assert created.name == "新音色" - assert result.id != result.voice_id, "id 和 voice_id 应为不同值" - assert result.voice_id == "tts-voice-001" - assert len(result.id) == 32 # UUID hex + def test_create_quota_exceeded(self, mock_repo): + """超过配额抛出 QuotaExceededError""" + mock_repo.count_by_user.return_value = 9999 + use_case = CreateVoiceLibraryUseCase(mock_repo) - def test_update_voice_id_does_not_change_id(self, update_use_case, mock_repo): - """更新 voice_id 不影响 item 主键 id""" - existing = VoiceLibraryItem( - id="stable-id-001", - user_id="user-001", - name="测试", - voice_id="old-voice", + command = CreateVoiceLibraryCommand( + user_id="user_1", + name="超限音色", + text="text", + voice_provider="cosyvoice", + voice_id="v", + voice_name="v", + audio_url="url", + duration=1.0, + file_size=100, + status="ready", + project_id="", + tags=[], + metadata_={}, ) - mock_repo.get.return_value = existing + with pytest.raises(QuotaExceededError): + use_case.execute(command, plan_name="free") + + mock_repo.create.assert_not_called() + + +class TestUpdateVoiceLibraryUseCase: + """UpdateVoiceLibraryUseCase 测试""" + + def test_update_name(self, mock_repo, sample_item): + """更新名称""" + mock_repo.get.return_value = sample_item + mock_repo.update.side_effect = lambda x: x + use_case = UpdateVoiceLibraryUseCase(mock_repo) + + command = UpdateVoiceLibraryCommand(id="v1", user_id="user_1", name="新名字") + result = use_case.execute(command) + + assert result.name == "新名字" + # 其他字段不变 + assert result.voice_name == "温柔女声" + mock_repo.get.assert_called_once_with("v1", "user_1") + mock_repo.update.assert_called_once() + + def test_update_status(self, mock_repo, sample_item): + """更新状态""" + mock_repo.get.return_value = sample_item + mock_repo.update.side_effect = lambda x: x + use_case = UpdateVoiceLibraryUseCase(mock_repo) + + command = UpdateVoiceLibraryCommand(id="v1", user_id="user_1", status="failed") + result = use_case.execute(command) + + assert result.status == "failed" + + def test_update_multiple_fields(self, mock_repo, sample_item): + """同时更新多个字段""" + mock_repo.get.return_value = sample_item + mock_repo.update.side_effect = lambda x: x + use_case = UpdateVoiceLibraryUseCase(mock_repo) command = UpdateVoiceLibraryCommand( - id="stable-id-001", - user_id="user-001", - voice_id="new-voice-999", + id="v1", + user_id="user_1", + name="更新后", + duration=15.0, + tags=["新标签"], ) + result = use_case.execute(command) - result = update_use_case.execute(command) + assert result.name == "更新后" + assert result.duration == 15.0 + assert result.tags == ["新标签"] - assert result.id == "stable-id-001", "item 主键 id 不应改变" - assert result.voice_id == "new-voice-999", "voice_id 应被更新" + def test_update_nonexistent_raises(self, mock_repo): + """更新不存在的抛出 NotFoundError""" + mock_repo.get.return_value = None + use_case = UpdateVoiceLibraryUseCase(mock_repo) - def test_create_then_update_voice_id_preserves_id(self, create_use_case, update_use_case, mock_repo): - """创建后再更新 voice_id，id 始终不变""" - # 创建 - create_cmd = CreateVoiceLibraryCommand( - user_id="user-001", - name="生命周期测试", - voice_id="initial-voice", - ) - created = create_use_case.execute(create_cmd, plan_name="free") - original_id = created.id + command = UpdateVoiceLibraryCommand(id="noexist", user_id="user_1", name="x") + with pytest.raises(NotFoundError, match="not found"): + use_case.execute(command) - # 更新 - mock_repo.get.return_value = created - update_cmd = UpdateVoiceLibraryCommand( - id=original_id, - user_id="user-001", - voice_id="updated-voice", - ) - updated = update_use_case.execute(update_cmd) + mock_repo.update.assert_not_called() - assert updated.id == original_id, "经过创建和更新，id 应保持一致" - assert updated.voice_id == "updated-voice" - assert updated.voice_id != "initial-voice" + +class TestDeleteVoiceLibraryUseCase: + """DeleteVoiceLibraryUseCase 测试""" + + def test_delete_success(self, mock_repo): + """删除成功""" + mock_repo.delete.return_value = True + use_case = DeleteVoiceLibraryUseCase(mock_repo) + + result = use_case.execute("v1", "user_1") + + assert result is True + mock_repo.delete.assert_called_once_with("v1", "user_1") + + def test_delete_nonexistent_returns_false(self, mock_repo): + """删除不存在的返回 False""" + mock_repo.delete.return_value = False + use_case = DeleteVoiceLibraryUseCase(mock_repo) + + result = use_case.execute("noexist", "user_1") + + assert result is False