xiaoxia-saas/tests/unit/test_tts_job.py

"""TTSJob 领域模型单元测试."""

from __future__ import annotations

from datetime import datetime, timezone
from time import sleep

import pytest

from packages.domain.tts_job import (
    TERMINAL_STATUSES,
    TTSJob,
    TTSJobStatus,
)

# ── 枚举测试 ──────────────────────────────────────────────────────────────────


class TestTTSJobStatus:
    """TTSJobStatus 枚举测试."""

    def test_status_values(self):
        """状态值正确."""
        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_status_count(self):
        """共5种状态."""
        assert len(TTSJobStatus) == 5

    def test_is_str_enum(self):
        """是StrEnum，可与字符串直接比较."""
        assert TTSJobStatus.PENDING == "pending"
        assert TTSJobStatus.COMPLETED == "completed"

    def test_from_string(self):
        """从字符串构建枚举."""
        assert TTSJobStatus("pending") == TTSJobStatus.PENDING
        assert TTSJobStatus("completed") == TTSJobStatus.COMPLETED

    def test_from_string_invalid(self):
        """无效字符串抛出ValueError."""
        with pytest.raises(ValueError):
            TTSJobStatus("invalid_status")


# ── 终态集合测试 ──────────────────────────────────────────────────────────────


class TestTerminalStatuses:
    """TERMINAL_STATUSES 终态集合测试."""

    def test_completed_is_terminal(self):
        """completed是终态."""
        assert TTSJobStatus.COMPLETED in TERMINAL_STATUSES

    def test_failed_is_terminal(self):
        """failed是终态."""
        assert TTSJobStatus.FAILED in TERMINAL_STATUSES

    def test_cancelled_is_terminal(self):
        """cancelled是终态."""
        assert TTSJobStatus.CANCELLED in TERMINAL_STATUSES

    def test_pending_not_terminal(self):
        """pending不是终态."""
        assert TTSJobStatus.PENDING not in TERMINAL_STATUSES

    def test_processing_not_terminal(self):
        """processing不是终态."""
        assert TTSJobStatus.PROCESSING not in TERMINAL_STATUSES

    def test_terminal_count(self):
        """共3个终态."""
        assert len(TERMINAL_STATUSES) == 3


# ── 工厂方法测试 ──────────────────────────────────────────────────────────────


