Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c864c9b2e1 |
Executable
+270
@@ -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()
|
||||
Executable
+300
@@ -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")
|
||||
Regular → Executable
+211
-629
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user