test(wave143): voice_clone_profile 单测补全 +82 #1066
Regular → Executable
+596
-252
@@ -1,306 +1,650 @@
|
||||
"""VoiceCloneProfile 领域模型单元测试 — Phase 3 CosyVoice 集成."""
|
||||
"""VoiceCloneProfile 领域模型单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from time import sleep
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
|
||||
from packages.domain.voice_clone_profile import (
|
||||
TERMINAL_STATUSES,
|
||||
VoiceCloneProfile,
|
||||
VoiceCloneStatus,
|
||||
)
|
||||
|
||||
# ── 枚举测试 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVoiceCloneStatus:
|
||||
"""VoiceCloneStatus 枚举测试."""
|
||||
|
||||
def test_status_values(self):
|
||||
"""状态值正确."""
|
||||
assert VoiceCloneStatus.PENDING.value == "pending"
|
||||
assert VoiceCloneStatus.PROCESSING.value == "processing"
|
||||
assert VoiceCloneStatus.READY.value == "ready"
|
||||
assert VoiceCloneStatus.FAILED.value == "failed"
|
||||
assert VoiceCloneStatus.DISABLED.value == "disabled"
|
||||
|
||||
def test_status_count(self):
|
||||
"""共5种状态."""
|
||||
assert len(VoiceCloneStatus) == 5
|
||||
|
||||
def test_is_str_enum(self):
|
||||
"""是StrEnum，可与字符串直接比较."""
|
||||
assert VoiceCloneStatus.PENDING == "pending"
|
||||
assert VoiceCloneStatus.READY + "" == "ready"
|
||||
|
||||
def test_from_string(self):
|
||||
"""从字符串构建枚举."""
|
||||
assert VoiceCloneStatus("pending") == VoiceCloneStatus.PENDING
|
||||
assert VoiceCloneStatus("ready") == VoiceCloneStatus.READY
|
||||
|
||||
def test_from_string_invalid(self):
|
||||
"""无效字符串抛出ValueError."""
|
||||
with pytest.raises(ValueError):
|
||||
VoiceCloneStatus("invalid_status")
|
||||
|
||||
|
||||
# ── 终态集合测试 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTerminalStatuses:
|
||||
"""TERMINAL_STATUSES 终态集合测试."""
|
||||
|
||||
def test_ready_is_terminal(self):
|
||||
"""ready是终态."""
|
||||
assert VoiceCloneStatus.READY in TERMINAL_STATUSES
|
||||
|
||||
def test_failed_is_terminal(self):
|
||||
"""failed是终态."""
|
||||
assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES
|
||||
|
||||
def test_disabled_is_terminal(self):
|
||||
"""disabled是终态."""
|
||||
assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES
|
||||
|
||||
def test_pending_not_terminal(self):
|
||||
"""pending不是终态."""
|
||||
assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES
|
||||
|
||||
def test_processing_not_terminal(self):
|
||||
"""processing不是终态."""
|
||||
assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES
|
||||
|
||||
def test_terminal_count(self):
|
||||
"""共3个终态."""
|
||||
assert len(TERMINAL_STATUSES) == 3
|
||||
|
||||
|
||||
# ── 工厂方法测试 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVoiceCloneProfileCreate:
|
||||
"""测试 VoiceCloneProfile.create() 工厂方法。"""
|
||||
"""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",
|
||||
)
|
||||
def test_create_basic(self):
|
||||
"""基础创建."""
|
||||
p = VoiceCloneProfile.create(user_id="user123", name="我的音色")
|
||||
assert p.id # 自动生成
|
||||
assert p.user_id == "user123"
|
||||
assert p.name == "我的音色"
|
||||
assert p.status == VoiceCloneStatus.PENDING
|
||||
assert p.retry_count == 0
|
||||
assert p.max_retries == 3
|
||||
|
||||
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_with_description(self):
|
||||
"""带描述创建."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", description=" 测试描述 ")
|
||||
assert p.description == "测试描述" # strip了
|
||||
|
||||
def test_create_minimal(self) -> None:
|
||||
"""使用最小参数创建。"""
|
||||
profile = VoiceCloneProfile.create(user_id="user_001", name="测试音色")
|
||||
def test_create_with_source_audio(self):
|
||||
"""带源音频URL创建."""
|
||||
url = "https://example.com/audio.wav"
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", source_audio_url=url)
|
||||
assert p.source_audio_url == url
|
||||
|
||||
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_with_language(self):
|
||||
"""指定语言."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", language="en-US")
|
||||
assert p.language == "en-US"
|
||||
|
||||
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_gender_normalized(self):
|
||||
"""性别自动转小写."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", gender="Male")
|
||||
assert p.gender == "male"
|
||||
|
||||
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_custom_max_retries(self):
|
||||
"""自定义最大重试次数."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", max_retries=5)
|
||||
assert p.max_retries == 5
|
||||
|
||||
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_metadata(self):
|
||||
"""元数据."""
|
||||
meta = {"age": 30, "accent": "北方"}
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", metadata=meta)
|
||||
assert p.metadata == meta
|
||||
# 不是同一个对象引用（深拷贝？）
|
||||
assert p.metadata is not meta or p.metadata == meta
|
||||
|
||||
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_metadata_none(self):
|
||||
"""metadata为None时默认为空dict."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", metadata=None)
|
||||
assert p.metadata == {}
|
||||
|
||||
def test_create_strips_whitespace(self) -> None:
|
||||
"""应去除首尾空白。"""
|
||||
profile = VoiceCloneProfile.create(
|
||||
user_id=" user_001 ",
|
||||
name=" 测试音色 ",
|
||||
description=" 描述 ",
|
||||
)
|
||||
def test_create_user_id_stripped(self):
|
||||
"""user_id去除空白."""
|
||||
p = VoiceCloneProfile.create(user_id=" user123 ", name="T")
|
||||
assert p.user_id == "user123"
|
||||
|
||||
assert profile.user_id == "user_001"
|
||||
assert profile.name == "测试音色"
|
||||
assert profile.description == "描述"
|
||||
def test_create_name_stripped(self):
|
||||
"""name去除空白."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name=" 我的音色 ")
|
||||
assert p.name == "我的音色"
|
||||
|
||||
def test_create_gender_normalized(self) -> None:
|
||||
"""gender 应转换为小写。"""
|
||||
profile = VoiceCloneProfile.create(
|
||||
user_id="user_001",
|
||||
name="测试",
|
||||
gender="FEMALE",
|
||||
)
|
||||
def test_create_empty_user_id(self):
|
||||
"""空user_id抛错."""
|
||||
with pytest.raises(ValueError, match="user_id"):
|
||||
VoiceCloneProfile.create(user_id="", name="T")
|
||||
|
||||
assert profile.gender == "female"
|
||||
def test_create_whitespace_user_id(self):
|
||||
"""纯空白user_id抛错."""
|
||||
with pytest.raises(ValueError, match="user_id"):
|
||||
VoiceCloneProfile.create(user_id=" ", name="T")
|
||||
|
||||
def test_create_empty_name(self):
|
||||
"""空name抛错."""
|
||||
with pytest.raises(ValueError, match="name"):
|
||||
VoiceCloneProfile.create(user_id="u1", name="")
|
||||
|
||||
def test_create_whitespace_name(self):
|
||||
"""纯空白name抛错."""
|
||||
with pytest.raises(ValueError, match="name"):
|
||||
VoiceCloneProfile.create(user_id="u1", name=" ")
|
||||
|
||||
def test_create_name_too_long(self):
|
||||
"""name超过100字符抛错."""
|
||||
long_name = "a" * 101
|
||||
with pytest.raises(ValueError, match="100"):
|
||||
VoiceCloneProfile.create(user_id="u1", name=long_name)
|
||||
|
||||
def test_create_name_exactly_100(self):
|
||||
"""name恰好100字符正常."""
|
||||
name = "a" * 100
|
||||
p = VoiceCloneProfile.create(user_id="u1", name=name)
|
||||
assert p.name == name
|
||||
|
||||
def test_create_has_created_at(self):
|
||||
"""创建后有created_at时间戳."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert isinstance(p.created_at, datetime)
|
||||
assert p.created_at.tzinfo is not None # 有时区
|
||||
|
||||
def test_create_has_updated_at(self):
|
||||
"""创建后有updated_at时间戳."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert isinstance(p.updated_at, datetime)
|
||||
|
||||
def test_create_id_is_hex(self):
|
||||
"""id是32位hex字符串(uuid4 hex)."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert len(p.id) == 32
|
||||
# 全部是hex字符
|
||||
int(p.id, 16) # 不抛错就是hex
|
||||
|
||||
|
||||
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:
|
||||
"""测试状态转换。"""
|
||||
class TestVoiceCloneProfileProperties:
|
||||
"""VoiceCloneProfile 属性测试."""
|
||||
|
||||
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_is_terminal_pending(self):
|
||||
"""pending不是终态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert p.is_terminal is False
|
||||
|
||||
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_is_terminal_processing(self):
|
||||
"""processing不是终态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
assert p.is_terminal is False
|
||||
|
||||
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_is_terminal_ready(self):
|
||||
"""ready是终态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("voice_001")
|
||||
assert p.is_terminal is True
|
||||
|
||||
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_is_terminal_failed(self):
|
||||
"""failed是终态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("超时")
|
||||
assert p.is_terminal is True
|
||||
|
||||
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_is_terminal_disabled(self):
|
||||
"""disabled是终态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_disabled()
|
||||
assert p.is_terminal is True
|
||||
|
||||
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_is_retryable_failed_within_limit(self):
|
||||
"""失败且未超过重试次数，可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", max_retries=3)
|
||||
p.mark_processing()
|
||||
p.mark_failed("error")
|
||||
assert p.is_retryable is True
|
||||
|
||||
def test_invalid_transition_raises(self) -> None:
|
||||
"""非法状态转换应抛出 ValueError。"""
|
||||
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
||||
def test_is_retryable_failed_at_limit(self):
|
||||
"""失败但已达重试上限，不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", max_retries=3)
|
||||
p.mark_processing()
|
||||
p.mark_failed("e1")
|
||||
p.prepare_retry() # retry_count=1
|
||||
p.mark_processing()
|
||||
p.mark_failed("e2")
|
||||
p.prepare_retry() # retry_count=2
|
||||
p.mark_processing()
|
||||
p.mark_failed("e3")
|
||||
p.prepare_retry() # retry_count=3
|
||||
p.mark_processing()
|
||||
p.mark_failed("e4") # retry_count=3, max=3
|
||||
assert p.is_retryable is False
|
||||
|
||||
def test_is_retryable_pending(self):
|
||||
"""pending状态不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert p.is_retryable is False
|
||||
|
||||
def test_is_retryable_ready(self):
|
||||
"""ready状态不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("v1")
|
||||
assert p.is_retryable is False
|
||||
|
||||
def test_is_retryable_processing(self):
|
||||
"""processing状态不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
assert p.is_retryable is False
|
||||
|
||||
def test_is_ready_with_voice_id(self):
|
||||
"""ready状态且有voice_id，is_ready为True."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("voice_001")
|
||||
assert p.is_ready is True
|
||||
|
||||
def test_is_ready_no_voice_id(self):
|
||||
"""ready状态但无voice_id，is_ready为False."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.status = VoiceCloneStatus.READY # 手动设为ready但无voice_id
|
||||
p.voice_id = ""
|
||||
assert p.is_ready is False
|
||||
|
||||
def test_is_ready_pending(self):
|
||||
"""pending状态is_ready为False."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
assert p.is_ready is False
|
||||
|
||||
def test_is_ready_failed(self):
|
||||
"""failed状态is_ready为False."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("err")
|
||||
assert p.is_ready is False
|
||||
|
||||
|
||||
# ── 状态转换测试 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTransitionTo:
|
||||
"""transition_to 状态转换测试."""
|
||||
|
||||
def test_pending_to_processing(self):
|
||||
"""pending → processing 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||||
assert p.status == VoiceCloneStatus.PROCESSING
|
||||
|
||||
def test_pending_to_failed(self):
|
||||
"""pending → failed 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.transition_to(VoiceCloneStatus.FAILED)
|
||||
assert p.status == VoiceCloneStatus.FAILED
|
||||
|
||||
def test_pending_to_disabled(self):
|
||||
"""pending → disabled 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
def test_pending_to_ready_invalid(self):
|
||||
"""pending → ready 非法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
with pytest.raises(ValueError, match="非法状态转换"):
|
||||
profile.mark_ready(voice_id="voice_001") # PENDING → READY 非法
|
||||
p.transition_to(VoiceCloneStatus.READY)
|
||||
|
||||
def test_invalid_status_string_raises(self) -> None:
|
||||
"""无效状态字符串应抛出 ValueError。"""
|
||||
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
||||
def test_processing_to_ready(self):
|
||||
"""processing → ready 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.transition_to(VoiceCloneStatus.READY)
|
||||
assert p.status == VoiceCloneStatus.READY
|
||||
|
||||
def test_processing_to_failed(self):
|
||||
"""processing → failed 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.transition_to(VoiceCloneStatus.FAILED)
|
||||
assert p.status == VoiceCloneStatus.FAILED
|
||||
|
||||
def test_processing_to_disabled(self):
|
||||
"""processing → disabled 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
def test_failed_to_pending(self):
|
||||
"""failed → pending 合法（重试）."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("err")
|
||||
p.transition_to(VoiceCloneStatus.PENDING)
|
||||
assert p.status == VoiceCloneStatus.PENDING
|
||||
|
||||
def test_failed_to_ready_invalid(self):
|
||||
"""failed → ready 非法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("err")
|
||||
with pytest.raises(ValueError):
|
||||
p.transition_to(VoiceCloneStatus.READY)
|
||||
|
||||
def test_ready_to_disabled(self):
|
||||
"""ready → disabled 合法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("v1")
|
||||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
def test_disabled_to_pending_invalid(self):
|
||||
"""disabled → pending 非法."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_disabled()
|
||||
with pytest.raises(ValueError):
|
||||
p.transition_to(VoiceCloneStatus.PENDING)
|
||||
|
||||
def test_transition_with_string(self):
|
||||
"""字符串输入的状态转换."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.transition_to("processing")
|
||||
assert p.status == VoiceCloneStatus.PROCESSING
|
||||
|
||||
def test_transition_with_invalid_string(self):
|
||||
"""无效字符串状态抛错."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
with pytest.raises(ValueError, match="无效状态"):
|
||||
profile.transition_to("invalid_status")
|
||||
p.transition_to("invalid")
|
||||
|
||||
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
|
||||
def test_transition_updates_updated_at(self):
|
||||
"""状态转换更新updated_at."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
old_updated = p.updated_at
|
||||
sleep(0.01)
|
||||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||||
assert p.updated_at > old_updated
|
||||
|
||||
def test_transition_error_message_contains_statuses(self):
|
||||
"""错误信息包含源状态和目标状态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
p.transition_to(VoiceCloneStatus.READY)
|
||||
msg = str(exc_info.value)
|
||||
assert "pending" in msg
|
||||
assert "ready" in msg
|
||||
|
||||
|
||||
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 == ""
|
||||
class TestMarkMethods:
|
||||
"""mark_* 系列方法测试."""
|
||||
|
||||
def test_prepare_retry_not_failed_raises(self) -> None:
|
||||
"""非 FAILED 状态重试应抛出 ValueError。"""
|
||||
profile = VoiceCloneProfile.create(user_id="user_001", name="测试")
|
||||
def test_mark_processing_clears_error(self):
|
||||
"""mark_processing 清除错误信息."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.error_message = "previous error"
|
||||
p.mark_processing()
|
||||
assert p.error_message == ""
|
||||
|
||||
def test_mark_processing_from_pending(self):
|
||||
"""从pending标记为processing."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
assert p.status == VoiceCloneStatus.PROCESSING
|
||||
|
||||
def test_mark_ready_with_voice_id(self):
|
||||
"""mark_ready 正常标记."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("voice_001")
|
||||
assert p.status == VoiceCloneStatus.READY
|
||||
assert p.voice_id == "voice_001"
|
||||
|
||||
def test_mark_ready_clears_error(self):
|
||||
"""mark_ready 清除错误信息."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.error_message = "some error"
|
||||
p.mark_ready("v1")
|
||||
assert p.error_message == ""
|
||||
|
||||
def test_mark_ready_empty_voice_id(self):
|
||||
"""空voice_id抛错."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
with pytest.raises(ValueError, match="voice_id"):
|
||||
p.mark_ready("")
|
||||
|
||||
def test_mark_ready_whitespace_voice_id(self):
|
||||
"""纯空白voice_id抛错."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
with pytest.raises(ValueError):
|
||||
p.mark_ready(" ")
|
||||
|
||||
def test_mark_ready_strips_voice_id(self):
|
||||
"""voice_id去除空白."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready(" voice_001 ")
|
||||
assert p.voice_id == "voice_001"
|
||||
|
||||
def test_mark_failed_sets_error(self):
|
||||
"""mark_failed 设置错误信息."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("连接超时")
|
||||
assert p.error_message == "连接超时"
|
||||
|
||||
def test_mark_failed_from_pending(self):
|
||||
"""从pending直接失败."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_failed("验证失败")
|
||||
assert p.status == VoiceCloneStatus.FAILED
|
||||
assert p.error_message == "验证失败"
|
||||
|
||||
def test_mark_disabled_from_pending(self):
|
||||
"""从pending禁用."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_disabled()
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
def test_mark_disabled_from_ready(self):
|
||||
"""从ready禁用."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("v1")
|
||||
p.mark_disabled()
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
|
||||
# ── 重试逻辑测试 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPrepareRetry:
|
||||
"""prepare_retry 重试逻辑测试."""
|
||||
|
||||
def test_prepare_retry_basic(self):
|
||||
"""基础重试成功."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", max_retries=3)
|
||||
p.mark_processing()
|
||||
p.mark_failed("err")
|
||||
p.prepare_retry()
|
||||
assert p.status == VoiceCloneStatus.PENDING
|
||||
assert p.retry_count == 1
|
||||
|
||||
def test_prepare_retry_clears_error(self):
|
||||
"""重试清除错误信息."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("big error")
|
||||
p.prepare_retry()
|
||||
assert p.error_message == ""
|
||||
|
||||
def test_prepare_retry_clears_voice_id(self):
|
||||
"""重试清除voice_id."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.voice_id = "old_voice"
|
||||
p.mark_processing()
|
||||
p.mark_failed("err")
|
||||
p.prepare_retry()
|
||||
assert p.voice_id == ""
|
||||
|
||||
def test_prepare_retry_not_failed(self):
|
||||
"""非failed状态不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
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("第二次失败")
|
||||
p.prepare_retry()
|
||||
|
||||
def test_prepare_retry_exceeds_max(self):
|
||||
"""超过最大重试次数不可重试."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T", max_retries=1)
|
||||
p.mark_processing()
|
||||
p.mark_failed("e1")
|
||||
p.prepare_retry() # retry_count=1
|
||||
p.mark_processing()
|
||||
p.mark_failed("e2")
|
||||
with pytest.raises(ValueError, match="不可重试"):
|
||||
profile.prepare_retry()
|
||||
p.prepare_retry()
|
||||
|
||||
def test_prepare_retry_error_has_details(self):
|
||||
"""错误信息包含详细状态."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
p.prepare_retry()
|
||||
msg = str(exc_info.value)
|
||||
assert "pending" in msg
|
||||
assert "retry_count" in msg
|
||||
assert "max_retries" in msg
|
||||
|
||||
|
||||
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",
|
||||
|
||||
class TestToDict:
|
||||
"""to_dict 序列化测试."""
|
||||
|
||||
def test_to_dict_keys(self):
|
||||
"""序列化字典包含所有预期字段."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="测试音色")
|
||||
d = p.to_dict()
|
||||
expected_keys = {
|
||||
"id",
|
||||
"user_id",
|
||||
"name",
|
||||
"description",
|
||||
"status",
|
||||
"source_audio_url",
|
||||
"voice_id",
|
||||
"voice_model",
|
||||
"language",
|
||||
"gender",
|
||||
"error_message",
|
||||
"retry_count",
|
||||
"max_retries",
|
||||
"is_retryable",
|
||||
"is_ready",
|
||||
"metadata",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
}
|
||||
assert set(d.keys()) == expected_keys
|
||||
|
||||
def test_to_dict_values(self):
|
||||
"""序列化值正确."""
|
||||
p = VoiceCloneProfile.create(
|
||||
user_id="user123",
|
||||
name="我的音色",
|
||||
description="测试用",
|
||||
language="zh-CN",
|
||||
gender="female",
|
||||
max_retries=5,
|
||||
metadata={"key": "value"},
|
||||
metadata={"source": "upload"},
|
||||
)
|
||||
d = p.to_dict()
|
||||
assert d["user_id"] == "user123"
|
||||
assert d["name"] == "我的音色"
|
||||
assert d["description"] == "测试用"
|
||||
assert d["status"] == "pending"
|
||||
assert d["language"] == "zh-CN"
|
||||
assert d["gender"] == "female"
|
||||
assert d["retry_count"] == 0
|
||||
assert d["max_retries"] == 5
|
||||
assert d["is_retryable"] is False
|
||||
assert d["is_ready"] is False
|
||||
assert d["metadata"] == {"source": "upload"}
|
||||
|
||||
result = profile.to_dict()
|
||||
def test_to_dict_ready_status(self):
|
||||
"""ready状态下序列化正确."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_ready("voice_001")
|
||||
d = p.to_dict()
|
||||
assert d["status"] == "ready"
|
||||
assert d["voice_id"] == "voice_001"
|
||||
assert d["is_ready"] is True
|
||||
assert d["is_retryable"] is False
|
||||
|
||||
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_failed_status(self):
|
||||
"""failed状态下序列化正确."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
p.mark_failed("超时错误")
|
||||
d = p.to_dict()
|
||||
assert d["status"] == "failed"
|
||||
assert d["error_message"] == "超时错误"
|
||||
assert d["is_retryable"] is True
|
||||
|
||||
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")
|
||||
def test_to_dict_datetime_format(self):
|
||||
"""时间字段是ISO格式字符串."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
d = p.to_dict()
|
||||
# ISO格式可以被datetime解析
|
||||
datetime.fromisoformat(d["created_at"])
|
||||
datetime.fromisoformat(d["updated_at"])
|
||||
|
||||
result = profile.to_dict()
|
||||
|
||||
assert result["status"] == "ready"
|
||||
assert result["voice_id"] == "voice_001"
|
||||
assert result["is_ready"] is True
|
||||
def test_to_dict_with_updated_at_after_transition(self):
|
||||
"""状态转换后updated_at被序列化."""
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||||
p.mark_processing()
|
||||
d = p.to_dict()
|
||||
assert d["updated_at"] is not None
|
||||
assert isinstance(d["updated_at"], str)
|
||||
|
||||
Reference in New Issue
Block a user