xiaoxia
502 lines
17 KiB
Python
Executable File
502 lines
17 KiB
Python
Executable File
"""VoiceCloneProfile 音色克隆档案单测.
|
|
|
|
覆盖：状态枚举、create创建校验、状态机转换、标记方法、
|
|
重试逻辑、属性判断、to_dict序列化。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from packages.domain.voice_clone_profile import (
|
|
TERMINAL_STATUSES,
|
|
VoiceCloneProfile,
|
|
VoiceCloneStatus,
|
|
)
|
|
|
|
|
|
class TestVoiceCloneStatus:
|
|
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_is_str_enum(self):
|
|
assert isinstance(VoiceCloneStatus.PENDING, str)
|
|
assert VoiceCloneStatus.PENDING == "pending"
|
|
|
|
def test_terminal_statuses(self):
|
|
assert VoiceCloneStatus.READY in TERMINAL_STATUSES
|
|
assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES
|
|
assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES
|
|
assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES
|
|
assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES
|
|
|
|
|
|
class TestVoiceCloneProfileCreate:
|
|
def test_create_minimal(self):
|
|
profile = VoiceCloneProfile.create(user_id="user_1", name="我的音色")
|
|
assert profile.user_id == "user_1"
|
|
assert profile.name == "我的音色"
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.id and len(profile.id) == 32
|
|
|
|
def test_create_with_all_params(self):
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user_1",
|
|
name="甜美女声",
|
|
description="适合播客的女声",
|
|
source_audio_url="https://ex.com/source.wav",
|
|
voice_model="cosyvoice-300m",
|
|
language="en-US",
|
|
gender="female",
|
|
max_retries=5,
|
|
metadata={"source": "upload"},
|
|
)
|
|
assert profile.description == "适合播客的女声"
|
|
assert profile.source_audio_url == "https://ex.com/source.wav"
|
|
assert profile.voice_model == "cosyvoice-300m"
|
|
assert profile.language == "en-US"
|
|
assert profile.gender == "female"
|
|
assert profile.max_retries == 5
|
|
assert profile.metadata == {"source": "upload"}
|
|
|
|
def test_create_defaults(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="t")
|
|
assert profile.description == ""
|
|
assert profile.source_audio_url == ""
|
|
assert profile.voice_model == ""
|
|
assert profile.language == "zh-CN"
|
|
assert profile.gender == "unknown"
|
|
assert profile.max_retries == 3
|
|
assert profile.metadata == {}
|
|
assert profile.voice_id == ""
|
|
assert profile.error_message == ""
|
|
assert profile.retry_count == 0
|
|
|
|
def test_create_strips_whitespace(self):
|
|
profile = VoiceCloneProfile.create(
|
|
user_id=" user_1 ",
|
|
name=" 测试音色 ",
|
|
description=" desc ",
|
|
language=" en-US ",
|
|
gender=" MALE ",
|
|
)
|
|
assert profile.user_id == "user_1"
|
|
assert profile.name == "测试音色"
|
|
assert profile.description == "desc"
|
|
assert profile.language == "en-US"
|
|
assert profile.gender == "male" # lowercased
|
|
|
|
def test_create_gender_lowercased(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="t", gender="Female")
|
|
assert profile.gender == "female"
|
|
|
|
def test_create_empty_user_id_raises(self):
|
|
try:
|
|
VoiceCloneProfile.create(user_id="", name="t")
|
|
except ValueError as e:
|
|
assert "user_id" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_create_whitespace_user_id_raises(self):
|
|
try:
|
|
VoiceCloneProfile.create(user_id=" ", name="t")
|
|
except ValueError as e:
|
|
assert "user_id" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_create_empty_name_raises(self):
|
|
try:
|
|
VoiceCloneProfile.create(user_id="u1", name="")
|
|
except ValueError as e:
|
|
assert "name" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_create_name_too_long_raises(self):
|
|
long_name = "a" * 101
|
|
try:
|
|
VoiceCloneProfile.create(user_id="u1", name=long_name)
|
|
except ValueError as e:
|
|
assert "100" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_create_name_exactly_100_ok(self):
|
|
name = "a" * 100
|
|
profile = VoiceCloneProfile.create(user_id="u1", name=name)
|
|
assert profile.name == name
|
|
|
|
def test_create_sets_created_at(self):
|
|
before = datetime.now(timezone.utc)
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="t")
|
|
after = datetime.now(timezone.utc)
|
|
assert before <= profile.created_at <= after
|
|
assert before <= profile.updated_at <= after
|
|
|
|
def test_create_metadata_none_defaults_to_empty(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="t", metadata=None)
|
|
assert profile.metadata == {}
|
|
|
|
|
|
class TestStatusProperties:
|
|
def test_is_terminal_pending(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
assert p.is_terminal is False
|
|
|
|
def test_is_terminal_processing(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
assert p.is_terminal is False
|
|
|
|
def test_is_terminal_ready(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("voice_123")
|
|
assert p.is_terminal is True
|
|
|
|
def test_is_terminal_failed(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_failed("error")
|
|
assert p.is_terminal is True
|
|
|
|
def test_is_terminal_disabled_from_pending(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_disabled()
|
|
assert p.is_terminal is True
|
|
|
|
def test_is_retryable_failed_within_limit(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=3)
|
|
p.mark_processing()
|
|
p.mark_failed("err")
|
|
assert p.is_retryable is True
|
|
|
|
def test_is_retryable_failed_at_limit(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=1)
|
|
p.mark_processing()
|
|
p.mark_failed("err1")
|
|
p.prepare_retry()
|
|
p.mark_processing()
|
|
p.mark_failed("err2")
|
|
assert p.is_retryable is False
|
|
|
|
def test_is_retryable_pending(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
assert p.is_retryable is False
|
|
|
|
def test_is_retryable_ready(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("v1")
|
|
assert p.is_retryable is False
|
|
|
|
def test_is_ready_success(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("voice_123")
|
|
assert p.is_ready is True
|
|
|
|
def test_is_ready_no_voice_id(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.status = VoiceCloneStatus.READY # 手动设状态，无voice_id
|
|
p.voice_id = ""
|
|
assert p.is_ready is False
|
|
|
|
|
|
class TestTransitionTo:
|
|
def test_pending_to_processing(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
assert p.status == VoiceCloneStatus.PROCESSING
|
|
|
|
def test_pending_to_failed(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.FAILED)
|
|
assert p.status == VoiceCloneStatus.FAILED
|
|
|
|
def test_pending_to_disabled(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert p.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_processing_to_ready(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
p.transition_to(VoiceCloneStatus.READY)
|
|
assert p.status == VoiceCloneStatus.READY
|
|
|
|
def test_processing_to_failed(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
p.transition_to(VoiceCloneStatus.FAILED)
|
|
assert p.status == VoiceCloneStatus.FAILED
|
|
|
|
def test_processing_to_disabled(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
p.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert p.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_failed_to_pending_retry(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
p.transition_to(VoiceCloneStatus.FAILED)
|
|
p.transition_to(VoiceCloneStatus.PENDING)
|
|
assert p.status == VoiceCloneStatus.PENDING
|
|
|
|
def test_ready_to_disabled(self):
|
|
"""已就绪音色可以被禁用."""
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
p.transition_to(VoiceCloneStatus.READY)
|
|
p.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert p.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_invalid_transition_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
try:
|
|
p.transition_to(VoiceCloneStatus.READY) # pending→ready 非法
|
|
except ValueError as e:
|
|
assert "非法状态转换" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_disabled_no_outgoing(self):
|
|
"""disabled状态不能转换到任何状态."""
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.DISABLED)
|
|
try:
|
|
p.transition_to(VoiceCloneStatus.PENDING)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_transition_with_string(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to("processing")
|
|
assert p.status == VoiceCloneStatus.PROCESSING
|
|
|
|
def test_transition_invalid_string_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
try:
|
|
p.transition_to("invalid_state")
|
|
except ValueError as e:
|
|
assert "无效状态" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_transition_updates_updated_at(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
old_updated = p.updated_at
|
|
import time
|
|
|
|
time.sleep(0.01)
|
|
p.transition_to(VoiceCloneStatus.PROCESSING)
|
|
assert p.updated_at > old_updated
|
|
|
|
|
|
class TestMarkMethods:
|
|
def test_mark_processing(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
assert p.status == VoiceCloneStatus.PROCESSING
|
|
assert p.error_message == ""
|
|
|
|
def test_mark_processing_clears_error(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.transition_to(VoiceCloneStatus.FAILED)
|
|
p.error_message = "old error"
|
|
p.retry_count = 1
|
|
p.transition_to(VoiceCloneStatus.PENDING)
|
|
p.mark_processing()
|
|
assert p.error_message == ""
|
|
|
|
def test_mark_ready_success(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("voice_id_123")
|
|
assert p.status == VoiceCloneStatus.READY
|
|
assert p.voice_id == "voice_id_123"
|
|
assert p.error_message == ""
|
|
|
|
def test_mark_ready_strips_whitespace(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready(" voice_456 ")
|
|
assert p.voice_id == "voice_456"
|
|
|
|
def test_mark_ready_empty_voice_id_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
try:
|
|
p.mark_ready("")
|
|
except ValueError as e:
|
|
assert "voice_id" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_mark_ready_whitespace_voice_id_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
try:
|
|
p.mark_ready(" ")
|
|
except ValueError as e:
|
|
assert "voice_id" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_mark_failed(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_failed("训练超时")
|
|
assert p.status == VoiceCloneStatus.FAILED
|
|
assert p.error_message == "训练超时"
|
|
|
|
def test_mark_disabled(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_disabled()
|
|
assert p.status == VoiceCloneStatus.DISABLED
|
|
|
|
|
|
class TestRetry:
|
|
def test_prepare_retry_success(self):
|
|
p = VoiceCloneProfile.create(user_id="u", 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
|
|
assert p.error_message == ""
|
|
assert p.voice_id == ""
|
|
|
|
def test_prepare_retry_multiple_times(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=3)
|
|
for i in range(3):
|
|
p.mark_processing()
|
|
p.mark_failed(f"err_{i}")
|
|
p.prepare_retry()
|
|
assert p.retry_count == i + 1
|
|
assert p.status == VoiceCloneStatus.PENDING
|
|
# 第4次应该失败
|
|
p.mark_processing()
|
|
p.mark_failed("err_3")
|
|
try:
|
|
p.prepare_retry()
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("expected ValueError on 4th retry")
|
|
|
|
def test_prepare_retry_not_failed_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
try:
|
|
p.prepare_retry()
|
|
except ValueError as e:
|
|
assert "不可重试" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_prepare_retry_ready_raises(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("v1")
|
|
try:
|
|
p.prepare_retry()
|
|
except ValueError as e:
|
|
assert "不可重试" in str(e)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
def test_prepare_retry_clears_voice_id(self):
|
|
p = VoiceCloneProfile.create(user_id="u", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("partial_id")
|
|
# 手动改到failed来测试
|
|
p.status = VoiceCloneStatus.FAILED
|
|
p.retry_count = 0
|
|
p.prepare_retry()
|
|
assert p.voice_id == ""
|
|
|
|
|
|
class TestToDict:
|
|
def test_to_dict_basic(self):
|
|
p = VoiceCloneProfile.create(user_id="u1", name="测试音色")
|
|
d = p.to_dict()
|
|
assert d["id"] == p.id
|
|
assert d["user_id"] == "u1"
|
|
assert d["name"] == "测试音色"
|
|
assert d["status"] == "pending"
|
|
assert d["is_retryable"] is False
|
|
assert d["is_ready"] is False
|
|
|
|
def test_to_dict_ready(self):
|
|
p = VoiceCloneProfile.create(user_id="u1", name="t")
|
|
p.mark_processing()
|
|
p.mark_ready("voice_42")
|
|
d = p.to_dict()
|
|
assert d["status"] == "ready"
|
|
assert d["voice_id"] == "voice_42"
|
|
assert d["is_ready"] is True
|
|
assert d["created_at"] is not None
|
|
assert d["updated_at"] is not None
|
|
|
|
def test_to_dict_failed_retryable(self):
|
|
p = VoiceCloneProfile.create(user_id="u1", name="t", max_retries=3)
|
|
p.mark_processing()
|
|
p.mark_failed("timeout")
|
|
d = p.to_dict()
|
|
assert d["status"] == "failed"
|
|
assert d["error_message"] == "timeout"
|
|
assert d["is_retryable"] is True
|
|
|
|
def test_to_dict_datetime_isoformat(self):
|
|
p = VoiceCloneProfile.create(user_id="u1", name="t")
|
|
d = p.to_dict()
|
|
parsed = datetime.fromisoformat(d["created_at"])
|
|
assert parsed.tzinfo is not None
|
|
|
|
def test_to_dict_metadata(self):
|
|
p = VoiceCloneProfile.create(user_id="u1", name="t", metadata={"key": "val", "num": 42})
|
|
d = p.to_dict()
|
|
assert d["metadata"] == {"key": "val", "num": 42}
|
|
|
|
def test_to_dict_all_fields_present(self):
|
|
p = VoiceCloneProfile.create(
|
|
user_id="u1",
|
|
name="t",
|
|
description="d",
|
|
source_audio_url="https://ex.com/s.wav",
|
|
voice_model="cosyvoice",
|
|
language="zh-CN",
|
|
gender="male",
|
|
)
|
|
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
|