xiaoxia
005b34d0dd
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
651 lines
24 KiB
Python
Executable File
651 lines
24 KiB
Python
Executable File
"""VoiceCloneProfile 领域模型单元测试."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
from time import sleep
|
||
|
||
import pytest
|
||
|
||
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 工厂方法测试."""
|
||
|
||
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
|
||
|
||
def test_create_with_description(self):
|
||
"""带描述创建."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="T", description=" 测试描述 ")
|
||
assert p.description == "测试描述" # strip了
|
||
|
||
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
|
||
|
||
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_gender_normalized(self):
|
||
"""性别自动转小写."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="T", gender="Male")
|
||
assert p.gender == "male"
|
||
|
||
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_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_metadata_none(self):
|
||
"""metadata为None时默认为空dict."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="T", metadata=None)
|
||
assert p.metadata == {}
|
||
|
||
def test_create_user_id_stripped(self):
|
||
"""user_id去除空白."""
|
||
p = VoiceCloneProfile.create(user_id=" user123 ", name="T")
|
||
assert p.user_id == "user123"
|
||
|
||
def test_create_name_stripped(self):
|
||
"""name去除空白."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name=" 我的音色 ")
|
||
assert p.name == "我的音色"
|
||
|
||
def test_create_empty_user_id(self):
|
||
"""空user_id抛错."""
|
||
with pytest.raises(ValueError, match="user_id"):
|
||
VoiceCloneProfile.create(user_id="", name="T")
|
||
|
||
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 TestVoiceCloneProfileProperties:
|
||
"""VoiceCloneProfile 属性测试."""
|
||
|
||
def test_is_terminal_pending(self):
|
||
"""pending不是终态."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||
assert p.is_terminal is False
|
||
|
||
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_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_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_is_terminal_disabled(self):
|
||
"""disabled是终态."""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="T")
|
||
p.mark_disabled()
|
||
assert p.is_terminal is True
|
||
|
||
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_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="非法状态转换"):
|
||
p.transition_to(VoiceCloneStatus.READY)
|
||
|
||
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="无效状态"):
|
||
p.transition_to("invalid")
|
||
|
||
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 TestMarkMethods:
|
||
"""mark_* 系列方法测试."""
|
||
|
||
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="不可重试"):
|
||
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="不可重试"):
|
||
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 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={"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"}
|
||
|
||
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
|
||
|
||
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_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"])
|
||
|
||
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)
|