class TestTTSJobCreate:
    """TTSJob.create 工厂方法测试."""

    def test_create_basic(self):
        """基础创建."""
        job = TTSJob.create(user_id="user123", input_text="你好世界")
        assert job.id
        assert job.user_id == "user123"
        assert job.input_text == "你好世界"
        assert job.status == TTSJobStatus.PENDING
        assert job.retry_count == 0
        assert job.max_retries == 3

    def test_create_with_voice_id(self):
        """带音色ID创建."""
        job = TTSJob.create(user_id="u1", input_text="hi", voice_id="voice_001")
        assert job.voice_id == "voice_001"

    def test_create_with_project_id(self):
        """带项目ID创建."""
        job = TTSJob.create(user_id="u1", input_text="hi", project_id="proj_001")
        assert job.project_id == "proj_001"

    def test_create_with_voice_clone_profile(self):
        """带音色克隆档案ID创建."""
        job = TTSJob.create(user_id="u1", input_text="hi", voice_clone_profile_id="vcp_001")
        assert job.voice_clone_profile_id == "vcp_001"

    def test_create_custom_sample_rate(self):
        """自定义采样率."""
        job = TTSJob.create(user_id="u1", input_text="hi", sample_rate=44100)
        assert job.sample_rate == 44100

    def test_create_default_sample_rate(self):
        """默认采样率22050."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.sample_rate == 22050

    @pytest.mark.parametrize("fmt", ["mp3", "wav", "pcm"])
    def test_create_valid_formats(self, fmt: str):
        """支持的输出格式."""
        job = TTSJob.create(user_id="u1", input_text="hi", format=fmt)
        assert job.format == fmt

    def test_create_default_format(self):
        """默认格式mp3."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.format == "mp3"

    def test_create_invalid_format(self):
        """不支持的格式抛错."""
        with pytest.raises(ValueError, match="不支持的输出格式"):
            TTSJob.create(user_id="u1", input_text="hi", format="aac")

    def test_create_custom_max_retries(self):
        """自定义最大重试次数."""
        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=5)
        assert job.max_retries == 5

    def test_create_metadata(self):
        """元数据."""
        meta = {"priority": "high", "source": "api"}
        job = TTSJob.create(user_id="u1", input_text="hi", metadata=meta)
        assert job.metadata == meta

    def test_create_metadata_none(self):
        """metadata为None时默认为空dict."""
        job = TTSJob.create(user_id="u1", input_text="hi", metadata=None)
        assert job.metadata == {}

    def test_create_strips_text(self):
        """input_text去除首尾空白."""
        job = TTSJob.create(user_id="u1", input_text="  你好世界  ")
        assert job.input_text == "你好世界"

    def test_create_strips_user_id(self):
        """user_id去除空白."""
        job = TTSJob.create(user_id="  user123  ", input_text="hi")
        assert job.user_id == "user123"

    def test_create_empty_user_id(self):
        """空user_id抛错."""
        with pytest.raises(ValueError, match="user_id"):
            TTSJob.create(user_id="", input_text="hi")

    def test_create_whitespace_user_id(self):
        """纯空白user_id抛错."""
        with pytest.raises(ValueError, match="user_id"):
            TTSJob.create(user_id="   ", input_text="hi")

    def test_create_empty_input_text(self):
        """空input_text抛错."""
        with pytest.raises(ValueError, match="input_text"):
            TTSJob.create(user_id="u1", input_text="")

    def test_create_whitespace_input_text(self):
        """纯空白input_text抛错."""
        with pytest.raises(ValueError, match="input_text"):
            TTSJob.create(user_id="u1", input_text="  \n  ")

    def test_create_input_text_too_long(self):
        """input_text超过10000字符抛错."""
        long_text = "a" * 10001
        with pytest.raises(ValueError, match="10000"):
            TTSJob.create(user_id="u1", input_text=long_text)

    def test_create_input_text_exactly_10000(self):
        """input_text恰好10000字符正常."""
        text = "a" * 10000
        job = TTSJob.create(user_id="u1", input_text=text)
        assert job.input_text == text

    def test_create_has_timestamps(self):
        """创建后有时间戳."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert isinstance(job.created_at, datetime)
        assert isinstance(job.updated_at, datetime)
        assert job.created_at.tzinfo is not None
        assert job.started_at is None
        assert job.completed_at is None

    def test_create_default_values(self):
        """默认值正确."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.voice_id == ""
        assert job.voice_model == ""
        assert job.project_id == ""
        assert job.voice_clone_profile_id == ""
        assert job.output_audio_url == ""
        assert job.output_audio_key == ""
        assert job.duration == 0.0
        assert job.file_size == 0

    def test_create_id_is_hex(self):
        """id是32位hex字符串."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert len(job.id) == 32
        int(job.id, 16)


# ── 属性测试 ──────────────────────────────────────────────────────────────────


class TestTTSJobProperties:
    """TTSJob 属性测试."""

    def test_is_terminal_pending(self):
        """pending不是终态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.is_terminal is False

    def test_is_terminal_processing(self):
        """processing不是终态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        assert job.is_terminal is False

    def test_is_terminal_completed(self):
        """completed是终态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("https://example.com/audio.mp3")
        assert job.is_terminal is True

    def test_is_terminal_failed(self):
        """failed是终态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("超时")
        assert job.is_terminal is True

    def test_is_terminal_cancelled(self):
        """cancelled是终态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_cancelled()
        assert job.is_terminal is True

    def test_is_retryable_failed_within_limit(self):
        """失败且未超过重试次数，可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3)
        job.mark_processing()
        job.mark_failed("error")
        assert job.is_retryable is True

    def test_is_retryable_failed_at_limit(self):
        """失败但已达重试上限，不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=1)
        job.mark_processing()
        job.mark_failed("e1")
        job.prepare_retry()  # retry_count=1
        job.mark_processing()
        job.mark_failed("e2")
        assert job.is_retryable is False

    def test_is_retryable_pending(self):
        """pending状态不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.is_retryable is False

    def test_is_retryable_completed(self):
        """completed状态不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("https://a.mp3")
        assert job.is_retryable is False

    def test_is_retryable_cancelled(self):
        """cancelled状态不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_cancelled()
        assert job.is_retryable is False

    def test_is_completed_with_output(self):
        """completed状态且有输出URL，is_completed为True."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("https://example.com/audio.mp3")
        assert job.is_completed is True

    def test_is_completed_no_output(self):
        """completed状态但无输出URL，is_completed为False."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.status = TTSJobStatus.COMPLETED
        job.output_audio_url = ""
        assert job.is_completed is False

    def test_is_completed_pending(self):
        """pending状态is_completed为False."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.is_completed is False

    def test_is_completed_failed(self):
        """failed状态is_completed为False."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("err")
        assert job.is_completed is False


# ── 状态转换测试 ──────────────────────────────────────────────────────────────


class TestTransitionTo:
    """transition_to 状态转换测试."""

    def test_pending_to_processing(self):
        """pending → processing 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.transition_to(TTSJobStatus.PROCESSING)
        assert job.status == TTSJobStatus.PROCESSING

    def test_pending_to_failed(self):
        """pending → failed 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.transition_to(TTSJobStatus.FAILED)
        assert job.status == TTSJobStatus.FAILED

    def test_pending_to_cancelled(self):
        """pending → cancelled 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.transition_to(TTSJobStatus.CANCELLED)
        assert job.status == TTSJobStatus.CANCELLED

    def test_pending_to_completed_invalid(self):
        """pending → completed 非法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        with pytest.raises(ValueError, match="非法状态转换"):
            job.transition_to(TTSJobStatus.COMPLETED)

    def test_processing_to_completed(self):
        """processing → completed 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.transition_to(TTSJobStatus.COMPLETED)
        assert job.status == TTSJobStatus.COMPLETED

    def test_processing_to_failed(self):
        """processing → failed 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.transition_to(TTSJobStatus.FAILED)
        assert job.status == TTSJobStatus.FAILED

    def test_processing_to_cancelled(self):
        """processing → cancelled 合法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.transition_to(TTSJobStatus.CANCELLED)
        assert job.status == TTSJobStatus.CANCELLED

    def test_failed_to_pending(self):
        """failed → pending 合法（重试）."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("err")
        job.transition_to(TTSJobStatus.PENDING)
        assert job.status == TTSJobStatus.PENDING

    def test_failed_to_completed_invalid(self):
        """failed → completed 非法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("err")
        with pytest.raises(ValueError):
            job.transition_to(TTSJobStatus.COMPLETED)

    def test_completed_to_pending_invalid(self):
        """completed → pending 非法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("https://a.mp3")
        with pytest.raises(ValueError):
            job.transition_to(TTSJobStatus.PENDING)

    def test_cancelled_to_pending_invalid(self):
        """cancelled → pending 非法."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_cancelled()
        with pytest.raises(ValueError):
            job.transition_to(TTSJobStatus.PENDING)

    def test_transition_with_string(self):
        """字符串输入的状态转换."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.transition_to("processing")
        assert job.status == TTSJobStatus.PROCESSING

    def test_transition_with_invalid_string(self):
        """无效字符串状态抛错."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        with pytest.raises(ValueError, match="无效状态"):
            job.transition_to("invalid")

    def test_transition_updates_updated_at(self):
        """状态转换更新updated_at."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        old_updated = job.updated_at
        sleep(0.01)
        job.transition_to(TTSJobStatus.PROCESSING)
        assert job.updated_at > old_updated

    def test_transition_error_message_contains_statuses(self):
        """错误信息包含源状态和目标状态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        with pytest.raises(ValueError) as exc_info:
            job.transition_to(TTSJobStatus.COMPLETED)
        msg = str(exc_info.value)
        assert "pending" in msg
        assert "completed" in msg


