xiaoxia-saas/tests/unit/test_voice_clone_workflow.py

"""Voice clone workflow service unit tests.

Covers VoiceCloneWorkflowService - start_clone, poll_and_process_clone,
process_clone_result, process_clone_failure, retry_clone.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError
from packages.application.voice_clone.workflow import (
    VoiceCloneWorkflowError,
    VoiceCloneWorkflowService,
)
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
from packages.shared.url_security import UrlSecurityError

# ── Helpers ─────────────────────────────────────────────


def make_profile(**kwargs):
    defaults = dict(
        id="profile-123",
        user_id="user-1",
        name="我的音色",
        description="",
        source_audio_url="",
        voice_model="",
        language="zh-CN",
        gender="unknown",
        max_retries=3,
    )
    defaults.update(kwargs)
    return VoiceCloneProfile(**defaults)


class FakeProfileRepository:
    def __init__(self, profile=None):
        self._profile = profile
        self.updated = []
        self.created = []
        self.get_called = 0

    def create(self, profile):
        self.created.append(profile)
        self._profile = profile
        return profile

    def get(self, profile_id):
        self.get_called += 1
        if self._profile and self._profile.id == profile_id:
            return self._profile
        return None

    def update(self, profile):
        self.updated.append(profile)
        self._profile = profile
        return profile

    def find_by_user(self, user_id):
        if self._profile and self._profile.user_id == user_id:
            return [self._profile]
        return []


class FakeCosyVoiceService:
    def __init__(self, submit_result=None, submit_error=None, poll_result=None, poll_error=None):
        self._submit_result = submit_result or {
            "voice_id": "voice_abc123",
            "status": "DEPLOYING",
            "request_id": "req-abc",
        }
        self._submit_error = submit_error
        self._poll_result = poll_result or {"voice_id": "voice_abc123"}
        self._poll_error = poll_error
        self.submit_calls = []
        self.poll_calls = []

    def submit_clone_task(self, **kwargs):
        self.submit_calls.append(kwargs)
        if self._submit_error:
            raise self._submit_error
        return self._submit_result

    def poll_clone_task(self, voice_id, timeout=300.0):
        self.poll_calls.append({"voice_id": voice_id, "timeout": timeout})
        if self._poll_error:
            raise self._poll_error
        return self._poll_result


# ── start_clone tests ───────────────────────────────────


class TestStartClone:
    def test_with_audio_url_submits_and_returns_processing(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService(
            submit_result={
                "voice_id": "voice_new123",
                "status": "DEPLOYING",
                "request_id": "req-xyz",
            }
        )
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/audio.mp3",
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="测试音色",
                source_audio_url="https://example.com/audio.mp3",
            )

        assert result.user_id == "user-1"
        assert result.name == "测试音色"
        assert result.status == VoiceCloneStatus.PROCESSING.value
        assert result.metadata["cosyvoice_task_id"] == "voice_new123"
        assert result.metadata["cosyvoice_request_id"] == "req-xyz"

        # CosyVoice was called
        assert len(cosy.submit_calls) == 1
        assert cosy.submit_calls[0]["audio_url"] == "https://safe.example.com/audio.mp3"
        assert cosy.submit_calls[0]["voice_name"] == "测试音色"

    def test_without_audio_url_returns_pending(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.start_clone(
            user_id="user-1",
            name="待上传音色",
            source_audio_url="",
        )

        assert result.status == VoiceCloneStatus.PENDING.value
        # No CosyVoice call
        assert len(cosy.submit_calls) == 0

    def test_sync_completion_status_ok(self):
        """CosyVoice returns OK status directly → mark ready."""
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService(
            submit_result={
                "voice_id": "voice_ready",
                "status": "OK",
                "request_id": "req-ok",
            }
        )
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="秒成音色",
                source_audio_url="https://example.com/a.mp3",
            )

        assert result.status == VoiceCloneStatus.READY.value
        assert result.voice_id == "voice_ready"

    def test_url_security_failure_marks_failed(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            side_effect=UrlSecurityError("SSRF detected: internal IP"),
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="危险音色",
                source_audio_url="https://10.0.0.1/internal.mp3",
            )

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "安全校验失败" in result.error_message
        # No CosyVoice call
        assert len(cosy.submit_calls) == 0

    def test_cosyvoice_error_marks_failed(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("API rate limit exceeded"))
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="失败音色",
                source_audio_url="https://example.com/a.mp3",
            )

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "API rate limit" in result.error_message

    def test_cosyvoice_auth_error_marks_failed(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceAuthError("Invalid API key"))
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="鉴权失败",
                source_audio_url="https://example.com/a.mp3",
            )

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "Invalid API key" in result.error_message

    def test_value_error_marks_failed(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService(submit_error=ValueError("audio_url is empty"))
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="参数错误",
                source_audio_url="https://example.com/a.mp3",
            )

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "audio_url is empty" in result.error_message

    def test_custom_language_and_model(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety", return_value="https://s.example.com/a.mp3"
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="English Voice",
                source_audio_url="https://example.com/a.mp3",
                language="en-US",
                voice_model="cosyvoice-v3",
            )

        assert cosy.submit_calls[0]["language"] == "en-US"
        assert result.language == "en-US"
        assert result.voice_model == "cosyvoice-v3"

    def test_metadata_passed_through(self):
        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety", return_value="https://s.example.com/a.mp3"
        ):
            result = svc.start_clone(
                user_id="user-1",
                name="带元数据",
                source_audio_url="https://example.com/a.mp3",
                metadata={"source": "upload", "format": "wav"},
            )

        assert result.metadata.get("source") == "upload"
        assert result.metadata.get("format") == "wav"
        # cosyvoice keys also present
        assert "cosyvoice_task_id" in result.metadata


# ── poll_and_process_clone tests ────────────────────────


class TestPollAndProcessClone:
    def test_poll_success_returns_ready(self):
        profile = make_profile()
        profile.mark_processing()
        profile.metadata = {"cosyvoice_task_id": "voice_pending"}
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService(poll_result={"voice_id": "voice_done"})
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.poll_and_process_clone("profile-123", timeout=60.0)

        assert result.status == VoiceCloneStatus.READY.value
        assert result.voice_id == "voice_done"
        assert len(cosy.poll_calls) == 1
        assert cosy.poll_calls[0]["voice_id"] == "voice_pending"
        assert cosy.poll_calls[0]["timeout"] == 60.0

    def test_profile_not_found_raises(self):
        from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

        repo = FakeProfileRepository()
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(VoiceCloneNotFoundError):
            svc.poll_and_process_clone("nonexistent")

    def test_no_task_id_raises_value_error(self):
        profile = make_profile()
        profile.mark_processing()
        profile.metadata = {}  # no task_id
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(ValueError, match="no cosyvoice_task_id"):
            svc.poll_and_process_clone("profile-123")

    def test_poll_timeout_propagates(self):
        from packages.application.cosyvoice_service import CosyVoiceTimeoutError

        profile = make_profile()
        profile.mark_processing()
        profile.metadata = {"cosyvoice_task_id": "voice_slow"}
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService(poll_error=CosyVoiceTimeoutError("timed out"))
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(CosyVoiceTimeoutError):
            svc.poll_and_process_clone("profile-123")


# ── process_clone_result tests ──────────────────────────


class TestProcessCloneResult:
    def test_marks_profile_ready(self):
        profile = make_profile()
        profile.mark_processing()
        repo = FakeProfileRepository(profile=profile)
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        result = svc.process_clone_result("profile-123", "voice_final123")

        assert result.status == VoiceCloneStatus.READY.value
        assert result.voice_id == "voice_final123"

    def test_profile_not_found_raises(self):
        from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

        repo = FakeProfileRepository()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        with pytest.raises(VoiceCloneNotFoundError):
            svc.process_clone_result("nonexistent", "voice_x")


# ── process_clone_failure tests ─────────────────────────


class TestProcessCloneFailure:
    def test_marks_profile_failed(self):
        profile = make_profile()
        profile.mark_processing()
        repo = FakeProfileRepository(profile=profile)
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        result = svc.process_clone_failure("profile-123", "审核未通过")

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "审核未通过" in result.error_message

    def test_profile_not_found_raises(self):
        from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

        repo = FakeProfileRepository()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        with pytest.raises(VoiceCloneNotFoundError):
            svc.process_clone_failure("nonexistent", "error")


# ── retry_clone tests ───────────────────────────────────


class TestRetryClone:
    def test_retry_success_submits_again(self):
        profile = make_profile(source_audio_url="https://example.com/orig.mp3")
        profile.mark_failed("previous error")
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService(
            submit_result={
                "voice_id": "voice_retry",
                "status": "DEPLOYING",
                "request_id": "req-retry",
            }
        )
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/orig.mp3",
        ):
            result = svc.retry_clone("profile-123", "user-1")

        # Should be processing again
        assert result.status == VoiceCloneStatus.PROCESSING.value
        assert result.metadata["cosyvoice_task_id"] == "voice_retry"
        assert len(cosy.submit_calls) == 1

    def test_retry_url_security_failure(self):
        profile = make_profile(source_audio_url="https://10.0.0.1/audio.mp3")
        profile.mark_failed("old error")
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService()
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety", side_effect=UrlSecurityError("internal IP")
        ):
            result = svc.retry_clone("profile-123", "user-1")

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "安全校验失败" in result.error_message
        assert len(cosy.submit_calls) == 0

    def test_retry_cosyvoice_error(self):
        profile = make_profile(source_audio_url="https://example.com/a.mp3")
        profile.mark_failed("old error")
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("still failing"))
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.retry_clone("profile-123", "user-1")

        assert result.status == VoiceCloneStatus.FAILED.value
        assert "still failing" in result.error_message

    def test_retry_sync_ok(self):
        profile = make_profile(source_audio_url="https://example.com/a.mp3")
        profile.mark_failed("old error")
        repo = FakeProfileRepository(profile=profile)
        cosy = FakeCosyVoiceService(
            submit_result={
                "voice_id": "voice_instant",
                "status": "OK",
                "request_id": "req-instant",
            }
        )
        svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)

        with patch(
            "packages.application.voice_clone.workflow.validate_url_safety",
            return_value="https://safe.example.com/a.mp3",
        ):
            result = svc.retry_clone("profile-123", "user-1")

        assert result.status == VoiceCloneStatus.READY.value
        assert result.voice_id == "voice_instant"


# ── Error class tests ───────────────────────────────────


class TestErrorClasses:
    def test_workflow_error_inherits_exception(self):
        err = VoiceCloneWorkflowError("test error")
        assert isinstance(err, Exception)
        assert str(err) == "test error"