xiaoxia-saas/tests/unit/test_tts_job_domain.py

"""TTSJob 领域层单元测试 - tts_job.py"""

import pytest

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


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

    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 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 工厂方法测试"""

    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-1"
        assert job.input_text == "你好世界"
        assert job.status == TTSJobStatus.PENDING
        assert job.voice_id == ""
        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 == {}

    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_empty_input_text_raises(self):
        with pytest.raises(ValueError, match="input_text 不能为空"):
            TTSJob.create(user_id="u1", input_text="  ")

    def test_create_text_too_long_raises(self):
        long_text = "a" * 10001
        with pytest.raises(ValueError, match="input_text 长度不能超过 10000"):
            TTSJob.create(user_id="u1", input_text=long_text)

    def test_create_text_exactly_10000_ok(self):
        text = "a" * 10000
        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="u1", input_text="test", format="flac")

    def test_create_mp3_format(self):
        job = TTSJob.create(user_id="u1", input_text="test", format="mp3")
        assert job.format == "mp3"

    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_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 TestTTSJobProperties:
    """属性测试"""

    def test_is_terminal_pending(self):
        job = TTSJob.create(user_id="u1", input_text="test")
        assert job.is_terminal is False

    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_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_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_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_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_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


class TestTTSJobTransitions:
    """状态转换测试"""

    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_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_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="非法状态转换"):
            job.transition_to(TTSJobStatus.COMPLETED)  # pending 不能直接到 completed

    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_transition_invalid_string_raises(self):
        job = TTSJob.create(user_id="u1", input_text="test")
        with pytest.raises(ValueError, match="无效状态"):
            job.transition_to("invalid")

    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)
        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:
    """重试逻辑测试"""

    def test_prepare_retry(self):
        job = TTSJob.create(user_id="u1", input_text="test", max_retries=3)
        job.mark_processing()
        job.mark_failed("超时")
        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_prepare_retry_not_retryable_raises(self):
        job = TTSJob.create(user_id="u1", input_text="test", max_retries=0)
        job.mark_processing()
        job.mark_failed("error")
        with pytest.raises(ValueError, match="不可重试"):
            job.prepare_retry()

    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()
            assert job.retry_count == i + 1
        # 第3次重试后 retry_count=3，等于 max_retries=3，不可再重试
        assert job.is_retryable is False


class TestTTSJobToDict:
    """to_dict 序列化测试"""

    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"] == "u1"
        assert d["input_text"] == "test"
        assert d["voice_id"] == "voice-1"
        assert d["status"] == "pending"
        assert d["is_retryable"] is False
        assert d["is_completed"] is False

    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

    def test_to_dict_after_completion(self):
        job = TTSJob.create(user_id="u1", input_text="test")
        job.mark_processing()
        job.mark_completed("http://test.mp3", duration=10.0)
        d = job.to_dict()
        assert d["status"] == "completed"
        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