# ── 操作方法测试 ──────────────────────────────────────────────────────────────


class TestMarkMethods:
    """mark_* 系列方法测试."""

    def test_mark_processing_sets_started_at(self):
        """mark_processing 设置started_at."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        assert job.started_at is None
        job.mark_processing()
        assert job.started_at is not None
        assert isinstance(job.started_at, datetime)

    def test_mark_processing_clears_error(self):
        """mark_processing 清除错误信息."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.error_message = "previous error"
        job.mark_processing()
        assert job.error_message == ""

    def test_mark_completed_sets_fields(self):
        """mark_completed 设置所有输出字段."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed(
            "https://example.com/out.mp3",
            output_audio_key="audio/001.mp3",
            duration=10.5,
            file_size=256000,
        )
        assert job.output_audio_url == "https://example.com/out.mp3"
        assert job.output_audio_key == "audio/001.mp3"
        assert job.duration == 10.5
        assert job.file_size == 256000

    def test_mark_completed_sets_completed_at(self):
        """mark_completed 设置completed_at."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        assert job.completed_at is None
        job.mark_completed("https://a.mp3")
        assert job.completed_at is not None
        assert isinstance(job.completed_at, datetime)

    def test_mark_completed_clears_error(self):
        """mark_completed 清除错误信息."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.error_message = "temp error"
        job.mark_completed("https://a.mp3")
        assert job.error_message == ""

    def test_mark_completed_empty_url(self):
        """空output_audio_url抛错."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        with pytest.raises(ValueError, match="output_audio_url"):
            job.mark_completed("")

    def test_mark_completed_whitespace_url(self):
        """纯空白URL抛错."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        with pytest.raises(ValueError):
            job.mark_completed("   ")

    def test_mark_completed_strips_url(self):
        """URL去除空白."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("  https://a.mp3  ")
        assert job.output_audio_url == "https://a.mp3"

    def test_mark_failed_sets_error(self):
        """mark_failed 设置错误信息."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("连接超时")
        assert job.error_message == "连接超时"

    def test_mark_failed_from_pending(self):
        """从pending直接失败."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_failed("验证失败")
        assert job.status == TTSJobStatus.FAILED
        assert job.error_message == "验证失败"

    def test_mark_cancelled_from_pending(self):
        """从pending取消."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_cancelled()
        assert job.status == TTSJobStatus.CANCELLED

    def test_mark_cancelled_from_processing(self):
        """从processing取消."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_cancelled()
        assert job.status == TTSJobStatus.CANCELLED


