From ad66a59f581b6af05b222d4dea662072060885fc Mon Sep 17 00:00:00 2001
From: xiaoxia <xiaoxia@xiaoxiajianji.com>
Date: Tue, 28 Jul 2026 16:16:18 +0800
Subject: [PATCH] =?UTF-8?q?test(wave174):=20tts=5Fjob=20TTS=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=E9=A2=86=E5=9F=9F=E6=A8=A1=E5=9E=8B=20+60=E6=B5=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/unit/domain/test_tts_job.py | 502 ++++++++++++++++++++++++++++++
 1 file changed, 502 insertions(+)
 create mode 100755 tests/unit/domain/test_tts_job.py

diff --git a/tests/unit/domain/test_tts_job.py b/tests/unit/domain/test_tts_job.py
new file mode 100755
index 000000000..ff2c1dada
--- /dev/null
+++ b/tests/unit/domain/test_tts_job.py
@@ -0,0 +1,502 @@
+"""TTSJob 领域模型单测.
+
+覆盖：状态枚举、create创建校验、状态机转换、标记方法、
+重试逻辑、属性判断、to_dict序列化。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from packages.domain.tts_job import (
+    TERMINAL_STATUSES,
+    TTSJob,
+    TTSJobStatus,
+)
+
+
+class TestTTSJobStatus:
+    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_is_str_enum(self):
+        assert isinstance(TTSJobStatus.PENDING, str)
+        assert TTSJobStatus.PENDING == "pending"
+
+    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:
+    def test_create_minimal(self):
+        job = TTSJob.create(user_id="user_1", input_text="你好世界")
+        assert job.user_id == "user_1"
+        assert job.input_text == "你好世界"
+        assert job.status == TTSJobStatus.PENDING
+        assert job.id  # 自动生成
+        assert len(job.id) == 32  # uuid4 hex
+
+    def test_create_with_all_params(self):
+        job = TTSJob.create(
+            user_id="user_1",
+            input_text="测试文本",
+            voice_id="voice_clone_123",
+            voice_model="cosyvoice-300m",
+            project_id="proj_456",
+            voice_clone_profile_id="profile_789",
+            sample_rate=16000,
+            format="wav",
+            max_retries=5,
+            metadata={"scene": "video"},
+        )
+        assert job.voice_id == "voice_clone_123"
+        assert job.voice_model == "cosyvoice-300m"
+        assert job.project_id == "proj_456"
+        assert job.voice_clone_profile_id == "profile_789"
+        assert job.sample_rate == 16000
+        assert job.format == "wav"
+        assert job.max_retries == 5
+        assert job.metadata == {"scene": "video"}
+
+    def test_create_defaults(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.sample_rate == 22050
+        assert job.format == "mp3"
+        assert job.max_retries == 3
+        assert job.metadata == {}
+
+    def test_create_strips_whitespace(self):
+        job = TTSJob.create(
+            user_id="  user_1  ",
+            input_text="  你好  ",
+            voice_id="  v1  ",
+        )
+        assert job.user_id == "user_1"
+        assert job.input_text == "你好"
+        assert job.voice_id == "v1"
+
+    def test_create_empty_user_id_raises(self):
+        try:
+            TTSJob.create(user_id="", input_text="hi")
+        except ValueError as e:
+            assert "user_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_whitespace_user_id_raises(self):
+        try:
+            TTSJob.create(user_id="   ", input_text="hi")
+        except ValueError as e:
+            assert "user_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_empty_input_text_raises(self):
+        try:
+            TTSJob.create(user_id="u1", input_text="")
+        except ValueError as e:
+            assert "input_text" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_input_text_too_long_raises(self):
+        long_text = "a" * 10001
+        try:
+            TTSJob.create(user_id="u1", input_text=long_text)
+        except ValueError as e:
+            assert "10000" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_input_text_exactly_10000_ok(self):
+        text = "a" * 10000
+        job = TTSJob.create(user_id="u1", input_text=text)
+        assert job.input_text == text
+
+    def test_create_invalid_format_raises(self):
+        try:
+            TTSJob.create(user_id="u1", input_text="hi", format="ogg")
+        except ValueError as e:
+            assert "不支持的输出格式" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_mp3_format_ok(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", format="mp3")
+        assert job.format == "mp3"
+
+    def test_create_wav_format_ok(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", format="wav")
+        assert job.format == "wav"
+
+    def test_create_pcm_format_ok(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", format="pcm")
+        assert job.format == "pcm"
+
+    def test_create_sets_created_at(self):
+        before = datetime.now(timezone.utc)
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        after = datetime.now(timezone.utc)
+        assert before <= job.created_at <= after
+        assert before <= job.updated_at <= after
+
+    def test_create_metadata_none_defaults_to_empty_dict(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", metadata=None)
+        assert job.metadata == {}
+
+
+class TestStatusProperties:
+    def test_is_terminal_pending(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        assert job.is_terminal is False
+
+    def test_is_terminal_processing(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        assert job.is_terminal is False
+
+    def test_is_terminal_completed(self):
+        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):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_failed("network error")
+        assert job.is_terminal is True
+
+    def test_is_terminal_cancelled(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.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("err")
+        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("err1")
+        job.prepare_retry()
+        job.mark_processing()
+        job.mark_failed("err2")
+        assert job.is_retryable is False
+
+    def test_is_retryable_pending(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        assert job.is_retryable is False
+
+    def test_is_retryable_completed(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed("https://ex.com/a.mp3")
+        assert job.is_retryable is False
+
+    def test_is_completed_success(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed("https://ex.com/a.mp3")
+        assert job.is_completed is True
+
+    def test_is_completed_no_url(self):
+        """completed状态但没有output_url的情况."""
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.status = TTSJobStatus.COMPLETED  # 手动设状态，无url
+        job.output_audio_url = ""
+        assert job.is_completed is False
+
+
+class TestTransitionTo:
+    def test_pending_to_processing(self):
+        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):
+        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):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.CANCELLED)
+        assert job.status == TTSJobStatus.CANCELLED
+
+    def test_processing_to_completed(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        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="hi")
+        job.transition_to(TTSJobStatus.PROCESSING)
+        job.transition_to(TTSJobStatus.FAILED)
+        assert job.status == TTSJobStatus.FAILED
+
+    def test_processing_to_cancelled(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.PROCESSING)
+        job.transition_to(TTSJobStatus.CANCELLED)
+        assert job.status == TTSJobStatus.CANCELLED
+
+    def test_failed_to_pending_retry(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        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="hi")
+        try:
+            job.transition_to(TTSJobStatus.COMPLETED)  # pending→completed 非法
+        except ValueError as e:
+            assert "非法状态转换" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_completed_to_pending_invalid(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.PROCESSING)
+        job.transition_to(TTSJobStatus.COMPLETED)
+        try:
+            job.transition_to(TTSJobStatus.PENDING)
+        except ValueError as e:
+            assert "非法状态转换" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_transition_with_string_status(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to("processing")
+        assert job.status == TTSJobStatus.PROCESSING
+
+    def test_transition_invalid_string_raises(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        try:
+            job.transition_to("invalid_status")
+        except ValueError as e:
+            assert "无效状态" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_transition_updates_updated_at(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        old_updated = job.updated_at
+        import time
+
+        time.sleep(0.01)
+        job.transition_to(TTSJobStatus.PROCESSING)
+        assert job.updated_at > old_updated
+
+    def test_cancelled_no_outgoing_transitions(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.CANCELLED)
+        try:
+            job.transition_to(TTSJobStatus.PENDING)
+        except ValueError:
+            pass
+        else:
+            raise AssertionError("expected ValueError")
+
+
+class TestMarkMethods:
+    def test_mark_processing(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        assert job.status == TTSJobStatus.PROCESSING
+        assert job.started_at is not None
+        assert job.error_message == ""
+
+    def test_mark_processing_clears_error(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.transition_to(TTSJobStatus.FAILED)
+        job.error_message = "old error"
+        job.retry_count = 1
+        # 先回到pending再mark_processing
+        job.transition_to(TTSJobStatus.PENDING)
+        job.mark_processing()
+        assert job.error_message == ""
+
+    def test_mark_completed_success(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed(
+            "https://ex.com/out.mp3",
+            output_audio_key="audio/123.mp3",
+            duration=5.5,
+            file_size=102400,
+        )
+        assert job.status == TTSJobStatus.COMPLETED
+        assert job.output_audio_url == "https://ex.com/out.mp3"
+        assert job.output_audio_key == "audio/123.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="hi")
+        job.mark_processing()
+        try:
+            job.mark_completed("")
+        except ValueError as e:
+            assert "output_audio_url" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_mark_completed_whitespace_url_raises(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        try:
+            job.mark_completed("   ")
+        except ValueError as e:
+            assert "output_audio_url" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_mark_failed(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_failed("connection timeout")
+        assert job.status == TTSJobStatus.FAILED
+        assert job.error_message == "connection timeout"
+
+    def test_mark_cancelled(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_cancelled()
+        assert job.status == TTSJobStatus.CANCELLED
+
+
+class TestRetry:
+    def test_prepare_retry_success(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
+        assert job.error_message == ""
+        assert job.started_at is None
+        assert job.completed_at is None
+
+    def test_prepare_retry_multiple_times(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3)
+        for i in range(3):
+            job.mark_processing()
+            job.mark_failed(f"err_{i}")
+            job.prepare_retry()
+            assert job.retry_count == i + 1
+        assert job.status == TTSJobStatus.PENDING
+        # 第4次应该失败
+        job.mark_processing()
+        job.mark_failed("err_3")
+        try:
+            job.prepare_retry()
+        except ValueError:
+            pass
+        else:
+            raise AssertionError("expected ValueError on 4th retry")
+
+    def test_prepare_retry_not_failed_raises(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        try:
+            job.prepare_retry()  # pending状态不能重试
+        except ValueError as e:
+            assert "不可重试" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_prepare_retry_completed_raises(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed("https://ex.com/a.mp3")
+        try:
+            job.prepare_retry()
+        except ValueError as e:
+            assert "不可重试" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_prepare_retry_resets_timestamps(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed("https://ex.com/a.mp3")
+        # 手动改状态到failed来测试
+        job.status = TTSJobStatus.FAILED
+        job.retry_count = 0
+        job.prepare_retry()
+        assert job.started_at is None
+        assert job.completed_at is None
+
+
+class TestToDict:
+    def test_to_dict_basic(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        d = job.to_dict()
+        assert d["id"] == job.id
+        assert d["user_id"] == "u1"
+        assert d["input_text"] == "hi"
+        assert d["status"] == "pending"
+        assert d["is_retryable"] is False
+        assert d["is_completed"] is False
+
+    def test_to_dict_completed(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        job.mark_processing()
+        job.mark_completed("https://ex.com/a.mp3", duration=3.0, file_size=5000)
+        d = job.to_dict()
+        assert d["status"] == "completed"
+        assert d["output_audio_url"] == "https://ex.com/a.mp3"
+        assert d["duration"] == 3.0
+        assert d["file_size"] == 5000
+        assert d["is_completed"] is True
+        assert d["started_at"] is not None
+        assert d["completed_at"] is not None
+        assert d["created_at"] is not None
+        assert d["updated_at"] is not None
+
+    def test_to_dict_failed_retryable(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3)
+        job.mark_processing()
+        job.mark_failed("timeout")
+        d = job.to_dict()
+        assert d["status"] == "failed"
+        assert d["error_message"] == "timeout"
+        assert d["is_retryable"] is True
+
+    def test_to_dict_datetime_isoformat(self):
+        job = TTSJob.create(user_id="u1", input_text="hi")
+        d = job.to_dict()
+        # ISO格式校验
+        parsed = datetime.fromisoformat(d["created_at"])
+        assert parsed.tzinfo is not None
+
+    def test_to_dict_none_timestamps(self):
+        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_metadata(self):
+        job = TTSJob.create(user_id="u1", input_text="hi", metadata={"key": "value", "num": 42})
+        d = job.to_dict()
+        assert d["metadata"] == {"key": "value", "num": 42}
-- 
2.54.0