xiaoxia
df38101bd9
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Successful in 40s
CI/CD Pipeline / Frontend Lint (push) Successful in 2m26s
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 3m35s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m55s
CI/CD Pipeline / Unit Tests (push) Successful in 4m6s
CI/CD Pipeline / Integration Tests (push) Successful in 1m45s
CI/CD Pipeline / Build Staging API Image (push) Successful in 12m21s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 15m52s
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 / ACR Image Cleanup (push) Has been cancelled
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
390 lines
16 KiB
Python
390 lines
16 KiB
Python
"""
|
|
VoiceCloneProfile 音色克隆档案领域模型单元测试
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from packages.domain.voice_clone_profile import (
|
|
TERMINAL_STATUSES,
|
|
VoiceCloneProfile,
|
|
VoiceCloneStatus,
|
|
)
|
|
|
|
|
|
class TestVoiceCloneStatus:
|
|
"""VoiceCloneStatus 枚举测试"""
|
|
|
|
def test_status_values(self):
|
|
assert VoiceCloneStatus.PENDING == "pending"
|
|
assert VoiceCloneStatus.PROCESSING == "processing"
|
|
assert VoiceCloneStatus.READY == "ready"
|
|
assert VoiceCloneStatus.FAILED == "failed"
|
|
assert VoiceCloneStatus.DISABLED == "disabled"
|
|
|
|
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:
|
|
"""VoiceCloneProfile.create 工厂方法测试"""
|
|
|
|
def test_create_minimal(self):
|
|
profile = VoiceCloneProfile.create(user_id="user123", name="我的音色")
|
|
assert profile.id is not None
|
|
assert len(profile.id) == 32 # uuid4 hex
|
|
assert profile.user_id == "user123"
|
|
assert profile.name == "我的音色"
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.retry_count == 0
|
|
assert profile.max_retries == 3
|
|
assert profile.gender == "unknown"
|
|
assert profile.language == "zh-CN"
|
|
assert profile.created_at is not None
|
|
assert profile.updated_at is not None
|
|
|
|
def test_create_with_all_fields(self):
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user456",
|
|
name="测试音色",
|
|
description="这是一个测试音色",
|
|
source_audio_url="https://example.com/audio.wav",
|
|
voice_model="cosyvoice-v2",
|
|
language="en-US",
|
|
gender="MALE",
|
|
max_retries=5,
|
|
metadata={"source": "upload"},
|
|
)
|
|
assert profile.user_id == "user456"
|
|
assert profile.name == "测试音色"
|
|
assert profile.description == "这是一个测试音色"
|
|
assert profile.source_audio_url == "https://example.com/audio.wav"
|
|
assert profile.voice_model == "cosyvoice-v2"
|
|
assert profile.language == "en-US"
|
|
assert profile.gender == "male" # 转小写
|
|
assert profile.max_retries == 5
|
|
assert profile.metadata == {"source": "upload"}
|
|
|
|
def test_create_strips_whitespace(self):
|
|
profile = VoiceCloneProfile.create(
|
|
user_id=" user789 ",
|
|
name=" 我的音色 ",
|
|
description=" 描述 ",
|
|
)
|
|
assert profile.user_id == "user789"
|
|
assert profile.name == "我的音色"
|
|
assert profile.description == "描述"
|
|
|
|
def test_create_empty_user_id_raises(self):
|
|
with pytest.raises(ValueError, match="user_id 不能为空"):
|
|
VoiceCloneProfile.create(user_id=" ", name="测试")
|
|
|
|
def test_create_empty_name_raises(self):
|
|
with pytest.raises(ValueError, match="name 不能为空"):
|
|
VoiceCloneProfile.create(user_id="user1", name=" ")
|
|
|
|
def test_create_name_too_long_raises(self):
|
|
long_name = "a" * 101
|
|
with pytest.raises(ValueError, match="name 长度不能超过 100 字符"):
|
|
VoiceCloneProfile.create(user_id="user1", name=long_name)
|
|
|
|
def test_create_name_exactly_100_chars_ok(self):
|
|
name = "a" * 100
|
|
profile = VoiceCloneProfile.create(user_id="user1", name=name)
|
|
assert profile.name == name
|
|
|
|
def test_create_default_metadata_is_dict(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
assert profile.metadata == {}
|
|
# 不应该共享同一个 dict
|
|
p2 = VoiceCloneProfile.create(user_id="u2", name="test2")
|
|
assert profile.metadata is not p2.metadata
|
|
|
|
|
|
class TestVoiceCloneProfileProperties:
|
|
"""属性测试"""
|
|
|
|
def test_is_terminal_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
assert profile.is_terminal is True
|
|
|
|
def test_is_terminal_failed(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
assert profile.is_terminal is True
|
|
|
|
def test_is_terminal_disabled(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.DISABLED
|
|
assert profile.is_terminal is True
|
|
|
|
def test_is_terminal_pending(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
assert profile.is_terminal is False
|
|
|
|
def test_is_terminal_processing(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
assert profile.is_terminal is False
|
|
|
|
def test_is_retryable_failed_within_limit(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.retry_count = 1
|
|
assert profile.is_retryable is True
|
|
|
|
def test_is_retryable_failed_at_limit(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.retry_count = 3
|
|
assert profile.is_retryable is False
|
|
|
|
def test_is_retryable_pending(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
assert profile.is_retryable is False
|
|
|
|
def test_is_ready_with_voice_id(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
profile.voice_id = "voice_123"
|
|
assert profile.is_ready is True
|
|
|
|
def test_is_ready_without_voice_id(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
profile.voice_id = ""
|
|
assert profile.is_ready is False
|
|
|
|
def test_is_ready_wrong_status(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.voice_id = "voice_123"
|
|
assert profile.is_ready is False # pending status
|
|
|
|
|
|
class TestStateTransitions:
|
|
"""状态转换测试"""
|
|
|
|
def test_pending_to_processing(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
old_updated = profile.updated_at
|
|
profile.transition_to(VoiceCloneStatus.PROCESSING)
|
|
assert profile.status == VoiceCloneStatus.PROCESSING
|
|
assert profile.updated_at >= old_updated
|
|
|
|
def test_pending_to_failed(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.transition_to(VoiceCloneStatus.FAILED)
|
|
assert profile.status == VoiceCloneStatus.FAILED
|
|
|
|
def test_pending_to_disabled(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_processing_to_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.transition_to(VoiceCloneStatus.READY)
|
|
assert profile.status == VoiceCloneStatus.READY
|
|
|
|
def test_processing_to_failed(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.transition_to(VoiceCloneStatus.FAILED)
|
|
assert profile.status == VoiceCloneStatus.FAILED
|
|
|
|
def test_processing_to_disabled(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_failed_to_pending_retry(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.transition_to(VoiceCloneStatus.PENDING)
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
|
|
def test_ready_to_disabled(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
profile.transition_to(VoiceCloneStatus.DISABLED)
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_transition_with_string(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.transition_to("processing")
|
|
assert profile.status == VoiceCloneStatus.PROCESSING
|
|
|
|
def test_transition_invalid_string_raises(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
with pytest.raises(ValueError, match="无效状态"):
|
|
profile.transition_to("invalid_status")
|
|
|
|
def test_invalid_transition_pending_to_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
profile.transition_to(VoiceCloneStatus.READY)
|
|
|
|
def test_invalid_transition_ready_to_processing(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
profile.transition_to(VoiceCloneStatus.PROCESSING)
|
|
|
|
def test_invalid_transition_failed_to_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
profile.transition_to(VoiceCloneStatus.READY)
|
|
|
|
def test_invalid_transition_disabled_to_pending(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.DISABLED
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
profile.transition_to(VoiceCloneStatus.PENDING)
|
|
|
|
|
|
class TestMarkMethods:
|
|
"""标记方法测试"""
|
|
|
|
def test_mark_processing(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.error_message = "some error"
|
|
profile.mark_processing()
|
|
assert profile.status == VoiceCloneStatus.PROCESSING
|
|
assert profile.error_message == ""
|
|
|
|
def test_mark_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.error_message = "old error"
|
|
profile.mark_ready("voice_abc123")
|
|
assert profile.status == VoiceCloneStatus.READY
|
|
assert profile.voice_id == "voice_abc123"
|
|
assert profile.error_message == ""
|
|
|
|
def test_mark_ready_empty_voice_id_raises(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
with pytest.raises(ValueError, match="voice_id 不能为空"):
|
|
profile.mark_ready(" ")
|
|
|
|
def test_mark_ready_strips_whitespace(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.mark_ready(" voice_123 ")
|
|
assert profile.voice_id == "voice_123"
|
|
|
|
def test_mark_failed(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.mark_failed("音频质量太差")
|
|
assert profile.status == VoiceCloneStatus.FAILED
|
|
assert profile.error_message == "音频质量太差"
|
|
|
|
def test_mark_disabled_from_pending(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.mark_disabled()
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
|
|
def test_mark_disabled_from_ready(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
profile.voice_id = "v1"
|
|
profile.mark_disabled()
|
|
assert profile.status == VoiceCloneStatus.DISABLED
|
|
assert profile.voice_id == "v1" # 禁用不清除voice_id
|
|
|
|
|
|
class TestPrepareRetry:
|
|
"""重试准备测试"""
|
|
|
|
def test_prepare_retry_success(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.retry_count = 1
|
|
profile.error_message = "timeout"
|
|
profile.voice_id = "old_voice"
|
|
profile.prepare_retry()
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.retry_count == 2
|
|
assert profile.error_message == ""
|
|
assert profile.voice_id == ""
|
|
|
|
def test_prepare_retry_first_time(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.prepare_retry()
|
|
assert profile.status == VoiceCloneStatus.PENDING
|
|
assert profile.retry_count == 1
|
|
|
|
def test_prepare_retry_exceeds_max_raises(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
|
profile.status = VoiceCloneStatus.FAILED
|
|
profile.retry_count = 3
|
|
with pytest.raises(ValueError, match="不可重试"):
|
|
profile.prepare_retry()
|
|
|
|
def test_prepare_retry_from_pending_raises(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
with pytest.raises(ValueError, match="不可重试"):
|
|
profile.prepare_retry()
|
|
|
|
def test_prepare_retry_from_ready_raises(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.READY
|
|
with pytest.raises(ValueError, match="不可重试"):
|
|
profile.prepare_retry()
|
|
|
|
|
|
class TestToDict:
|
|
"""序列化测试"""
|
|
|
|
def test_to_dict_basic(self):
|
|
profile = VoiceCloneProfile.create(
|
|
user_id="user1",
|
|
name="测试音色",
|
|
description="desc",
|
|
max_retries=2,
|
|
)
|
|
d = profile.to_dict()
|
|
assert d["id"] == profile.id
|
|
assert d["user_id"] == "user1"
|
|
assert d["name"] == "测试音色"
|
|
assert d["description"] == "desc"
|
|
assert d["status"] == "pending"
|
|
assert d["retry_count"] == 0
|
|
assert d["max_retries"] == 2
|
|
assert d["is_retryable"] is False
|
|
assert d["is_ready"] is False
|
|
assert isinstance(d["created_at"], str)
|
|
assert isinstance(d["updated_at"], str)
|
|
|
|
def test_to_dict_ready_state(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.status = VoiceCloneStatus.PROCESSING
|
|
profile.mark_ready("voice_123")
|
|
d = profile.to_dict()
|
|
assert d["status"] == "ready"
|
|
assert d["voice_id"] == "voice_123"
|
|
assert d["is_ready"] is True
|
|
assert d["is_retryable"] is False
|
|
|
|
def test_to_dict_failed_state(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
|
profile.mark_failed("some error")
|
|
d = profile.to_dict()
|
|
assert d["status"] == "failed"
|
|
assert d["error_message"] == "some error"
|
|
assert d["is_retryable"] is True # retry_count=0, max_retries=3
|
|
|
|
def test_to_dict_includes_metadata(self):
|
|
profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata={"key": "value", "num": 42})
|
|
d = profile.to_dict()
|
|
assert d["metadata"] == {"key": "value", "num": 42}
|