# ── 重试逻辑测试 ──────────────────────────────────────────────────────────────


class TestPrepareRetry:
    """prepare_retry 重试逻辑测试."""

    def test_prepare_retry_basic(self):
        """基础重试成功."""
        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3)
        job.mark_processing()
        job.mark_failed("err")
        job.prepare_retry()
        assert job.status == TTSJobStatus.PENDING
        assert job.retry_count == 1

    def test_prepare_retry_clears_error(self):
        """重试清除错误信息."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("big error")
        job.prepare_retry()
        assert job.error_message == ""

    def test_prepare_retry_clears_started_at(self):
        """重试清除started_at."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        assert job.started_at is not None
        job.mark_failed("err")
        job.prepare_retry()
        assert job.started_at is None

    def test_prepare_retry_clears_completed_at(self):
        """重试清除completed_at."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("err")
        job.completed_at = datetime.now(timezone.utc)  # 模拟设置过
        job.prepare_retry()
        assert job.completed_at is None

    def test_prepare_retry_not_failed(self):
        """非failed状态不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        with pytest.raises(ValueError, match="不可重试"):
            job.prepare_retry()

    def test_prepare_retry_exceeds_max(self):
        """超过最大重试次数不可重试."""
        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=1)
        job.mark_processing()
        job.mark_failed("e1")
        job.prepare_retry()  # retry_count=1
        job.mark_processing()
        job.mark_failed("e2")
        with pytest.raises(ValueError, match="不可重试"):
            job.prepare_retry()

    def test_prepare_retry_error_has_details(self):
        """错误信息包含详细状态."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        with pytest.raises(ValueError) as exc_info:
            job.prepare_retry()
        msg = str(exc_info.value)
        assert "pending" in msg
        assert "retry_count" in msg
        assert "max_retries" in msg


# ── 序列化测试 ────────────────────────────────────────────────────────────────


class TestToDict:
    """to_dict 序列化测试."""

    def test_to_dict_keys(self):
        """序列化字典包含所有预期字段."""
        job = TTSJob.create(user_id="u1", input_text="测试文本")
        d = job.to_dict()
        expected_keys = {
            "id",
            "user_id",
            "project_id",
            "voice_clone_profile_id",
            "status",
            "input_text",
            "voice_id",
            "voice_model",
            "output_audio_url",
            "output_audio_key",
            "duration",
            "file_size",
            "sample_rate",
            "format",
            "error_message",
            "retry_count",
            "max_retries",
            "is_retryable",
            "is_completed",
            "metadata",
            "started_at",
            "completed_at",
            "created_at",
            "updated_at",
        }
        assert set(d.keys()) == expected_keys

    def test_to_dict_values(self):
        """序列化值正确."""
        job = TTSJob.create(
            user_id="user123",
            input_text="你好世界",
            voice_id="voice_001",
            project_id="proj_001",
            sample_rate=44100,
            format="wav",
            max_retries=5,
            metadata={"source": "api"},
        )
        d = job.to_dict()
        assert d["user_id"] == "user123"
        assert d["input_text"] == "你好世界"
        assert d["voice_id"] == "voice_001"
        assert d["project_id"] == "proj_001"
        assert d["status"] == "pending"
        assert d["sample_rate"] == 44100
        assert d["format"] == "wav"
        assert d["retry_count"] == 0
        assert d["max_retries"] == 5
        assert d["is_retryable"] is False
        assert d["is_completed"] is False
        assert d["metadata"] == {"source": "api"}

    def test_to_dict_completed_status(self):
        """completed状态下序列化正确."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_completed("https://example.com/out.mp3", duration=5.5, file_size=128000)
        d = job.to_dict()
        assert d["status"] == "completed"
        assert d["output_audio_url"] == "https://example.com/out.mp3"
        assert d["duration"] == 5.5
        assert d["file_size"] == 128000
        assert d["is_completed"] is True
        assert d["is_retryable"] is False
        assert d["started_at"] is not None
        assert d["completed_at"] is not None

    def test_to_dict_failed_status(self):
        """failed状态下序列化正确."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        job.mark_failed("超时错误")
        d = job.to_dict()
        assert d["status"] == "failed"
        assert d["error_message"] == "超时错误"
        assert d["is_retryable"] is True
        assert d["is_completed"] is False

    def test_to_dict_nullable_times(self):
        """空时间字段为None."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        d = job.to_dict()
        assert d["started_at"] is None
        assert d["completed_at"] is None

    def test_to_dict_datetime_format(self):
        """时间字段是ISO格式字符串."""
        job = TTSJob.create(user_id="u1", input_text="hi")
        job.mark_processing()
        d = job.to_dict()
        datetime.fromisoformat(d["created_at"])
        datetime.fromisoformat(d["updated_at"])
        datetime.fromisoformat(d["started_at"])