tests/unit/domain/test_voice_clone_profile.py

@@ -0,0 +1,501 @@
+"""VoiceCloneProfile 音色克隆档案单测.
+
+覆盖：状态枚举、create创建校验、状态机转换、标记方法、
+重试逻辑、属性判断、to_dict序列化。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from packages.domain.voice_clone_profile import (
+    TERMINAL_STATUSES,
+    VoiceCloneProfile,
+    VoiceCloneStatus,
+)
+
+
+class TestVoiceCloneStatus:
+    def test_status_values(self):
+        assert VoiceCloneStatus.PENDING.value == "pending"
+        assert VoiceCloneStatus.PROCESSING.value == "processing"
+        assert VoiceCloneStatus.READY.value == "ready"
+        assert VoiceCloneStatus.FAILED.value == "failed"
+        assert VoiceCloneStatus.DISABLED.value == "disabled"
+
+    def test_status_is_str_enum(self):
+        assert isinstance(VoiceCloneStatus.PENDING, str)
+        assert VoiceCloneStatus.PENDING == "pending"
+
+    def test_terminal_statuses(self):
+        assert VoiceCloneStatus.READY in TERMINAL_STATUSES
+        assert VoiceCloneStatus.FAILED in TERMINAL_STATUSES
+        assert VoiceCloneStatus.DISABLED in TERMINAL_STATUSES
+        assert VoiceCloneStatus.PENDING not in TERMINAL_STATUSES
+        assert VoiceCloneStatus.PROCESSING not in TERMINAL_STATUSES
+
+
+class TestVoiceCloneProfileCreate:
+    def test_create_minimal(self):
+        profile = VoiceCloneProfile.create(user_id="user_1", name="我的音色")
+        assert profile.user_id == "user_1"
+        assert profile.name == "我的音色"
+        assert profile.status == VoiceCloneStatus.PENDING
+        assert profile.id and len(profile.id) == 32
+
+    def test_create_with_all_params(self):
+        profile = VoiceCloneProfile.create(
+            user_id="user_1",
+            name="甜美女声",
+            description="适合播客的女声",
+            source_audio_url="https://ex.com/source.wav",
+            voice_model="cosyvoice-300m",
+            language="en-US",
+            gender="female",
+            max_retries=5,
+            metadata={"source": "upload"},
+        )
+        assert profile.description == "适合播客的女声"
+        assert profile.source_audio_url == "https://ex.com/source.wav"
+        assert profile.voice_model == "cosyvoice-300m"
+        assert profile.language == "en-US"
+        assert profile.gender == "female"
+        assert profile.max_retries == 5
+        assert profile.metadata == {"source": "upload"}
+
+    def test_create_defaults(self):
+        profile = VoiceCloneProfile.create(user_id="u1", name="t")
+        assert profile.description == ""
+        assert profile.source_audio_url == ""
+        assert profile.voice_model == ""
+        assert profile.language == "zh-CN"
+        assert profile.gender == "unknown"
+        assert profile.max_retries == 3
+        assert profile.metadata == {}
+        assert profile.voice_id == ""
+        assert profile.error_message == ""
+        assert profile.retry_count == 0
+
+    def test_create_strips_whitespace(self):
+        profile = VoiceCloneProfile.create(
+            user_id="  user_1  ",
+            name="  测试音色  ",
+            description="  desc  ",
+            language="  en-US  ",
+            gender="  MALE  ",
+        )
+        assert profile.user_id == "user_1"
+        assert profile.name == "测试音色"
+        assert profile.description == "desc"
+        assert profile.language == "en-US"
+        assert profile.gender == "male"  # lowercased
+
+    def test_create_gender_lowercased(self):
+        profile = VoiceCloneProfile.create(user_id="u1", name="t", gender="Female")
+        assert profile.gender == "female"
+
+    def test_create_empty_user_id_raises(self):
+        try:
+            VoiceCloneProfile.create(user_id="", name="t")
+        except ValueError as e:
+            assert "user_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_whitespace_user_id_raises(self):
+        try:
+            VoiceCloneProfile.create(user_id="   ", name="t")
+        except ValueError as e:
+            assert "user_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_empty_name_raises(self):
+        try:
+            VoiceCloneProfile.create(user_id="u1", name="")
+        except ValueError as e:
+            assert "name" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_name_too_long_raises(self):
+        long_name = "a" * 101
+        try:
+            VoiceCloneProfile.create(user_id="u1", name=long_name)
+        except ValueError as e:
+            assert "100" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_create_name_exactly_100_ok(self):
+        name = "a" * 100
+        profile = VoiceCloneProfile.create(user_id="u1", name=name)
+        assert profile.name == name
+
+    def test_create_sets_created_at(self):
+        before = datetime.now(timezone.utc)
+        profile = VoiceCloneProfile.create(user_id="u1", name="t")
+        after = datetime.now(timezone.utc)
+        assert before <= profile.created_at <= after
+        assert before <= profile.updated_at <= after
+
+    def test_create_metadata_none_defaults_to_empty(self):
+        profile = VoiceCloneProfile.create(user_id="u1", name="t", metadata=None)
+        assert profile.metadata == {}
+
+
+class TestStatusProperties:
+    def test_is_terminal_pending(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        assert p.is_terminal is False
+
+    def test_is_terminal_processing(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        assert p.is_terminal is False
+
+    def test_is_terminal_ready(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("voice_123")
+        assert p.is_terminal is True
+
+    def test_is_terminal_failed(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_failed("error")
+        assert p.is_terminal is True
+
+    def test_is_terminal_disabled_from_pending(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_disabled()
+        assert p.is_terminal is True
+
+    def test_is_retryable_failed_within_limit(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=3)
+        p.mark_processing()
+        p.mark_failed("err")
+        assert p.is_retryable is True
+
+    def test_is_retryable_failed_at_limit(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=1)
+        p.mark_processing()
+        p.mark_failed("err1")
+        p.prepare_retry()
+        p.mark_processing()
+        p.mark_failed("err2")
+        assert p.is_retryable is False
+
+    def test_is_retryable_pending(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        assert p.is_retryable is False
+
+    def test_is_retryable_ready(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("v1")
+        assert p.is_retryable is False
+
+    def test_is_ready_success(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("voice_123")
+        assert p.is_ready is True
+
+    def test_is_ready_no_voice_id(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.status = VoiceCloneStatus.READY  # 手动设状态，无voice_id
+        p.voice_id = ""
+        assert p.is_ready is False
+
+
+class TestTransitionTo:
+    def test_pending_to_processing(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        assert p.status == VoiceCloneStatus.PROCESSING
+
+    def test_pending_to_failed(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.FAILED)
+        assert p.status == VoiceCloneStatus.FAILED
+
+    def test_pending_to_disabled(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.DISABLED)
+        assert p.status == VoiceCloneStatus.DISABLED
+
+    def test_processing_to_ready(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        p.transition_to(VoiceCloneStatus.READY)
+        assert p.status == VoiceCloneStatus.READY
+
+    def test_processing_to_failed(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        p.transition_to(VoiceCloneStatus.FAILED)
+        assert p.status == VoiceCloneStatus.FAILED
+
+    def test_processing_to_disabled(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        p.transition_to(VoiceCloneStatus.DISABLED)
+        assert p.status == VoiceCloneStatus.DISABLED
+
+    def test_failed_to_pending_retry(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        p.transition_to(VoiceCloneStatus.FAILED)
+        p.transition_to(VoiceCloneStatus.PENDING)
+        assert p.status == VoiceCloneStatus.PENDING
+
+    def test_ready_to_disabled(self):
+        """已就绪音色可以被禁用."""
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        p.transition_to(VoiceCloneStatus.READY)
+        p.transition_to(VoiceCloneStatus.DISABLED)
+        assert p.status == VoiceCloneStatus.DISABLED
+
+    def test_invalid_transition_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        try:
+            p.transition_to(VoiceCloneStatus.READY)  # pending→ready 非法
+        except ValueError as e:
+            assert "非法状态转换" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_disabled_no_outgoing(self):
+        """disabled状态不能转换到任何状态."""
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.DISABLED)
+        try:
+            p.transition_to(VoiceCloneStatus.PENDING)
+        except ValueError:
+            pass
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_transition_with_string(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to("processing")
+        assert p.status == VoiceCloneStatus.PROCESSING
+
+    def test_transition_invalid_string_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        try:
+            p.transition_to("invalid_state")
+        except ValueError as e:
+            assert "无效状态" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_transition_updates_updated_at(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        old_updated = p.updated_at
+        import time
+
+        time.sleep(0.01)
+        p.transition_to(VoiceCloneStatus.PROCESSING)
+        assert p.updated_at > old_updated
+
+
+class TestMarkMethods:
+    def test_mark_processing(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        assert p.status == VoiceCloneStatus.PROCESSING
+        assert p.error_message == ""
+
+    def test_mark_processing_clears_error(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.transition_to(VoiceCloneStatus.FAILED)
+        p.error_message = "old error"
+        p.retry_count = 1
+        p.transition_to(VoiceCloneStatus.PENDING)
+        p.mark_processing()
+        assert p.error_message == ""
+
+    def test_mark_ready_success(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("voice_id_123")
+        assert p.status == VoiceCloneStatus.READY
+        assert p.voice_id == "voice_id_123"
+        assert p.error_message == ""
+
+    def test_mark_ready_strips_whitespace(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("  voice_456  ")
+        assert p.voice_id == "voice_456"
+
+    def test_mark_ready_empty_voice_id_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        try:
+            p.mark_ready("")
+        except ValueError as e:
+            assert "voice_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_mark_ready_whitespace_voice_id_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        try:
+            p.mark_ready("   ")
+        except ValueError as e:
+            assert "voice_id" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_mark_failed(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_failed("训练超时")
+        assert p.status == VoiceCloneStatus.FAILED
+        assert p.error_message == "训练超时"
+
+    def test_mark_disabled(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_disabled()
+        assert p.status == VoiceCloneStatus.DISABLED
+
+
+class TestRetry:
+    def test_prepare_retry_success(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=3)
+        p.mark_processing()
+        p.mark_failed("err")
+        p.prepare_retry()
+        assert p.status == VoiceCloneStatus.PENDING
+        assert p.retry_count == 1
+        assert p.error_message == ""
+        assert p.voice_id == ""
+
+    def test_prepare_retry_multiple_times(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t", max_retries=3)
+        for i in range(3):
+            p.mark_processing()
+            p.mark_failed(f"err_{i}")
+            p.prepare_retry()
+            assert p.retry_count == i + 1
+        assert p.status == VoiceCloneStatus.PENDING
+        # 第4次应该失败
+        p.mark_processing()
+        p.mark_failed("err_3")
+        try:
+            p.prepare_retry()
+        except ValueError:
+            pass
+        else:
+            raise AssertionError("expected ValueError on 4th retry")
+
+    def test_prepare_retry_not_failed_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        try:
+            p.prepare_retry()
+        except ValueError as e:
+            assert "不可重试" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_prepare_retry_ready_raises(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("v1")
+        try:
+            p.prepare_retry()
+        except ValueError as e:
+            assert "不可重试" in str(e)
+        else:
+            raise AssertionError("expected ValueError")
+
+    def test_prepare_retry_clears_voice_id(self):
+        p = VoiceCloneProfile.create(user_id="u", name="t")
+        p.mark_processing()
+        p.mark_ready("partial_id")
+        # 手动改到failed来测试
+        p.status = VoiceCloneStatus.FAILED
+        p.retry_count = 0
+        p.prepare_retry()
+        assert p.voice_id == ""
+
+
+class TestToDict:
+    def test_to_dict_basic(self):
+        p = VoiceCloneProfile.create(user_id="u1", name="测试音色")
+        d = p.to_dict()
+        assert d["id"] == p.id
+        assert d["user_id"] == "u1"
+        assert d["name"] == "测试音色"
+        assert d["status"] == "pending"
+        assert d["is_retryable"] is False
+        assert d["is_ready"] is False
+
+    def test_to_dict_ready(self):
+        p = VoiceCloneProfile.create(user_id="u1", name="t")
+        p.mark_processing()
+        p.mark_ready("voice_42")
+        d = p.to_dict()
+        assert d["status"] == "ready"
+        assert d["voice_id"] == "voice_42"
+        assert d["is_ready"] is True
+        assert d["created_at"] is not None
+        assert d["updated_at"] is not None
+
+    def test_to_dict_failed_retryable(self):
+        p = VoiceCloneProfile.create(user_id="u1", name="t", max_retries=3)
+        p.mark_processing()
+        p.mark_failed("timeout")
+        d = p.to_dict()
+        assert d["status"] == "failed"
+        assert d["error_message"] == "timeout"
+        assert d["is_retryable"] is True
+
+    def test_to_dict_datetime_isoformat(self):
+        p = VoiceCloneProfile.create(user_id="u1", name="t")
+        d = p.to_dict()
+        parsed = datetime.fromisoformat(d["created_at"])
+        assert parsed.tzinfo is not None
+
+    def test_to_dict_metadata(self):
+        p = VoiceCloneProfile.create(user_id="u1", name="t", metadata={"key": "val", "num": 42})
+        d = p.to_dict()
+        assert d["metadata"] == {"key": "val", "num": 42}
+
+    def test_to_dict_all_fields_present(self):
+        p = VoiceCloneProfile.create(
+            user_id="u1",
+            name="t",
+            description="d",
+            source_audio_url="https://ex.com/s.wav",
+            voice_model="cosyvoice",
+            language="zh-CN",
+            gender="male",
+        )
+        d = p.to_dict()
+        expected_keys = {
+            "id",
+            "user_id",
+            "name",
+            "description",
+            "status",
+            "source_audio_url",
+            "voice_id",
+            "voice_model",
+            "language",
+            "gender",
+            "error_message",
+            "retry_count",
+            "max_retries",
+            "is_retryable",
+            "is_ready",
+            "metadata",
+            "created_at",
+            "updated_at",
+        }
+        assert set(d.keys()) == expected_keys