CI Bot
2df7bc9dc8
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m41s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m45s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m54s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m1s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m34s
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 5m49s
CI/CD Pipeline / Frontend Lint (push) Successful in 6m0s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m27s
CI/CD Pipeline / Integration Tests (push) Successful in 2m20s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m11s
CI/CD Pipeline / Unit Tests (push) Failing after 5m52s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 30s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 1m48s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m10s
360 lines
13 KiB
Python
Executable File
360 lines
13 KiB
Python
Executable File
"""VoiceCloneProfile 领域层单元测试 - voice_clone_profile.py"""
|
||
|
||
import pytest
|
||
|
||
from packages.domain.voice_clone_profile import (
|
||
TERMINAL_STATUSES,
|
||
VoiceCloneProfile,
|
||
VoiceCloneStatus,
|
||
)
|
||
|
||
|
||
class TestVoiceCloneStatus:
|
||
"""VoiceCloneStatus 枚举测试"""
|
||
|
||
def test_all_statuses_have_values(self):
|
||
for s in VoiceCloneStatus:
|
||
assert isinstance(s.value, str)
|
||
assert s.value
|
||
|
||
def test_str_enum(self):
|
||
assert VoiceCloneStatus.PENDING == "pending"
|
||
assert isinstance(VoiceCloneStatus.PENDING, str)
|
||
|
||
def test_terminal_statuses(self):
|
||
assert VoiceCloneStatus.READY in TERMINAL_STATUSES
|
||
assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES
|
||
assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES
|
||
|
||
def test_non_terminal(self):
|
||
assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES
|
||
assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES
|
||
|
||
|
||
class TestVoiceCloneProfileCreate:
|
||
"""create 工厂方法测试"""
|
||
|
||
def test_create_basic(self):
|
||
profile = VoiceCloneProfile.create(user_id="user-1", name="我的音色")
|
||
assert profile.id
|
||
assert len(profile.id) == 32
|
||
assert profile.user_id == "user-1"
|
||
assert profile.name == "我的音色"
|
||
assert profile.status == VoiceCloneStatus.PENDING
|
||
assert profile.description == ""
|
||
assert profile.source_audio_url == ""
|
||
assert profile.voice_id == ""
|
||
assert profile.language == "zh-CN"
|
||
assert profile.gender == "unknown"
|
||
assert profile.retry_count == 0
|
||
assert profile.max_retries == 3
|
||
assert profile.metadata == {}
|
||
|
||
def test_create_empty_user_id_raises(self):
|
||
with pytest.raises(ValueError, match="user_id 不能为空"):
|
||
VoiceCloneProfile.create(user_id=" ", name="test")
|
||
|
||
def test_create_empty_name_raises(self):
|
||
with pytest.raises(ValueError, match="name 不能为空"):
|
||
VoiceCloneProfile.create(user_id="u1", name=" ")
|
||
|
||
def test_create_name_too_long_raises(self):
|
||
long_name = "a" * 101
|
||
with pytest.raises(ValueError, match="name 长度不能超过 100"):
|
||
VoiceCloneProfile.create(user_id="u1", name=long_name)
|
||
|
||
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_with_description(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", description="温暖男声")
|
||
assert profile.description == "温暖男声"
|
||
|
||
def test_create_with_source_audio(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", source_audio_url="http://audio.com/source.wav")
|
||
assert profile.source_audio_url == "http://audio.com/source.wav"
|
||
|
||
def test_create_with_language(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", language="en-US")
|
||
assert profile.language == "en-US"
|
||
|
||
def test_create_gender_lowercased(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", gender="MALE")
|
||
assert profile.gender == "male"
|
||
|
||
def test_create_with_metadata(self):
|
||
meta = {"source": "upload", "duration": 30}
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata=meta)
|
||
assert profile.metadata == meta
|
||
|
||
def test_create_none_metadata_defaults_empty(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata=None)
|
||
assert profile.metadata == {}
|
||
|
||
def test_create_fields_stripped(self):
|
||
profile = VoiceCloneProfile.create(
|
||
user_id=" u1 ",
|
||
name=" test ",
|
||
description=" desc ",
|
||
source_audio_url=" url ",
|
||
voice_model=" model ",
|
||
language=" zh-CN ",
|
||
)
|
||
assert profile.user_id == "u1"
|
||
assert profile.name == "test"
|
||
assert profile.description == "desc"
|
||
assert profile.source_audio_url == "url"
|
||
assert profile.voice_model == "model"
|
||
assert profile.language == "zh-CN"
|
||
|
||
def test_create_custom_max_retries(self):
|
||
profile = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=5)
|
||
assert profile.max_retries == 5
|
||
|
||
|
||
class TestVoiceCloneProfileProperties:
|
||
"""属性测试"""
|
||
|
||
def test_is_terminal_pending(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
assert p.is_terminal is False
|
||
|
||
def test_is_terminal_ready(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
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="u1", name="test")
|
||
p.mark_processing()
|
||
p.mark_failed("error")
|
||
assert p.is_terminal is True
|
||
|
||
def test_is_terminal_disabled(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_disabled()
|
||
assert p.is_terminal is True
|
||
|
||
def test_is_retryable_failed_within_limit(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test", 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="test", max_retries=1)
|
||
p.mark_processing()
|
||
p.mark_failed("error")
|
||
p.retry_count = 1
|
||
assert p.is_retryable is False
|
||
|
||
def test_is_ready_true(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_processing()
|
||
p.mark_ready("voice-123")
|
||
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="test")
|
||
p.status = VoiceCloneStatus.READY
|
||
p.voice_id = ""
|
||
assert p.is_ready is False
|
||
|
||
def test_is_ready_not_ready_status(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.voice_id = "some-id"
|
||
assert p.is_ready is False # 状态是 PENDING
|
||
|
||
|
||
class TestVoiceCloneProfileTransitions:
|
||
"""状态转换测试"""
|
||
|
||
def test_pending_to_processing(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||
assert p.status == VoiceCloneStatus.PROCESSING
|
||
|
||
def test_pending_to_failed(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.transition_to(VoiceCloneStatus.FAILED)
|
||
assert p.status == VoiceCloneStatus.FAILED
|
||
|
||
def test_pending_to_disabled(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||
assert p.status == VoiceCloneStatus.DISABLED
|
||
|
||
def test_processing_to_ready(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
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="u1", name="test")
|
||
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="u1", name="test")
|
||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||
assert p.status == VoiceCloneStatus.DISABLED
|
||
|
||
def test_ready_to_disabled(self):
|
||
"""已就绪音色可以被禁用"""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_processing()
|
||
p.mark_ready("voice-1")
|
||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||
assert p.status == VoiceCloneStatus.DISABLED
|
||
|
||
def test_failed_to_pending_retry(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||
p.transition_to(VoiceCloneStatus.FAILED)
|
||
p.transition_to(VoiceCloneStatus.PENDING)
|
||
assert p.status == VoiceCloneStatus.PENDING
|
||
|
||
def test_invalid_transition_raises(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
with pytest.raises(ValueError, match="非法状态转换"):
|
||
p.transition_to(VoiceCloneStatus.READY) # pending 不能直接到 ready
|
||
|
||
def test_disabled_to_pending_raises(self):
|
||
"""禁用后不能回到 pending"""
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_disabled()
|
||
with pytest.raises(ValueError, match="非法状态转换"):
|
||
p.transition_to(VoiceCloneStatus.PENDING)
|
||
|
||
def test_transition_with_string(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.transition_to("processing")
|
||
assert p.status == VoiceCloneStatus.PROCESSING
|
||
|
||
def test_transition_invalid_string_raises(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
with pytest.raises(ValueError, match="无效状态"):
|
||
p.transition_to("invalid")
|
||
|
||
def test_transition_updates_updated_at(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
old = p.updated_at
|
||
import time
|
||
|
||
time.sleep(0.001)
|
||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||
assert p.updated_at >= old
|
||
|
||
|
||
class TestVoiceCloneProfileMarkMethods:
|
||
"""便捷标记方法测试"""
|
||
|
||
def test_mark_processing(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.error_message = "prev error"
|
||
p.mark_processing()
|
||
assert p.status == VoiceCloneStatus.PROCESSING
|
||
assert p.error_message == ""
|
||
|
||
def test_mark_ready(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_processing()
|
||
p.mark_ready("voice-abc123")
|
||
assert p.status == VoiceCloneStatus.READY
|
||
assert p.voice_id == "voice-abc123"
|
||
assert p.error_message == ""
|
||
|
||
def test_mark_ready_empty_voice_id_raises(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_processing()
|
||
with pytest.raises(ValueError, match="voice_id 不能为空"):
|
||
p.mark_ready(" ")
|
||
|
||
def test_mark_failed(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
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="u1", name="test")
|
||
p.mark_disabled()
|
||
assert p.status == VoiceCloneStatus.DISABLED
|
||
|
||
|
||
class TestVoiceCloneProfileRetry:
|
||
"""重试逻辑测试"""
|
||
|
||
def test_prepare_retry(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
||
p.mark_processing()
|
||
p.mark_failed("超时")
|
||
p.voice_id = "partial-id"
|
||
p.prepare_retry()
|
||
assert p.status == VoiceCloneStatus.PENDING
|
||
assert p.retry_count == 1
|
||
assert p.error_message == ""
|
||
assert p.voice_id == "" # 重试时清空 voice_id
|
||
|
||
def test_prepare_retry_not_retryable_raises(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=0)
|
||
p.mark_processing()
|
||
p.mark_failed("error")
|
||
with pytest.raises(ValueError, match="不可重试"):
|
||
p.prepare_retry()
|
||
|
||
def test_multiple_retries(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test", max_retries=3)
|
||
for i in range(3):
|
||
p.mark_processing()
|
||
p.mark_failed(f"error-{i}")
|
||
p.prepare_retry()
|
||
assert p.retry_count == i + 1
|
||
assert p.is_retryable is False
|
||
|
||
|
||
class TestVoiceCloneProfileToDict:
|
||
"""to_dict 序列化测试"""
|
||
|
||
def test_to_dict_contains_fields(self):
|
||
p = VoiceCloneProfile.create(
|
||
user_id="u1",
|
||
name="我的音色",
|
||
description="测试",
|
||
language="en-US",
|
||
gender="female",
|
||
)
|
||
d = p.to_dict()
|
||
assert d["id"] == p.id
|
||
assert d["user_id"] == "u1"
|
||
assert d["name"] == "我的音色"
|
||
assert d["description"] == "测试"
|
||
assert d["status"] == "pending"
|
||
assert d["language"] == "en-US"
|
||
assert d["gender"] == "female"
|
||
assert d["is_retryable"] is False
|
||
assert d["is_ready"] is False
|
||
|
||
def test_to_dict_datetime_are_strings(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
d = p.to_dict()
|
||
assert isinstance(d["created_at"], str)
|
||
assert isinstance(d["updated_at"], str)
|
||
|
||
def test_to_dict_after_ready(self):
|
||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||
p.mark_processing()
|
||
p.mark_ready("voice-123")
|
||
d = p.to_dict()
|
||
assert d["status"] == "ready"
|
||
assert d["voice_id"] == "voice-123"
|
||
assert d["is_ready"] is True
|