xiaoxia
9554373f24
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Failing after 1364h38m32s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1364h38m35s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1364h38m36s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1364h38m37s
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been skipped
CI/CD Pipeline / Unit Tests (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Failing after 1365h10m45s
CI/CD Pipeline / Build Production API Image (push) Failing after 1365h10m47s
新增7个领域模块单元测试：classification、edit_plan_clip、filter_presets、generated_video、preset_bgm、template_clip_config、transition_presets，共231个用例
394 lines
15 KiB
Python
Executable File
394 lines
15 KiB
Python
Executable File
"""tts_job 领域模型单元测试."""
|
||
|
||
import pytest
|
||
from domain.tts_job import TERMINAL_STATUSES, TTSJob, TTSJobStatus
|
||
|
||
|
||
class TestTTSJobStatus:
|
||
"""TTSJobStatus 枚举测试."""
|
||
|
||
def test_values(self):
|
||
assert TTSJobStatus.PENDING == "pending"
|
||
assert TTSJobStatus.PROCESSING == "processing"
|
||
assert TTSJobStatus.COMPLETED == "completed"
|
||
assert TTSJobStatus.FAILED == "failed"
|
||
assert TTSJobStatus.CANCELLED == "cancelled"
|
||
|
||
def test_terminal_statuses(self):
|
||
assert TTSJobStatus.COMPLETED in TERMINAL_STATUSES
|
||
assert TTSJobStatus.FAILED in TERMINAL_STATUSES
|
||
assert TTSJobStatus.CANCELLED in TERMINAL_STATUSES
|
||
assert TTSJobStatus.PENDING not in TERMINAL_STATUSES
|
||
assert TTSJobStatus.PROCESSING not in TERMINAL_STATUSES
|
||
|
||
|
||
class TestTTSJobCreate:
|
||
"""TTSJob.create 工厂方法测试."""
|
||
|
||
def test_create_with_required_fields(self):
|
||
job = TTSJob.create(user_id="user_001", input_text="你好世界")
|
||
assert job.id
|
||
assert len(job.id) == 32
|
||
assert job.user_id == "user_001"
|
||
assert job.input_text == "你好世界"
|
||
assert job.status == TTSJobStatus.PENDING
|
||
assert job.voice_id == ""
|
||
assert job.sample_rate == 22050
|
||
assert job.format == "mp3"
|
||
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"):
|
||
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="")
|
||
|
||
def test_create_input_text_too_long_raises(self):
|
||
long_text = "a" * 10001
|
||
with pytest.raises(ValueError, match="10000"):
|
||
TTSJob.create(user_id="u", input_text=long_text)
|
||
|
||
def test_create_input_text_at_limit_ok(self):
|
||
text = "a" * 10000
|
||
job = TTSJob.create(user_id="u", input_text=text)
|
||
assert job.input_text == text
|
||
|
||
def test_create_invalid_format_raises(self):
|
||
with pytest.raises(ValueError, match="不支持的输出格式"):
|
||
TTSJob.create(user_id="u", input_text="t", 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_none_metadata_defaults_to_empty_dict(self):
|
||
job = TTSJob.create(user_id="u", input_text="t", 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
|
||
|
||
|
||
class TestTTSJobStateMachine:
|
||
"""TTSJob 状态机测试."""
|
||
|
||
@pytest.fixture
|
||
def pending_job(self):
|
||
return TTSJob.create(user_id="user_001", input_text="测试")
|
||
|
||
def test_initial_status_is_pending(self, pending_job):
|
||
assert pending_job.status == TTSJobStatus.PENDING
|
||
assert not pending_job.is_terminal
|
||
|
||
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_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_pending_can_be_cancelled(self, pending_job):
|
||
pending_job.mark_cancelled()
|
||
assert pending_job.status == TTSJobStatus.CANCELLED
|
||
|
||
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_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_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
|
||
|
||
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_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_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")
|
||
with pytest.raises(ValueError, match="非法状态转换"):
|
||
pending_job.mark_processing()
|
||
|
||
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_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):
|
||
with pytest.raises(ValueError, match="无效状态"):
|
||
pending_job.transition_to("invalid_status")
|
||
|
||
def test_state_transition_updates_updated_at(self, pending_job):
|
||
old_updated = pending_job.updated_at
|
||
import time
|
||
|
||
time.sleep(0.001)
|
||
pending_job.mark_processing()
|
||
assert pending_job.updated_at > old_updated
|
||
|
||
|
||
class TestTTSJobRetry:
|
||
"""TTSJob 重试逻辑测试."""
|
||
|
||
def test_failed_can_retry(self):
|
||
job = TTSJob.create(user_id="u", input_text="t", 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.prepare_retry()
|
||
assert job.status == TTSJobStatus.PENDING
|
||
assert job.retry_count == 1
|
||
assert job.error_message == ""
|
||
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，还可以重试
|
||
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")
|
||
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="不可重试"):
|
||
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
|
||
|
||
|
||
class TestTTSJobToDict:
|
||
"""to_dict 序列化测试."""
|
||
|
||
def test_pending_job_to_dict(self):
|
||
job = TTSJob.create(user_id="user_001", input_text="测试文本", voice_id="v001")
|
||
d = job.to_dict()
|
||
assert d["id"] == job.id
|
||
assert d["user_id"] == "user_001"
|
||
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"] == {}
|
||
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")
|
||
job.mark_processing()
|
||
job.mark_completed(output_audio_url="https://example.com/out.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
|