xiaoxia
aa0b0de90f
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
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 / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
404 lines
14 KiB
Python
Executable File
404 lines
14 KiB
Python
Executable File
"""TTSJob领域模型测试 — 状态机 + 状态转换 + 属性方法."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from packages.domain.tts_job import (
|
||
TERMINAL_STATUSES,
|
||
TTSJob,
|
||
TTSJobStatus,
|
||
)
|
||
|
||
|
||
def _make_job(
|
||
*,
|
||
status: TTSJobStatus = TTSJobStatus.PENDING,
|
||
retry_count: int = 0,
|
||
max_retries: int = 3,
|
||
output_audio_url: str = "",
|
||
) -> TTSJob:
|
||
"""快速创建测试用TTSJob."""
|
||
return TTSJob(
|
||
id="job_1",
|
||
user_id="user_1",
|
||
input_text="测试文本",
|
||
voice_id="female_warm",
|
||
status=status,
|
||
output_audio_url=output_audio_url,
|
||
retry_count=retry_count,
|
||
max_retries=max_retries,
|
||
)
|
||
|
||
|
||
class TestTTSJobStatus:
|
||
"""TTSJobStatus 枚举测试."""
|
||
|
||
def test_all_statuses_exist(self):
|
||
"""所有5种状态都存在."""
|
||
assert TTSJobStatus.PENDING.value == "pending"
|
||
assert TTSJobStatus.PROCESSING.value == "processing"
|
||
assert TTSJobStatus.COMPLETED.value == "completed"
|
||
assert TTSJobStatus.FAILED.value == "failed"
|
||
assert TTSJobStatus.CANCELLED.value == "cancelled"
|
||
|
||
def test_terminal_statuses(self):
|
||
"""终态集合包含COMPLETED/FAILED/CANCELLED."""
|
||
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_pending_job(self):
|
||
"""创建成功，默认PENDING状态."""
|
||
job = TTSJob.create(
|
||
user_id="u1",
|
||
input_text="你好世界",
|
||
voice_id="v1",
|
||
)
|
||
assert job.id is not None
|
||
assert len(job.id) == 32
|
||
assert job.user_id == "u1"
|
||
assert job.input_text == "你好世界"
|
||
assert job.voice_id == "v1"
|
||
assert job.status == TTSJobStatus.PENDING
|
||
assert job.retry_count == 0
|
||
assert job.max_retries == 3
|
||
assert job.sample_rate == 22050
|
||
|
||
def test_create_empty_user_id_raises(self):
|
||
"""空user_id抛异常."""
|
||
with pytest.raises(ValueError, match="user_id"):
|
||
TTSJob.create(user_id=" ", input_text="hi", voice_id="v")
|
||
|
||
def test_create_empty_text_raises(self):
|
||
"""空input_text抛异常."""
|
||
with pytest.raises(ValueError, match="input_text"):
|
||
TTSJob.create(user_id="u1", input_text="", voice_id="v")
|
||
|
||
def test_create_strips_whitespace(self):
|
||
"""首尾空白被去除."""
|
||
job = TTSJob.create(
|
||
user_id=" u1 ",
|
||
input_text=" 你好 ",
|
||
voice_id=" v1 ",
|
||
)
|
||
assert job.user_id == "u1"
|
||
assert job.input_text == "你好"
|
||
assert job.voice_id == "v1"
|
||
|
||
|
||
class TestTerminalStatus:
|
||
"""is_terminal 终态判定测试."""
|
||
|
||
def test_pending_not_terminal(self):
|
||
"""PENDING不是终态."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
assert job.is_terminal is False
|
||
|
||
def test_processing_not_terminal(self):
|
||
"""PROCESSING不是终态."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
assert job.is_terminal is False
|
||
|
||
def test_completed_is_terminal(self):
|
||
"""COMPLETED是终态."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED)
|
||
assert job.is_terminal is True
|
||
|
||
def test_failed_is_terminal(self):
|
||
"""FAILED是终态."""
|
||
job = _make_job(status=TTSJobStatus.FAILED)
|
||
assert job.is_terminal is True
|
||
|
||
def test_cancelled_is_terminal(self):
|
||
"""CANCELLED是终态."""
|
||
job = _make_job(status=TTSJobStatus.CANCELLED)
|
||
assert job.is_terminal is True
|
||
|
||
|
||
class TestIsRetryable:
|
||
"""is_retryable 可重试判定测试."""
|
||
|
||
def test_failed_within_limit_is_retryable(self):
|
||
"""失败且未超上限→可重试."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=1, max_retries=3)
|
||
assert job.is_retryable is True
|
||
|
||
def test_failed_at_limit_not_retryable(self):
|
||
"""失败且已达上限→不可重试."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=3, max_retries=3)
|
||
assert job.is_retryable is False
|
||
|
||
def test_failed_over_limit_not_retryable(self):
|
||
"""失败且超上限→不可重试."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=5, max_retries=3)
|
||
assert job.is_retryable is False
|
||
|
||
def test_pending_not_retryable(self):
|
||
"""PENDING不可重试."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
assert job.is_retryable is False
|
||
|
||
def test_completed_not_retryable(self):
|
||
"""COMPLETED不可重试."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED)
|
||
assert job.is_retryable is False
|
||
|
||
def test_cancelled_not_retryable(self):
|
||
"""CANCELLED不可重试."""
|
||
job = _make_job(status=TTSJobStatus.CANCELLED)
|
||
assert job.is_retryable is False
|
||
|
||
|
||
class TestIsCompleted:
|
||
"""is_completed 完成判定测试."""
|
||
|
||
def test_completed_with_output_is_completed(self):
|
||
"""COMPLETED + 有输出URL→已完成."""
|
||
job = _make_job(
|
||
status=TTSJobStatus.COMPLETED,
|
||
output_audio_url="https://example.com/out.mp3",
|
||
)
|
||
assert job.is_completed is True
|
||
|
||
def test_completed_without_output_not_completed(self):
|
||
"""COMPLETED但无输出URL→不算完成."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED, output_audio_url="")
|
||
assert job.is_completed is False
|
||
|
||
def test_pending_not_completed(self):
|
||
"""PENDING不是完成."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
assert job.is_completed is False
|
||
|
||
def test_failed_not_completed(self):
|
||
"""FAILED不是完成."""
|
||
job = _make_job(status=TTSJobStatus.FAILED)
|
||
assert job.is_completed is False
|
||
|
||
|
||
class TestStateTransitions:
|
||
"""状态机转换测试."""
|
||
|
||
def test_pending_to_processing(self):
|
||
"""PENDING → PROCESSING 合法."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.transition_to(TTSJobStatus.PROCESSING)
|
||
assert job.status == TTSJobStatus.PROCESSING
|
||
|
||
def test_pending_to_failed(self):
|
||
"""PENDING → FAILED 合法."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.transition_to(TTSJobStatus.FAILED)
|
||
assert job.status == TTSJobStatus.FAILED
|
||
|
||
def test_pending_to_cancelled(self):
|
||
"""PENDING → CANCELLED 合法."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.transition_to(TTSJobStatus.CANCELLED)
|
||
assert job.status == TTSJobStatus.CANCELLED
|
||
|
||
def test_pending_to_completed_illegal(self):
|
||
"""PENDING → COMPLETED 非法."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
with pytest.raises(ValueError, match="非法状态转换"):
|
||
job.transition_to(TTSJobStatus.COMPLETED)
|
||
|
||
def test_processing_to_completed(self):
|
||
"""PROCESSING → COMPLETED 合法."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.transition_to(TTSJobStatus.COMPLETED)
|
||
assert job.status == TTSJobStatus.COMPLETED
|
||
|
||
def test_processing_to_failed(self):
|
||
"""PROCESSING → FAILED 合法."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.transition_to(TTSJobStatus.FAILED)
|
||
assert job.status == TTSJobStatus.FAILED
|
||
|
||
def test_failed_to_pending_retry(self):
|
||
"""FAILED → PENDING 合法（重试）."""
|
||
job = _make_job(status=TTSJobStatus.FAILED)
|
||
job.transition_to(TTSJobStatus.PENDING)
|
||
assert job.status == TTSJobStatus.PENDING
|
||
|
||
def test_failed_to_completed_illegal(self):
|
||
"""FAILED → COMPLETED 非法."""
|
||
job = _make_job(status=TTSJobStatus.FAILED)
|
||
with pytest.raises(ValueError):
|
||
job.transition_to(TTSJobStatus.COMPLETED)
|
||
|
||
def test_completed_to_anything_illegal(self):
|
||
"""COMPLETED → 任何状态都非法（终态不可转换）."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED)
|
||
with pytest.raises(ValueError):
|
||
job.transition_to(TTSJobStatus.FAILED)
|
||
with pytest.raises(ValueError):
|
||
job.transition_to(TTSJobStatus.PENDING)
|
||
|
||
def test_cancelled_to_anything_illegal(self):
|
||
"""CANCELLED → 任何状态都非法."""
|
||
job = _make_job(status=TTSJobStatus.CANCELLED)
|
||
with pytest.raises(ValueError):
|
||
job.transition_to(TTSJobStatus.PENDING)
|
||
|
||
def test_transition_updates_updated_at(self):
|
||
"""状态转换更新updated_at."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
old_updated = job.updated_at
|
||
job.transition_to(TTSJobStatus.PROCESSING)
|
||
assert job.updated_at >= old_updated
|
||
|
||
def test_transition_with_string_status(self):
|
||
"""字符串状态也能转换."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.transition_to("processing")
|
||
assert job.status == TTSJobStatus.PROCESSING
|
||
|
||
def test_transition_with_invalid_string_raises(self):
|
||
"""无效字符串状态抛异常."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
with pytest.raises(ValueError, match="无效状态"):
|
||
job.transition_to("invalid_status")
|
||
|
||
|
||
class TestMarkProcessing:
|
||
"""mark_processing 标记处理中测试."""
|
||
|
||
def test_mark_processing_sets_status_and_time(self):
|
||
"""标记处理中更新状态+开始时间+清错误."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.error_message = "旧错误"
|
||
job.mark_processing()
|
||
assert job.status == TTSJobStatus.PROCESSING
|
||
assert job.started_at is not None
|
||
assert job.error_message == ""
|
||
|
||
|
||
class TestMarkCompleted:
|
||
"""mark_completed 标记完成测试."""
|
||
|
||
def test_mark_completed_success(self):
|
||
"""成功标记完成."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.mark_completed(
|
||
output_audio_url="https://example.com/out.mp3",
|
||
output_audio_key="tts/jobs/job_1/out.mp3",
|
||
duration=10.5,
|
||
file_size=204800,
|
||
)
|
||
assert job.status == TTSJobStatus.COMPLETED
|
||
assert job.output_audio_url == "https://example.com/out.mp3"
|
||
assert job.output_audio_key == "tts/jobs/job_1/out.mp3"
|
||
assert job.duration == pytest.approx(10.5)
|
||
assert job.file_size == 204800
|
||
assert job.completed_at is not None
|
||
assert job.error_message == ""
|
||
|
||
def test_mark_completed_empty_url_raises(self):
|
||
"""空URL抛异常."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
with pytest.raises(ValueError, match="output_audio_url"):
|
||
job.mark_completed(output_audio_url=" ")
|
||
|
||
def test_mark_completed_strips_url(self):
|
||
"""URL首尾空白被去除."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.mark_completed(output_audio_url=" https://example.com/out.mp3 ")
|
||
assert job.output_audio_url == "https://example.com/out.mp3"
|
||
|
||
def test_mark_completed_from_pending_illegal(self):
|
||
"""从PENDING直接标记完成非法（先processing）."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
with pytest.raises(ValueError, match="非法状态转换"):
|
||
job.mark_completed(output_audio_url="https://x.com/out.mp3")
|
||
|
||
|
||
class TestMarkFailed:
|
||
"""mark_failed 标记失败测试."""
|
||
|
||
def test_mark_failed_from_pending(self):
|
||
"""从PENDING标记失败."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.mark_failed("网络超时")
|
||
assert job.status == TTSJobStatus.FAILED
|
||
assert job.error_message == "网络超时"
|
||
|
||
def test_mark_failed_from_processing(self):
|
||
"""从PROCESSING标记失败."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.mark_failed("合成失败")
|
||
assert job.status == TTSJobStatus.FAILED
|
||
assert job.error_message == "合成失败"
|
||
|
||
def test_mark_failed_from_completed_illegal(self):
|
||
"""从COMPLETED标记失败非法."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED, output_audio_url="https://x.com/out.mp3")
|
||
with pytest.raises(ValueError):
|
||
job.mark_failed("错误")
|
||
|
||
|
||
class TestMarkCancelled:
|
||
"""mark_cancelled 标记取消测试."""
|
||
|
||
def test_cancel_from_pending(self):
|
||
"""从PENDING取消."""
|
||
job = _make_job(status=TTSJobStatus.PENDING)
|
||
job.mark_cancelled()
|
||
assert job.status == TTSJobStatus.CANCELLED
|
||
|
||
def test_cancel_from_processing(self):
|
||
"""从PROCESSING取消."""
|
||
job = _make_job(status=TTSJobStatus.PROCESSING)
|
||
job.mark_cancelled()
|
||
assert job.status == TTSJobStatus.CANCELLED
|
||
|
||
def test_cancel_from_completed_illegal(self):
|
||
"""从COMPLETED取消非法."""
|
||
job = _make_job(status=TTSJobStatus.COMPLETED, output_audio_url="https://x.com/out.mp3")
|
||
with pytest.raises(ValueError):
|
||
job.mark_cancelled()
|
||
|
||
|
||
class TestPrepareRetry:
|
||
"""prepare_retry 重试准备测试."""
|
||
|
||
def test_retry_resets_to_pending(self):
|
||
"""重试重置为PENDING，retry_count+1."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=0, max_retries=3)
|
||
job.error_message = "失败了"
|
||
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_at_max_raises(self):
|
||
"""已达重试上限时不能再重试."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=3, max_retries=3)
|
||
with pytest.raises(ValueError, match="不可重试"):
|
||
job.prepare_retry()
|
||
|
||
def test_retry_from_pending_raises(self):
|
||
"""PENDING状态不能重试."""
|
||
job = _make_job(status=TTSJobStatus.PENDING, retry_count=0, max_retries=3)
|
||
with pytest.raises(ValueError):
|
||
job.prepare_retry()
|
||
|
||
def test_multiple_retries_increment(self):
|
||
"""多次重试计数递增."""
|
||
job = _make_job(status=TTSJobStatus.FAILED, retry_count=0, max_retries=5)
|
||
job.prepare_retry()
|
||
assert job.retry_count == 1
|
||
# 模拟再次失败
|
||
job.mark_failed("又失败了")
|
||
job.prepare_retry()
|
||
assert job.retry_count == 2
|