From 0a8bae07722c804927d72baa00be60d2161669a5 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 10:38:45 +0800 Subject: [PATCH] test(wave143): add voice_clone_profile unit tests (+82) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - VoiceCloneStatus 枚举: 5种状态值、字符串比较、从字符串构建 - TERMINAL_STATUSES: 3个终态验证 - create 工厂方法: 基础创建、各字段(description/source_audio/language/gender/max_retries/metadata)、strip处理 - 参数校验: 空user_id/空name/超长name/纯空白 - 属性: is_terminal(5种状态)、is_retryable(各种场景)、is_ready(4种场景) - transition_to: 合法转换(8条路径)、非法转换(4条)、字符串输入、updated_at更新 - mark_*方法: mark_processing/mark_ready/mark_failed/mark_disabled 各类场景 - prepare_retry: 正常重试、清除错误/voice_id、非failed不可重试、超上限不可重试 - to_dict: 字段完整性、值正确性、ready/failed状态、ISO时间格式 --- tests/unit/test_voice_clone_profile.py | 848 +++++++++++++++++-------- 1 file changed, 596 insertions(+), 252 deletions(-) mode change 100644 => 100755 tests/unit/test_voice_clone_profile.py diff --git a/tests/unit/test_voice_clone_profile.py b/tests/unit/test_voice_clone_profile.py old mode 100644 new mode 100755 index 8f1840c2b..bbaa35ae3 --- a/tests/unit/test_voice_clone_profile.py +++ b/tests/unit/test_voice_clone_profile.py @@ -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) -- 2.54.0