Compare commits

...

1 Commits

Author SHA1 Message Date
CI Bot c864c9b2e1 test: P3-1 第35波单元测试
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 9s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m31s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m3s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 28s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m9s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 11s
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 25s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 52s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 20s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m47s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m15s
AI Code Review / AI Code Review (pull_request) Successful in 3m20s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 3m6s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 3m59s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 42s
- voice_library use case: 19个测试(列表带count/获取/创建+配额/更新/删除)
- voice_clone use case: 21个测试(创建/列表/获取/状态查询/删除/重试/权限校验)
- tts_job use case: 22个测试(创建/列表/获取/状态查询/删除/权限校验)

合计+62
2026-07-24 10:59:58 +08:00
3 changed files with 781 additions and 629 deletions
+270
View File
@@ -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()
+300
View File
@@ -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")
+211 -629
View File
@@ -1,14 +1,8 @@
"""
配音库(Voice LibraryUse 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.idUUID)不应与 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 测试"""
原 bugAPI 路由层误将 command.iditem 主键)用作 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