test(P3-1): 第53波 domain 单测 - tts_job/voice_clone_profile/video_share（+133） #843
+243
-293
@@ -1,242 +1,280 @@
|
||||
"""tts_job 领域模型单元测试."""
|
||||
"""TTSJob 领域层单元测试 - tts_job.py"""
|
||||
|
||||
import pytest
|
||||
from domain.tts_job import TERMINAL_STATUSES, TTSJob, TTSJobStatus
|
||||
|
||||
from packages.domain.tts_job import (
|
||||
TERMINAL_STATUSES,
|
||||
TTSJob,
|
||||
TTSJobStatus,
|
||||
)
|
||||
|
||||
|
||||
class TestTTSJobStatus:
|
||||
"""TTSJobStatus 枚举测试."""
|
||||
"""TTSJobStatus 枚举测试"""
|
||||
|
||||
def test_values(self):
|
||||
def test_all_statuses_have_values(self):
|
||||
for s in TTSJobStatus:
|
||||
assert isinstance(s.value, str)
|
||||
assert s.value
|
||||
|
||||
def test_str_enum_behavior(self):
|
||||
assert TTSJobStatus.PENDING == "pending"
|
||||
assert TTSJobStatus.PROCESSING == "processing"
|
||||
assert TTSJobStatus.COMPLETED == "completed"
|
||||
assert TTSJobStatus.FAILED == "failed"
|
||||
assert TTSJobStatus.CANCELLED == "cancelled"
|
||||
assert isinstance(TTSJobStatus.PENDING, str)
|
||||
|
||||
def test_terminal_statuses(self):
|
||||
assert TTSJobStatus.COMPLETED in TERMINAL_STATUSES
|
||||
assert TTSJobStatus.FAILED in TERMINAL_STATUSES
|
||||
assert TTSJobStatus.CANCELLED in TERMINAL_STATUSES
|
||||
|
||||
def test_non_terminal_statuses(self):
|
||||
assert TTSJobStatus.PENDING not in TERMINAL_STATUSES
|
||||
assert TTSJobStatus.PROCESSING not in TERMINAL_STATUSES
|
||||
|
||||
|
||||
class TestTTSJobCreate:
|
||||
"""TTSJob.create 工厂方法测试."""
|
||||
"""TTSJob.create 工厂方法测试"""
|
||||
|
||||
def test_create_with_required_fields(self):
|
||||
job = TTSJob.create(user_id="user_001", input_text="你好世界")
|
||||
def test_create_basic(self):
|
||||
job = TTSJob.create(user_id="user-1", input_text="你好世界")
|
||||
assert job.id
|
||||
assert len(job.id) == 32
|
||||
assert job.user_id == "user_001"
|
||||
assert job.user_id == "user-1"
|
||||
assert job.input_text == "你好世界"
|
||||
assert job.status == TTSJobStatus.PENDING
|
||||
assert job.voice_id == ""
|
||||
assert job.sample_rate == 22050
|
||||
assert job.voice_model == ""
|
||||
assert job.format == "mp3"
|
||||
assert job.sample_rate == 22050
|
||||
assert job.retry_count == 0
|
||||
assert job.max_retries == 3
|
||||
assert job.metadata == {}
|
||||
assert job.created_at is not None
|
||||
assert job.updated_at is not None
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
job = TTSJob.create(
|
||||
user_id="user_002",
|
||||
input_text="测试文本",
|
||||
voice_id="voice_001",
|
||||
voice_model="cosyvoice",
|
||||
project_id="proj_001",
|
||||
voice_clone_profile_id="clone_001",
|
||||
sample_rate=16000,
|
||||
format="wav",
|
||||
max_retries=5,
|
||||
metadata={"key": "value"},
|
||||
)
|
||||
assert job.voice_id == "voice_001"
|
||||
assert job.voice_model == "cosyvoice"
|
||||
assert job.project_id == "proj_001"
|
||||
assert job.voice_clone_profile_id == "clone_001"
|
||||
assert job.sample_rate == 16000
|
||||
assert job.format == "wav"
|
||||
assert job.max_retries == 5
|
||||
assert job.metadata == {"key": "value"}
|
||||
|
||||
def test_create_strips_strings(self):
|
||||
job = TTSJob.create(
|
||||
user_id=" user_003 ",
|
||||
input_text=" 测试文本 ",
|
||||
voice_id=" voice_001 ",
|
||||
voice_model=" cosyvoice ",
|
||||
project_id=" proj_001 ",
|
||||
voice_clone_profile_id=" clone_001 ",
|
||||
format="wav",
|
||||
)
|
||||
assert job.user_id == "user_003"
|
||||
assert job.input_text == "测试文本"
|
||||
assert job.voice_id == "voice_001"
|
||||
assert job.voice_model == "cosyvoice"
|
||||
assert job.project_id == "proj_001"
|
||||
assert job.voice_clone_profile_id == "clone_001"
|
||||
assert job.format == "wav"
|
||||
|
||||
def test_create_empty_user_id_raises(self):
|
||||
with pytest.raises(ValueError, match="user_id"):
|
||||
TTSJob.create(user_id="", input_text="test")
|
||||
|
||||
def test_create_whitespace_user_id_raises(self):
|
||||
with pytest.raises(ValueError, match="user_id"):
|
||||
with pytest.raises(ValueError, match="user_id 不能为空"):
|
||||
TTSJob.create(user_id=" ", input_text="test")
|
||||
|
||||
def test_create_empty_input_text_raises(self):
|
||||
with pytest.raises(ValueError, match="input_text"):
|
||||
TTSJob.create(user_id="u", input_text="")
|
||||
with pytest.raises(ValueError, match="input_text 不能为空"):
|
||||
TTSJob.create(user_id="u1", input_text=" ")
|
||||
|
||||
def test_create_input_text_too_long_raises(self):
|
||||
def test_create_text_too_long_raises(self):
|
||||
long_text = "a" * 10001
|
||||
with pytest.raises(ValueError, match="10000"):
|
||||
TTSJob.create(user_id="u", input_text=long_text)
|
||||
with pytest.raises(ValueError, match="input_text 长度不能超过 10000"):
|
||||
TTSJob.create(user_id="u1", input_text=long_text)
|
||||
|
||||
def test_create_input_text_at_limit_ok(self):
|
||||
def test_create_text_exactly_10000_ok(self):
|
||||
text = "a" * 10000
|
||||
job = TTSJob.create(user_id="u", input_text=text)
|
||||
assert job.input_text == text
|
||||
job = TTSJob.create(user_id="u1", input_text=text)
|
||||
assert len(job.input_text) == 10000
|
||||
|
||||
def test_create_invalid_format_raises(self):
|
||||
with pytest.raises(ValueError, match="不支持的输出格式"):
|
||||
TTSJob.create(user_id="u", input_text="t", format="flac")
|
||||
TTSJob.create(user_id="u1", input_text="test", format="flac")
|
||||
|
||||
def test_create_supported_formats(self):
|
||||
for fmt in ["mp3", "wav", "pcm"]:
|
||||
job = TTSJob.create(user_id="u", input_text="t", format=fmt)
|
||||
assert job.format == fmt
|
||||
def test_create_mp3_format(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", format="mp3")
|
||||
assert job.format == "mp3"
|
||||
|
||||
def test_create_none_metadata_defaults_to_empty_dict(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t", metadata=None)
|
||||
def test_create_wav_format(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", format="wav")
|
||||
assert job.format == "wav"
|
||||
|
||||
def test_create_pcm_format(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", format="pcm")
|
||||
assert job.format == "pcm"
|
||||
|
||||
def test_create_with_voice_id(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", voice_id="voice-1")
|
||||
assert job.voice_id == "voice-1"
|
||||
|
||||
def test_create_with_project_id(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", project_id="proj-1")
|
||||
assert job.project_id == "proj-1"
|
||||
|
||||
def test_create_with_clone_profile(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", voice_clone_profile_id="vc-1")
|
||||
assert job.voice_clone_profile_id == "vc-1"
|
||||
|
||||
def test_create_with_metadata(self):
|
||||
meta = {"source": "api", "priority": "high"}
|
||||
job = TTSJob.create(user_id="u1", input_text="test", metadata=meta)
|
||||
assert job.metadata == meta
|
||||
|
||||
def test_create_none_metadata_defaults_empty(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", metadata=None)
|
||||
assert job.metadata == {}
|
||||
|
||||
def test_create_ids_are_unique(self):
|
||||
j1 = TTSJob.create(user_id="u", input_text="t")
|
||||
j2 = TTSJob.create(user_id="u", input_text="t")
|
||||
assert j1.id != j2.id
|
||||
def test_create_input_text_stripped(self):
|
||||
job = TTSJob.create(user_id=" u1 ", input_text=" hello ")
|
||||
assert job.user_id == "u1"
|
||||
assert job.input_text == "hello"
|
||||
|
||||
def test_create_custom_max_retries(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=5)
|
||||
assert job.max_retries == 5
|
||||
|
||||
|
||||
class TestTTSJobStateMachine:
|
||||
"""TTSJob 状态机测试."""
|
||||
class TestTTSJobProperties:
|
||||
"""属性测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def pending_job(self):
|
||||
return TTSJob.create(user_id="user_001", input_text="测试")
|
||||
def test_is_terminal_pending(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
assert job.is_terminal is False
|
||||
|
||||
def test_initial_status_is_pending(self, pending_job):
|
||||
assert pending_job.status == TTSJobStatus.PENDING
|
||||
assert not pending_job.is_terminal
|
||||
def test_is_terminal_completed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_completed("http://audio.com/a.mp3")
|
||||
assert job.is_terminal is True
|
||||
|
||||
def test_pending_to_processing(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
assert pending_job.status == TTSJobStatus.PROCESSING
|
||||
assert pending_job.started_at is not None
|
||||
assert pending_job.error_message == ""
|
||||
def test_is_terminal_failed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_failed("error")
|
||||
assert job.is_terminal is True
|
||||
|
||||
def test_pending_can_fail_directly(self, pending_job):
|
||||
"""pending 可以直接到 failed（比如入参校验失败）"""
|
||||
pending_job.mark_failed("校验失败")
|
||||
assert pending_job.status == TTSJobStatus.FAILED
|
||||
assert pending_job.error_message == "校验失败"
|
||||
def test_is_retryable_failed_within_limit(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=3)
|
||||
job.mark_processing()
|
||||
job.mark_failed("error")
|
||||
assert job.is_retryable is True
|
||||
|
||||
def test_pending_can_be_cancelled(self, pending_job):
|
||||
pending_job.mark_cancelled()
|
||||
assert pending_job.status == TTSJobStatus.CANCELLED
|
||||
def test_is_retryable_failed_at_limit(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=1)
|
||||
job.mark_processing()
|
||||
job.mark_failed("error")
|
||||
job.retry_count = 1
|
||||
assert job.is_retryable is False
|
||||
|
||||
def test_processing_to_completed(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
assert pending_job.status == TTSJobStatus.COMPLETED
|
||||
assert pending_job.output_audio_url == "https://example.com/out.mp3"
|
||||
assert pending_job.completed_at is not None
|
||||
assert pending_job.error_message == ""
|
||||
def test_is_completed_true(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_completed("http://audio.com/a.mp3")
|
||||
assert job.is_completed is True
|
||||
|
||||
def test_processing_to_failed(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_failed("API 超时")
|
||||
assert pending_job.status == TTSJobStatus.FAILED
|
||||
assert pending_job.error_message == "API 超时"
|
||||
def test_is_completed_no_url(self):
|
||||
"""completed 状态但没有 output_audio_url 为空，is_completed 为 False"""
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.status = TTSJobStatus.COMPLETED
|
||||
job.output_audio_url = ""
|
||||
assert job.is_completed is False
|
||||
|
||||
def test_processing_can_be_cancelled(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_cancelled()
|
||||
assert pending_job.status == TTSJobStatus.CANCELLED
|
||||
|
||||
def test_completed_is_terminal(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
assert pending_job.is_terminal
|
||||
assert pending_job.is_completed
|
||||
class TestTTSJobTransitions:
|
||||
"""状态转换测试"""
|
||||
|
||||
def test_failed_is_terminal_but_retryable(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_failed("error")
|
||||
assert pending_job.is_terminal
|
||||
assert pending_job.is_retryable
|
||||
def test_pending_to_processing(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.PROCESSING)
|
||||
assert job.status == TTSJobStatus.PROCESSING
|
||||
|
||||
def test_cancelled_is_terminal_and_not_retryable(self, pending_job):
|
||||
pending_job.mark_cancelled()
|
||||
assert pending_job.is_terminal
|
||||
assert not pending_job.is_retryable
|
||||
def test_pending_to_failed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.FAILED)
|
||||
assert job.status == TTSJobStatus.FAILED
|
||||
|
||||
def test_invalid_transition_completed_to_processing_raises(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
def test_pending_to_cancelled(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.CANCELLED)
|
||||
assert job.status == TTSJobStatus.CANCELLED
|
||||
|
||||
def test_processing_to_completed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.PROCESSING)
|
||||
job.transition_to(TTSJobStatus.COMPLETED)
|
||||
assert job.status == TTSJobStatus.COMPLETED
|
||||
|
||||
def test_processing_to_failed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.PROCESSING)
|
||||
job.transition_to(TTSJobStatus.FAILED)
|
||||
assert job.status == TTSJobStatus.FAILED
|
||||
|
||||
def test_failed_to_pending_retry(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to(TTSJobStatus.PROCESSING)
|
||||
job.transition_to(TTSJobStatus.FAILED)
|
||||
job.transition_to(TTSJobStatus.PENDING)
|
||||
assert job.status == TTSJobStatus.PENDING
|
||||
|
||||
def test_invalid_transition_raises(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
with pytest.raises(ValueError, match="非法状态转换"):
|
||||
pending_job.mark_processing()
|
||||
job.transition_to(TTSJobStatus.COMPLETED) # pending 不能直接到 completed
|
||||
|
||||
def test_invalid_transition_completed_to_failed_raises(self, pending_job):
|
||||
pending_job.mark_processing()
|
||||
pending_job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
with pytest.raises(ValueError, match="非法状态转换"):
|
||||
pending_job.mark_failed("test")
|
||||
def test_transition_with_string(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.transition_to("processing")
|
||||
assert job.status == TTSJobStatus.PROCESSING
|
||||
|
||||
def test_cancelled_cannot_transition(self, pending_job):
|
||||
pending_job.mark_cancelled()
|
||||
with pytest.raises(ValueError):
|
||||
pending_job.mark_processing()
|
||||
with pytest.raises(ValueError):
|
||||
pending_job.mark_failed("test")
|
||||
|
||||
def test_transition_to_with_string(self, pending_job):
|
||||
"""transition_to 支持字符串参数"""
|
||||
pending_job.transition_to("processing")
|
||||
assert pending_job.status == TTSJobStatus.PROCESSING
|
||||
|
||||
def test_transition_to_invalid_string_raises(self, pending_job):
|
||||
def test_transition_invalid_string_raises(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
with pytest.raises(ValueError, match="无效状态"):
|
||||
pending_job.transition_to("invalid_status")
|
||||
job.transition_to("invalid")
|
||||
|
||||
def test_state_transition_updates_updated_at(self, pending_job):
|
||||
old_updated = pending_job.updated_at
|
||||
def test_transition_updates_updated_at(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
old = job.updated_at
|
||||
import time
|
||||
|
||||
time.sleep(0.001)
|
||||
pending_job.mark_processing()
|
||||
assert pending_job.updated_at > old_updated
|
||||
job.transition_to(TTSJobStatus.PROCESSING)
|
||||
assert job.updated_at >= old
|
||||
|
||||
|
||||
class TestTTSJobMarkMethods:
|
||||
"""便捷标记方法测试"""
|
||||
|
||||
def test_mark_processing(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.error_message = "previous error"
|
||||
job.mark_processing()
|
||||
assert job.status == TTSJobStatus.PROCESSING
|
||||
assert job.started_at is not None
|
||||
assert job.error_message == ""
|
||||
|
||||
def test_mark_completed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_completed(
|
||||
"http://example.com/audio.mp3",
|
||||
output_audio_key="audio/key.mp3",
|
||||
duration=5.5,
|
||||
file_size=102400,
|
||||
)
|
||||
assert job.status == TTSJobStatus.COMPLETED
|
||||
assert job.output_audio_url == "http://example.com/audio.mp3"
|
||||
assert job.output_audio_key == "audio/key.mp3"
|
||||
assert job.duration == 5.5
|
||||
assert job.file_size == 102400
|
||||
assert job.completed_at is not None
|
||||
assert job.error_message == ""
|
||||
|
||||
def test_mark_completed_empty_url_raises(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
with pytest.raises(ValueError, match="output_audio_url 不能为空"):
|
||||
job.mark_completed(" ")
|
||||
|
||||
def test_mark_failed(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_failed("API 调用超时")
|
||||
assert job.status == TTSJobStatus.FAILED
|
||||
assert job.error_message == "API 调用超时"
|
||||
|
||||
def test_mark_cancelled(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_cancelled()
|
||||
assert job.status == TTSJobStatus.CANCELLED
|
||||
|
||||
|
||||
class TestTTSJobRetry:
|
||||
"""TTSJob 重试逻辑测试."""
|
||||
"""重试逻辑测试"""
|
||||
|
||||
def test_failed_can_retry(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t", max_retries=3)
|
||||
def test_prepare_retry(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=3)
|
||||
job.mark_processing()
|
||||
job.mark_failed("error")
|
||||
assert job.is_retryable
|
||||
assert job.retry_count == 0
|
||||
|
||||
def test_prepare_retry_resets_to_pending(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.mark_failed("error")
|
||||
|
||||
job.mark_failed("超时")
|
||||
job.prepare_retry()
|
||||
assert job.status == TTSJobStatus.PENDING
|
||||
assert job.retry_count == 1
|
||||
@@ -244,150 +282,62 @@ class TestTTSJobRetry:
|
||||
assert job.started_at is None
|
||||
assert job.completed_at is None
|
||||
|
||||
def test_retry_up_to_max_retries(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t", max_retries=2)
|
||||
# 第 1 次失败 + 重试 → retry_count=1，还可以重试
|
||||
def test_prepare_retry_not_retryable_raises(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=0)
|
||||
job.mark_processing()
|
||||
job.mark_failed("e1")
|
||||
assert job.is_retryable
|
||||
job.prepare_retry()
|
||||
assert job.retry_count == 1
|
||||
|
||||
# 第 2 次失败 → retry_count=1，还是 failed 状态，还可以重试（max_retries=2）
|
||||
job.mark_processing()
|
||||
job.mark_failed("e2")
|
||||
assert job.is_retryable # retry_count=1 < max_retries=2
|
||||
job.prepare_retry()
|
||||
assert job.retry_count == 2
|
||||
|
||||
# 第 3 次失败 → retry_count=2，达到上限，不可重试
|
||||
job.mark_processing()
|
||||
job.mark_failed("e3")
|
||||
assert not job.is_retryable # retry_count=2 == max_retries=2
|
||||
|
||||
def test_retry_exceed_max_raises(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t", max_retries=1)
|
||||
job.mark_processing()
|
||||
job.mark_failed("e")
|
||||
job.prepare_retry() # 第 1 次重试，用完了
|
||||
|
||||
job.mark_processing()
|
||||
job.mark_failed("e2")
|
||||
job.mark_failed("error")
|
||||
with pytest.raises(ValueError, match="不可重试"):
|
||||
job.prepare_retry()
|
||||
|
||||
def test_pending_not_retryable(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
assert not job.is_retryable
|
||||
with pytest.raises(ValueError, match="不可重试"):
|
||||
def test_multiple_retries(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test", max_retries=3)
|
||||
for i in range(3):
|
||||
job.mark_processing()
|
||||
job.mark_failed(f"error-{i}")
|
||||
job.prepare_retry()
|
||||
|
||||
def test_completed_not_retryable(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
assert not job.is_retryable
|
||||
with pytest.raises(ValueError, match="不可重试"):
|
||||
job.prepare_retry()
|
||||
|
||||
def test_cancelled_not_retryable(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_cancelled()
|
||||
assert not job.is_retryable
|
||||
|
||||
|
||||
class TestTTSJobMarkCompleted:
|
||||
"""mark_completed 方法测试."""
|
||||
|
||||
def test_requires_output_url(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
with pytest.raises(ValueError, match="output_audio_url"):
|
||||
job.mark_completed(output_audio_url="")
|
||||
|
||||
def test_sets_all_fields(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.mark_completed(
|
||||
output_audio_url="https://example.com/out.mp3",
|
||||
output_audio_key="audio/001.mp3",
|
||||
duration=30.5,
|
||||
file_size=102400,
|
||||
)
|
||||
assert job.output_audio_url == "https://example.com/out.mp3"
|
||||
assert job.output_audio_key == "audio/001.mp3"
|
||||
assert job.duration == 30.5
|
||||
assert job.file_size == 102400
|
||||
assert job.completed_at is not None
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.mark_completed(
|
||||
output_audio_url=" https://example.com/out.mp3 ",
|
||||
output_audio_key=" audio/001.mp3 ",
|
||||
)
|
||||
assert job.output_audio_url == "https://example.com/out.mp3"
|
||||
assert job.output_audio_key == "audio/001.mp3"
|
||||
|
||||
|
||||
class TestTTSJobIsCompleted:
|
||||
"""is_completed 属性测试."""
|
||||
|
||||
def test_completed_with_url_is_completed(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.mark_completed(output_audio_url="https://example.com/out.mp3")
|
||||
assert job.is_completed
|
||||
|
||||
def test_completed_without_url_not_completed(self):
|
||||
"""极端情况：completed 状态但没有 URL（理论不会发生）"""
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_processing()
|
||||
job.transition_to(TTSJobStatus.COMPLETED) # 直接转，不设 URL
|
||||
assert not job.is_completed
|
||||
|
||||
def test_pending_not_completed(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
assert not job.is_completed
|
||||
assert job.retry_count == i + 1
|
||||
# 第3次重试后 retry_count=3，等于 max_retries=3，不可再重试
|
||||
assert job.is_retryable is False
|
||||
|
||||
|
||||
class TestTTSJobToDict:
|
||||
"""to_dict 序列化测试."""
|
||||
"""to_dict 序列化测试"""
|
||||
|
||||
def test_pending_job_to_dict(self):
|
||||
job = TTSJob.create(user_id="user_001", input_text="测试文本", voice_id="v001")
|
||||
def test_to_dict_contains_fields(self):
|
||||
job = TTSJob.create(
|
||||
user_id="u1",
|
||||
input_text="test",
|
||||
voice_id="voice-1",
|
||||
project_id="proj-1",
|
||||
)
|
||||
d = job.to_dict()
|
||||
assert d["id"] == job.id
|
||||
assert d["user_id"] == "user_001"
|
||||
assert d["user_id"] == "u1"
|
||||
assert d["input_text"] == "test"
|
||||
assert d["voice_id"] == "voice-1"
|
||||
assert d["status"] == "pending"
|
||||
assert d["input_text"] == "测试文本"
|
||||
assert d["voice_id"] == "v001"
|
||||
assert d["retry_count"] == 0
|
||||
assert d["is_retryable"] is False
|
||||
assert d["is_completed"] is False
|
||||
assert d["metadata"] == {}
|
||||
|
||||
def test_to_dict_datetime_are_strings(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
d = job.to_dict()
|
||||
assert isinstance(d["created_at"], str)
|
||||
assert isinstance(d["updated_at"], str)
|
||||
|
||||
def test_to_dict_none_datetime(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
d = job.to_dict()
|
||||
assert d["started_at"] is None
|
||||
assert d["completed_at"] is None
|
||||
assert d["created_at"] is not None
|
||||
assert d["updated_at"] is not None
|
||||
|
||||
def test_completed_job_to_dict(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
def test_to_dict_after_completion(self):
|
||||
job = TTSJob.create(user_id="u1", input_text="test")
|
||||
job.mark_processing()
|
||||
job.mark_completed(output_audio_url="https://example.com/out.mp3", duration=10.0)
|
||||
job.mark_completed("http://test.mp3", duration=10.0)
|
||||
d = job.to_dict()
|
||||
assert d["status"] == "completed"
|
||||
assert d["output_audio_url"] == "https://example.com/out.mp3"
|
||||
assert d["duration"] == 10.0
|
||||
assert d["is_completed"] is True
|
||||
assert d["started_at"] is not None
|
||||
assert d["completed_at"] is not None
|
||||
|
||||
def test_failed_job_to_dict(self):
|
||||
job = TTSJob.create(user_id="u", input_text="t")
|
||||
job.mark_failed("出错了")
|
||||
d = job.to_dict()
|
||||
assert d["status"] == "failed"
|
||||
assert d["error_message"] == "出错了"
|
||||
assert d["is_retryable"] is True
|
||||
|
||||
Executable
+224
@@ -0,0 +1,224 @@
|
||||
"""VideoShare 领域层单元测试 - video_share.py"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.video_share import (
|
||||
VideoShare,
|
||||
_hash_password,
|
||||
generate_share_token,
|
||||
)
|
||||
|
||||
|
||||
class TestGenerateShareToken:
|
||||
"""generate_share_token 函数测试"""
|
||||
|
||||
def test_default_length(self):
|
||||
token = generate_share_token()
|
||||
assert len(token) == 12
|
||||
|
||||
def test_custom_length(self):
|
||||
token = generate_share_token(20)
|
||||
assert len(token) == 20
|
||||
|
||||
def test_url_friendly_characters(self):
|
||||
"""token 只包含 URL 友好的字符（没有 l, i, o, 0, 1 等易混字符）"""
|
||||
token = generate_share_token(100)
|
||||
# 不应包含易混淆字符
|
||||
assert "l" not in token
|
||||
assert "i" not in token
|
||||
assert "o" not in token
|
||||
assert "0" not in token
|
||||
assert "1" not in token
|
||||
|
||||
def test_randomness(self):
|
||||
"""两次生成的 token 不同（概率上）"""
|
||||
tokens = {generate_share_token() for _ in range(100)}
|
||||
# 100 次应该几乎不可能重复
|
||||
assert len(tokens) > 95
|
||||
|
||||
|
||||
class TestHashPassword:
|
||||
"""_hash_password 函数测试"""
|
||||
|
||||
def test_empty_password_returns_empty(self):
|
||||
assert _hash_password("") == ""
|
||||
|
||||
def test_none_password_returns_empty(self):
|
||||
assert _hash_password(None) == "" # type: ignore
|
||||
|
||||
def test_hash_is_deterministic(self):
|
||||
"""相同密码哈希结果相同"""
|
||||
h1 = _hash_password("mypassword")
|
||||
h2 = _hash_password("mypassword")
|
||||
assert h1 == h2
|
||||
|
||||
def test_hash_differs_for_different_passwords(self):
|
||||
"""不同密码哈希结果不同"""
|
||||
h1 = _hash_password("password1")
|
||||
h2 = _hash_password("password2")
|
||||
assert h1 != h2
|
||||
|
||||
def test_hash_is_hex_string(self):
|
||||
"""哈希是 64 位十六进制字符串（SHA-256）"""
|
||||
h = _hash_password("test")
|
||||
assert len(h) == 64
|
||||
int(h, 16) # 应该能解析为十六进制
|
||||
|
||||
def test_hash_includes_salt(self):
|
||||
"""加盐后与直接 SHA-256 不同"""
|
||||
from hashlib import sha256
|
||||
direct = sha256("mypass".encode()).hexdigest()
|
||||
salted = _hash_password("mypass")
|
||||
assert direct != salted
|
||||
|
||||
|
||||
class TestVideoShareCreate:
|
||||
"""VideoShare.create 工厂方法测试"""
|
||||
|
||||
def test_create_basic(self):
|
||||
share = VideoShare.create(video_id="video-1", user_id="user-1")
|
||||
assert share.id
|
||||
assert len(share.id) == 32
|
||||
assert share.video_id == "video-1"
|
||||
assert share.user_id == "user-1"
|
||||
assert share.share_token
|
||||
assert len(share.share_token) == 12
|
||||
assert share.password_hash is None
|
||||
assert share.expires_at is None
|
||||
assert share.view_count == 0
|
||||
assert share.download_count == 0
|
||||
assert share.is_active is True
|
||||
assert share.created_at is not None
|
||||
assert share.updated_at is not None
|
||||
|
||||
def test_create_empty_video_id_raises(self):
|
||||
with pytest.raises(ValueError, match="video_id cannot be empty"):
|
||||
VideoShare.create(video_id=" ", user_id="u1")
|
||||
|
||||
def test_create_empty_user_id_raises(self):
|
||||
with pytest.raises(ValueError, match="user_id cannot be empty"):
|
||||
VideoShare.create(video_id="v1", user_id=" ")
|
||||
|
||||
def test_create_with_password(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", password="secret123")
|
||||
assert share.password_hash is not None
|
||||
assert share.password_hash != "secret123" # 已哈希
|
||||
assert len(share.password_hash) == 64 # SHA-256 hex
|
||||
|
||||
def test_create_with_expires_at(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(days=7)
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future)
|
||||
assert share.expires_at == future
|
||||
|
||||
def test_create_past_expires_at_raises(self):
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
with pytest.raises(ValueError, match="expires_at cannot be in the past"):
|
||||
VideoShare.create(video_id="v1", user_id="u1", expires_at=past)
|
||||
|
||||
def test_create_fields_stripped(self):
|
||||
share = VideoShare.create(video_id=" v1 ", user_id=" u1 ")
|
||||
assert share.video_id == "v1"
|
||||
assert share.user_id == "u1"
|
||||
|
||||
def test_unique_tokens(self):
|
||||
"""不同分享有不同的 token"""
|
||||
shares = [VideoShare.create(video_id="v1", user_id="u1") for _ in range(20)]
|
||||
tokens = [s.share_token for s in shares]
|
||||
assert len(set(tokens)) == 20
|
||||
|
||||
|
||||
class TestVideoShareProperties:
|
||||
"""属性测试"""
|
||||
|
||||
def test_has_password_true(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", password="secret")
|
||||
assert share.has_password is True
|
||||
|
||||
def test_has_password_false(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.has_password is False
|
||||
|
||||
def test_is_expired_no_expiry(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.is_expired is False
|
||||
|
||||
def test_is_expired_future_expiry(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(days=7)
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future)
|
||||
assert share.is_expired is False
|
||||
|
||||
def test_is_expired_past_expiry(self):
|
||||
# 直接设置 expires_at 为过去时间（绕过 create 的校验）
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
assert share.is_expired is True
|
||||
|
||||
def test_is_accessible_active_not_expired(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.is_accessible is True
|
||||
|
||||
def test_is_accessible_inactive(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
share.is_active = False
|
||||
assert share.is_accessible is False
|
||||
|
||||
def test_is_accessible_expired(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
assert share.is_accessible is False
|
||||
|
||||
def test_is_accessible_inactive_and_expired(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
share.is_active = False
|
||||
share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
assert share.is_accessible is False
|
||||
|
||||
|
||||
class TestVideoSharePassword:
|
||||
"""密码验证测试"""
|
||||
|
||||
def test_verify_password_correct(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", password="secret123")
|
||||
assert share.verify_password("secret123") is True
|
||||
|
||||
def test_verify_password_wrong(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", password="secret123")
|
||||
assert share.verify_password("wrongpass") is False
|
||||
|
||||
def test_verify_no_password_always_true(self):
|
||||
"""没有设置密码时，任何密码都通过（包括空密码）"""
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.verify_password("") is True
|
||||
assert share.verify_password("anything") is True
|
||||
|
||||
def test_verify_empty_password_with_password_set(self):
|
||||
"""有密码时，空密码不通过"""
|
||||
share = VideoShare.create(video_id="v1", user_id="u1", password="secret")
|
||||
assert share.verify_password("") is False
|
||||
|
||||
|
||||
class TestVideoShareCounters:
|
||||
"""计数方法测试"""
|
||||
|
||||
def test_increment_view_count(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.view_count == 0
|
||||
share.increment_view_count()
|
||||
assert share.view_count == 1
|
||||
share.increment_view_count()
|
||||
assert share.view_count == 2
|
||||
|
||||
def test_increment_download_count(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.download_count == 0
|
||||
share.increment_download_count()
|
||||
assert share.download_count == 1
|
||||
|
||||
def test_revoke(self):
|
||||
share = VideoShare.create(video_id="v1", user_id="u1")
|
||||
assert share.is_active is True
|
||||
share.revoke()
|
||||
assert share.is_active is False
|
||||
assert share.is_accessible is False
|
||||
Regular → Executable
+249
-276
@@ -1,6 +1,4 @@
|
||||
"""
|
||||
VoiceCloneProfile 音色克隆档案领域模型单元测试
|
||||
"""
|
||||
"""VoiceCloneProfile 领域层单元测试 - voice_clone_profile.py"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -14,376 +12,351 @@ from packages.domain.voice_clone_profile import (
|
||||
class TestVoiceCloneStatus:
|
||||
"""VoiceCloneStatus 枚举测试"""
|
||||
|
||||
def test_status_values(self):
|
||||
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 VoiceCloneStatus.PROCESSING == "processing"
|
||||
assert VoiceCloneStatus.READY == "ready"
|
||||
assert VoiceCloneStatus.FAILED == "failed"
|
||||
assert VoiceCloneStatus.DISABLED == "disabled"
|
||||
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:
|
||||
"""VoiceCloneProfile.create 工厂方法测试"""
|
||||
"""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"
|
||||
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.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 == "描述"
|
||||
assert profile.metadata == {}
|
||||
|
||||
def test_create_empty_user_id_raises(self):
|
||||
with pytest.raises(ValueError, match="user_id 不能为空"):
|
||||
VoiceCloneProfile.create(user_id=" ", name="测试")
|
||||
VoiceCloneProfile.create(user_id=" ", name="test")
|
||||
|
||||
def test_create_empty_name_raises(self):
|
||||
with pytest.raises(ValueError, match="name 不能为空"):
|
||||
VoiceCloneProfile.create(user_id="user1", 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="user1", name=long_name)
|
||||
with pytest.raises(ValueError, match="name 长度不能超过 100"):
|
||||
VoiceCloneProfile.create(user_id="u1", name=long_name)
|
||||
|
||||
def test_create_name_exactly_100_chars_ok(self):
|
||||
def test_create_name_exactly_100_ok(self):
|
||||
name = "a" * 100
|
||||
profile = VoiceCloneProfile.create(user_id="user1", name=name)
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name=name)
|
||||
assert profile.name == name
|
||||
|
||||
def test_create_default_metadata_is_dict(self):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
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 == {}
|
||||
# 不应该共享同一个 dict
|
||||
p2 = VoiceCloneProfile.create(user_id="u2", name="test2")
|
||||
assert profile.metadata is not p2.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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.READY
|
||||
assert profile.is_terminal is True
|
||||
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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.FAILED
|
||||
assert profile.is_terminal is True
|
||||
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):
|
||||
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
|
||||
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):
|
||||
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
|
||||
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):
|
||||
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
|
||||
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_retryable_pending(self):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
assert profile.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_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_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_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
|
||||
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 TestStateTransitions:
|
||||
class TestVoiceCloneProfileTransitions:
|
||||
"""状态转换测试"""
|
||||
|
||||
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
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.transition_to(VoiceCloneStatus.PROCESSING)
|
||||
assert p.status == VoiceCloneStatus.PROCESSING
|
||||
|
||||
def test_pending_to_failed(self):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.transition_to(VoiceCloneStatus.FAILED)
|
||||
assert profile.status == VoiceCloneStatus.FAILED
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.transition_to(VoiceCloneStatus.FAILED)
|
||||
assert p.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
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.transition_to(VoiceCloneStatus.DISABLED)
|
||||
assert p.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
|
||||
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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.PROCESSING
|
||||
profile.transition_to(VoiceCloneStatus.FAILED)
|
||||
assert profile.status == VoiceCloneStatus.FAILED
|
||||
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):
|
||||
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
|
||||
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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.READY
|
||||
profile.transition_to(VoiceCloneStatus.DISABLED)
|
||||
assert profile.status == VoiceCloneStatus.DISABLED
|
||||
"""已就绪音色可以被禁用"""
|
||||
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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.transition_to("processing")
|
||||
assert profile.status == VoiceCloneStatus.PROCESSING
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.transition_to("processing")
|
||||
assert p.status == VoiceCloneStatus.PROCESSING
|
||||
|
||||
def test_transition_invalid_string_raises(self):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
with pytest.raises(ValueError, match="无效状态"):
|
||||
profile.transition_to("invalid_status")
|
||||
p.transition_to("invalid")
|
||||
|
||||
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)
|
||||
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 TestMarkMethods:
|
||||
"""标记方法测试"""
|
||||
class TestVoiceCloneProfileMarkMethods:
|
||||
"""便捷标记方法测试"""
|
||||
|
||||
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 == ""
|
||||
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):
|
||||
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 == ""
|
||||
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):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.PROCESSING
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.mark_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"
|
||||
p.mark_ready(" ")
|
||||
|
||||
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 == "音频质量太差"
|
||||
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_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
|
||||
def test_mark_disabled(self):
|
||||
p = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
p.mark_disabled()
|
||||
assert p.status == VoiceCloneStatus.DISABLED
|
||||
|
||||
|
||||
class TestPrepareRetry:
|
||||
"""重试准备测试"""
|
||||
class TestVoiceCloneProfileRetry:
|
||||
"""重试逻辑测试"""
|
||||
|
||||
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(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_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
|
||||
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="不可重试"):
|
||||
profile.prepare_retry()
|
||||
p.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()
|
||||
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 TestToDict:
|
||||
"""序列化测试"""
|
||||
class TestVoiceCloneProfileToDict:
|
||||
"""to_dict 序列化测试"""
|
||||
|
||||
def test_to_dict_basic(self):
|
||||
profile = VoiceCloneProfile.create(
|
||||
user_id="user1",
|
||||
name="测试音色",
|
||||
description="desc",
|
||||
max_retries=2,
|
||||
def test_to_dict_contains_fields(self):
|
||||
p = VoiceCloneProfile.create(
|
||||
user_id="u1",
|
||||
name="我的音色",
|
||||
description="测试",
|
||||
language="en-US",
|
||||
gender="female",
|
||||
)
|
||||
d = profile.to_dict()
|
||||
assert d["id"] == profile.id
|
||||
assert d["user_id"] == "user1"
|
||||
assert d["name"] == "测试音色"
|
||||
assert d["description"] == "desc"
|
||||
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["retry_count"] == 0
|
||||
assert d["max_retries"] == 2
|
||||
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_ready_state(self):
|
||||
profile = VoiceCloneProfile.create(user_id="u1", name="test")
|
||||
profile.status = VoiceCloneStatus.PROCESSING
|
||||
profile.mark_ready("voice_123")
|
||||
d = profile.to_dict()
|
||||
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["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}
|
||||
|
||||
Reference in New Issue
Block a user