灵应
f7c8b441d6
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 173h2m54s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 173h2m59s
Deploy / Deploy Staging (push) Failing after 173h22m24s
CI/CD Pipeline / Frontend Lint (push) Failing after 173h22m54s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 173h23m2s
Task 3.01: CosyVoice 配置 - 在 SharedSettings 中添加 CosyVoice 配置字段 - 更新 .env.example 和 .env.production.example Task 3.02: VoiceCloneProfile 领域模型 - 创建 VoiceCloneProfile 实体（音色克隆档案） - 状态机: pending → processing → ready/failed → disabled - 支持重试机制（retry_count/max_retries） - 创建 VoiceCloneProfileRepository 端口接口 Task 3.03: TTSJob 领域模型 - 创建 TTSJob 实体（TTS 任务） - 关联 VoiceCloneProfile（voice_clone_profile_id） - 状态机: pending → processing → completed/failed → cancelled - 支持重试机制 - 创建 TTSJobRepository 端口接口 单元测试: - VoiceCloneProfile: 31 个测试用例 - TTSJob: 34 个测试用例 - 全部 65 个测试通过 遵循六边形架构: Domain → Port → Adapter
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""VoiceCloneProfile 领域模型单元测试 — Phase 3 CosyVoice 集成."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
|
|
|
|
|
|
class TestVoiceCloneProfileCreate:
|
|
"""测试 VoiceCloneProfile.create() 工厂方法。"""
|
|
|
|
def test_create_success(self) -> None:
|
|
"""正常创建音色克隆档案。"""
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user_001",
|
|
name="我的音色",
|
|
description="用于配音的自定义音色",
|
|
source_audio_url="https://example.com/audio.wav",
|
|
voice_model="cosyvoice-v1",
|
|
language="zh-CN",
|
|
gender="female",
|
|
)
|
|
|
|
assert profile.id
|
|
assert profile.user_id == "user_001"
|
|
assert profile.name == "我的音色"
|
|
assert profile.description == "用于配音的自定义音色"
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.source_audio_url == "https://example.com/audio.wav"
|
|
assert profile.voice_model == "cosyvoice-v1"
|
|
assert profile.language == "zh-CN"
|
|
assert profile.gender == "female"
|
|
assert profile.retry_count == 0
|
|
assert profile.max_retries == 3
|
|
assert profile.created_at
|
|
assert profile.updated_at
|
|
|
|
def test_create_minimal(self) -> None:
|
|
"""使用最小参数创建。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试音色")
|
|
|
|
assert profile.user_id == "user_001"
|
|
assert profile.name == "测试音色"
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.description == ""
|
|
assert profile.source_audio_url == ""
|
|
assert profile.language == "zh-CN"
|
|
assert profile.gender == "unknown"
|
|
|
|
def test_create_empty_user_id_raises(self) -> None:
|
|
"""空 user_id 应抛出 ValueError。"""
|
|
with pytest.raises(ValueError, match="user_id 不能为空"):
|
|
VoiceCloneProfile.create(user_id="", name="测试")
|
|
|
|
def test_create_whitespace_user_id_raises(self) -> None:
|
|
"""空白 user_id 应抛出 ValueError。"""
|
|
with pytest.raises(ValueError, match="user_id 不能为空"):
|
|
VoiceCloneProfile.create(user_id=" ", name="测试")
|
|
|
|
def test_create_empty_name_raises(self) -> None:
|
|
"""空 name 应抛出 ValueError。"""
|
|
with pytest.raises(ValueError, match="name 不能为空"):
|
|
VoiceCloneProfile.create(user_id="user_001", name="")
|
|
|
|
def test_create_name_too_long_raises(self) -> None:
|
|
"""name 超过 100 字符应抛出 ValueError。"""
|
|
with pytest.raises(ValueError, match="name 长度不能超过 100 字符"):
|
|
VoiceCloneProfile.create(user_id="user_001", name="a" * 101)
|
|
|
|
def test_create_strips_whitespace(self) -> None:
|
|
"""应去除首尾空白。"""
|
|
profile = VoiceCloneProfile.create(
|
|
user_id=" user_001 ",
|
|
name=" 测试音色 ",
|
|
description=" 描述 ",
|
|
)
|
|
|
|
assert profile.user_id == "user_001"
|
|
assert profile.name == "测试音色"
|
|
assert profile.description == "描述"
|
|
|
|
def test_create_gender_normalized(self) -> None:
|
|
"""gender 应转换为小写。"""
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user_001",
|
|
name="测试",
|
|
gender="FEMALE",
|
|
)
|
|
|
|
assert profile.gender == "female"
|
|
|
|
|
|
class TestVoiceCloneProfileStatus:
|
|
"""测试状态相关属性和方法。"""
|
|
|
|
def test_initial_status_is_pending(self) -> None:
|
|
"""初始状态应为 PENDING。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
|
|
def test_is_terminal_pending(self) -> None:
|
|
"""PENDING 不是终态。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
assert not profile.is_terminal
|
|
|
|
def test_is_terminal_ready(self) -> None:
|
|
"""READY 是终态。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_ready(voice_id="voice_001")
|
|
assert profile.is_terminal
|
|
|
|
def test_is_terminal_failed(self) -> None:
|
|
"""FAILED 是终态。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_failed("克隆失败")
|
|
assert profile.is_terminal
|
|
|
|
def test_is_retryable_not_failed(self) -> None:
|
|
"""非 FAILED 状态不可重试。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
assert not profile.is_retryable
|
|
|
|
def test_is_retryable_failed_under_limit(self) -> None:
|
|
"""FAILED 且未超过重试上限时可重试。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_failed("克隆失败")
|
|
assert profile.is_retryable
|
|
|
|
def test_is_retryable_failed_over_limit(self) -> None:
|
|
"""超过重试上限时不可重试。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试", max_retries=1)
|
|
profile.mark_processing()
|
|
profile.mark_failed("第一次失败")
|
|
profile.prepare_retry()
|
|
profile.mark_processing()
|
|
profile.mark_failed("第二次失败")
|
|
assert not profile.is_retryable
|
|
|
|
def test_is_ready_with_voice_id(self) -> None:
|
|
"""READY 且有 voice_id 时应返回 True。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_ready(voice_id="voice_001")
|
|
assert profile.is_ready
|
|
|
|
def test_is_ready_without_voice_id(self) -> None:
|
|
"""READY 但无 voice_id 时应返回 False。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.status = VoiceCloneStatus.READY
|
|
profile.voice_id = ""
|
|
assert not profile.is_ready
|
|
|
|
|
|
class TestVoiceCloneProfileTransitions:
|
|
"""测试状态转换。"""
|
|
|
|
def test_mark_processing(self) -> None:
|
|
"""PENDING → PROCESSING 转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
assert profile.status == VoiceCloneStatus.PROCESSING
|
|
assert profile.error_message == ""
|
|
|
|
def test_mark_ready(self) -> None:
|
|
"""PROCESSING → READY 转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_ready(voice_id="voice_001")
|
|
assert profile.status == VoiceCloneStatus.READY
|
|
assert profile.voice_id == "voice_001"
|
|
assert profile.error_message == ""
|
|
|
|
def test_mark_ready_empty_voice_id_raises(self) -> None:
|
|
"""mark_ready 空 voice_id 应抛出 ValueError。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
with pytest.raises(ValueError, match="voice_id 不能为空"):
|
|
profile.mark_ready(voice_id="")
|
|
|
|
def test_mark_failed(self) -> None:
|
|
"""PROCESSING → FAILED 转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_failed("API 调用失败")
|
|
assert profile.status == VoiceCloneStatus.FAILED
|
|
assert profile.error_message == "API 调用失败"
|
|
|
|
def test_mark_disabled_from_pending(self) -> None:
|
|
"""PENDING → DISABLED 转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_disabled()
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_mark_disabled_from_ready(self) -> None:
|
|
"""READY → DISABLED 转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_ready(voice_id="voice_001")
|
|
profile.mark_disabled()
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_invalid_transition_raises(self) -> None:
|
|
"""非法状态转换应抛出 ValueError。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
profile.mark_ready(voice_id="voice_001") # PENDING → READY 非法
|
|
|
|
def test_invalid_status_string_raises(self) -> None:
|
|
"""无效状态字符串应抛出 ValueError。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
with pytest.raises(ValueError, match="无效状态"):
|
|
profile.transition_to("invalid_status")
|
|
|
|
def test_transition_to_with_string(self) -> None:
|
|
"""支持字符串形式的状态转换。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.transition_to("processing")
|
|
assert profile.status == VoiceCloneStatus.PROCESSING
|
|
|
|
|
|
class TestVoiceCloneProfileRetry:
|
|
"""测试重试逻辑。"""
|
|
|
|
def test_prepare_retry_success(self) -> None:
|
|
"""成功重试。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_failed("失败")
|
|
profile.prepare_retry()
|
|
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.retry_count == 1
|
|
assert profile.error_message == ""
|
|
assert profile.voice_id == ""
|
|
|
|
def test_prepare_retry_not_failed_raises(self) -> None:
|
|
"""非 FAILED 状态重试应抛出 ValueError。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
with pytest.raises(ValueError, match="不可重试"):
|
|
profile.prepare_retry()
|
|
|
|
def test_prepare_retry_over_limit_raises(self) -> None:
|
|
"""超过重试上限重试应抛出 ValueError。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试", max_retries=1)
|
|
profile.mark_processing()
|
|
profile.mark_failed("第一次失败")
|
|
profile.prepare_retry()
|
|
profile.mark_processing()
|
|
profile.mark_failed("第二次失败")
|
|
|
|
with pytest.raises(ValueError, match="不可重试"):
|
|
profile.prepare_retry()
|
|
|
|
|
|
class TestVoiceCloneProfileToDict:
|
|
"""测试序列化。"""
|
|
|
|
def test_to_dict_contains_all_fields(self) -> None:
|
|
"""to_dict 应包含所有字段。"""
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user_001",
|
|
name="测试音色",
|
|
description="描述",
|
|
source_audio_url="https://example.com/audio.wav",
|
|
voice_model="cosyvoice-v1",
|
|
language="zh-CN",
|
|
gender="female",
|
|
max_retries=5,
|
|
metadata={"key": "value"},
|
|
)
|
|
|
|
result = profile.to_dict()
|
|
|
|
assert result["id"] == profile.id
|
|
assert result["user_id"] == "user_001"
|
|
assert result["name"] == "测试音色"
|
|
assert result["description"] == "描述"
|
|
assert result["status"] == "pending"
|
|
assert result["source_audio_url"] == "https://example.com/audio.wav"
|
|
assert result["voice_model"] == "cosyvoice-v1"
|
|
assert result["language"] == "zh-CN"
|
|
assert result["gender"] == "female"
|
|
assert result["retry_count"] == 0
|
|
assert result["max_retries"] == 5
|
|
assert result["is_retryable"] is False
|
|
assert result["is_ready"] is False
|
|
assert result["metadata"] == {"key": "value"}
|
|
assert result["created_at"] is not None
|
|
assert result["updated_at"] is not None
|
|
|
|
def test_to_dict_after_state_change(self) -> None:
|
|
"""状态变更后 to_dict 应反映最新状态。"""
|
|
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
|
profile.mark_processing()
|
|
profile.mark_ready(voice_id="voice_001")
|
|
|
|
result = profile.to_dict()
|
|
|
|
assert result["status"] == "ready"
|
|
assert result["voice_id"] == "voice_001"
|
|
assert result["is_ready"] is True
|