test(p3-1): 第27波 voice_clone_workflow 单测 (22个) #733
@@ -1,466 +1,481 @@
|
||||
"""VoiceCloneWorkflowService 单元测试。"""
|
||||
"""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
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.application.cosyvoice_service import (
|
||||
CosyVoiceAuthError,
|
||||
CosyVoiceError,
|
||||
CosyVoiceService,
|
||||
from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError
|
||||
from packages.application.voice_clone.workflow import (
|
||||
VoiceCloneWorkflowError,
|
||||
VoiceCloneWorkflowService,
|
||||
)
|
||||
from packages.application.voice_clone.use_cases import (
|
||||
VoiceCloneNotFoundError,
|
||||
VoiceCloneNotRetryableError,
|
||||
)
|
||||
from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
|
||||
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
|
||||
from packages.shared.url_security import UrlSecurityError
|
||||
|
||||
# ── Helpers ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_profile(
|
||||
*,
|
||||
status: VoiceCloneStatus = VoiceCloneStatus.PENDING,
|
||||
source_audio_url: str = "https://example.com/audio.wav",
|
||||
retry_count: int = 0,
|
||||
max_retries: int = 3,
|
||||
metadata: dict | None = None,
|
||||
) -> VoiceCloneProfile:
|
||||
"""创建测试用 VoiceCloneProfile。"""
|
||||
profile = VoiceCloneProfile.create(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url=source_audio_url,
|
||||
max_retries=max_retries,
|
||||
metadata=metadata,
|
||||
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,
|
||||
)
|
||||
profile.status = status
|
||||
profile.retry_count = retry_count
|
||||
return profile
|
||||
defaults.update(kwargs)
|
||||
return VoiceCloneProfile(**defaults)
|
||||
|
||||
|
||||
def _make_service(
|
||||
*,
|
||||
repo: MagicMock | None = None,
|
||||
cosyvoice: MagicMock | None = None,
|
||||
) -> VoiceCloneWorkflowService:
|
||||
"""创建测试用 VoiceCloneWorkflowService。"""
|
||||
mock_repo = repo or MagicMock()
|
||||
mock_cosyvoice = cosyvoice or MagicMock(spec=CosyVoiceService)
|
||||
return VoiceCloneWorkflowService(repository=mock_repo, cosyvoice_service=mock_cosyvoice)
|
||||
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 []
|
||||
|
||||
|
||||
# ── start_clone ──────────────────────────────────────────
|
||||
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:
|
||||
"""测试 start_clone 方法。"""
|
||||
|
||||
def test_start_clone_with_deploying(self) -> None:
|
||||
"""提交克隆后返回 DEPLOYING 状态，profile 保持 processing。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
# CosyVoice 返回 voice_id + DEPLOYING 状态（需轮询）
|
||||
mock_cosyvoice.submit_clone_task.return_value = {
|
||||
"voice_id": "voice-abc",
|
||||
"status": "DEPLOYING",
|
||||
"request_id": "req-123",
|
||||
}
|
||||
|
||||
# repo.create 和 repo.update 返回传入的 profile
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="https://example.com/audio.wav",
|
||||
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)
|
||||
|
||||
assert profile.status == VoiceCloneStatus.PROCESSING
|
||||
assert profile.metadata["cosyvoice_task_id"] == "voice-abc"
|
||||
assert profile.metadata["cosyvoice_request_id"] == "req-123"
|
||||
mock_cosyvoice.submit_clone_task.assert_called_once()
|
||||
assert mock_repo.create.call_count == 1
|
||||
# update 至少调用 2 次：mark_processing + 保存 voice_id
|
||||
assert mock_repo.update.call_count >= 2
|
||||
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",
|
||||
)
|
||||
|
||||
def test_start_clone_with_ok_status(self) -> None:
|
||||
"""CosyVoice 直接返回 OK 状态，profile 变为 ready。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
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"
|
||||
|
||||
mock_cosyvoice.submit_clone_task.return_value = {
|
||||
"voice_id": "voice-sync-123",
|
||||
"status": "OK",
|
||||
"request_id": "req-456",
|
||||
}
|
||||
# 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"] == "测试音色"
|
||||
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
def test_without_audio_url_returns_pending(self):
|
||||
repo = FakeProfileRepository()
|
||||
cosy = FakeCosyVoiceService()
|
||||
svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="https://example.com/audio.wav",
|
||||
)
|
||||
|
||||
assert profile.status == VoiceCloneStatus.READY
|
||||
assert profile.voice_id == "voice-sync-123"
|
||||
|
||||
def test_start_clone_cosyvoice_error(self) -> None:
|
||||
"""CosyVoice 提交失败，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceError("API 调用失败")
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="https://example.com/audio.wav",
|
||||
)
|
||||
|
||||
assert profile.status == VoiceCloneStatus.FAILED
|
||||
assert "API 调用失败" in profile.error_message
|
||||
|
||||
def test_start_clone_auth_error(self) -> None:
|
||||
"""CosyVoice 认证失败，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceAuthError("认证失败")
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="https://example.com/audio.wav",
|
||||
)
|
||||
|
||||
assert profile.status == VoiceCloneStatus.FAILED
|
||||
assert "认证失败" in profile.error_message
|
||||
|
||||
def test_start_clone_without_audio_url(self) -> None:
|
||||
"""没有音频 URL 时，profile 保持 pending 状态（P2-1 修复后）。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
result = svc.start_clone(
|
||||
user_id="user-1",
|
||||
name="待上传音色",
|
||||
source_audio_url="",
|
||||
)
|
||||
|
||||
# P2-1: 没有音频 URL 时不标记 processing，保持 pending
|
||||
assert profile.status == VoiceCloneStatus.PENDING
|
||||
mock_cosyvoice.submit_clone_task.assert_not_called()
|
||||
assert result.status == VoiceCloneStatus.PENDING.value
|
||||
# No CosyVoice call
|
||||
assert len(cosy.submit_calls) == 0
|
||||
|
||||
def test_start_clone_ssrf_internal_url_rejected(self) -> None:
|
||||
"""SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="http://127.0.0.1/audio.wav",
|
||||
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)
|
||||
|
||||
# 内网 IP 应该被拒绝，标记为 failed
|
||||
assert profile.status == VoiceCloneStatus.FAILED
|
||||
assert "安全校验失败" in profile.error_message
|
||||
mock_cosyvoice.submit_clone_task.assert_not_called()
|
||||
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",
|
||||
)
|
||||
|
||||
def test_start_clone_ssrf_private_ip_rejected(self) -> None:
|
||||
"""SSRF 防护：私有网段 IP 应该被拒绝，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
assert result.status == VoiceCloneStatus.READY.value
|
||||
assert result.voice_id == "voice_ready"
|
||||
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
def test_url_security_failure_marks_failed(self):
|
||||
repo = FakeProfileRepository()
|
||||
cosy = FakeCosyVoiceService()
|
||||
svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="http://192.168.1.100/audio.wav",
|
||||
)
|
||||
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 profile.status == VoiceCloneStatus.FAILED
|
||||
assert "安全校验失败" in profile.error_message
|
||||
mock_cosyvoice.submit_clone_task.assert_not_called()
|
||||
|
||||
def test_start_clone_ssrf_public_url_passes(self) -> None:
|
||||
"""SSRF 防护：正常公网 URL 应该通过校验。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
mock_cosyvoice.submit_clone_task.return_value = {
|
||||
"voice_id": "voice-ssrf-test",
|
||||
"status": "DEPLOYING",
|
||||
"request_id": "req-ssrf",
|
||||
}
|
||||
mock_repo.create.side_effect = lambda p: p
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
profile = service.start_clone(
|
||||
user_id="user-123",
|
||||
name="测试音色",
|
||||
source_audio_url="https://example.com/audio.wav",
|
||||
)
|
||||
|
||||
# 公网 URL 应该正常通过
|
||||
assert profile.status == VoiceCloneStatus.PROCESSING
|
||||
mock_cosyvoice.submit_clone_task.assert_called_once()
|
||||
|
||||
|
||||
# ── process_clone_result ─────────────────────────────────
|
||||
|
||||
|
||||
class TestProcessCloneResult:
|
||||
"""测试 process_clone_result 方法。"""
|
||||
|
||||
def test_process_clone_result_success(self) -> None:
|
||||
"""克隆成功，profile 标记为 ready。"""
|
||||
mock_repo = MagicMock()
|
||||
profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
result = service.process_clone_result(profile.id, "voice-xyz")
|
||||
|
||||
assert result.status == VoiceCloneStatus.READY
|
||||
assert result.voice_id == "voice-xyz"
|
||||
|
||||
def test_process_clone_result_not_found(self) -> None:
|
||||
"""profile 不存在时抛出异常。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
with pytest.raises(VoiceCloneNotFoundError):
|
||||
service.process_clone_result("nonexistent", "voice-xyz")
|
||||
|
||||
|
||||
# ── process_clone_failure ────────────────────────────────
|
||||
|
||||
|
||||
class TestProcessCloneFailure:
|
||||
"""测试 process_clone_failure 方法。"""
|
||||
|
||||
def test_process_clone_failure(self) -> None:
|
||||
"""克隆失败，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
result = service.process_clone_failure(profile.id, "超时错误")
|
||||
|
||||
assert result.status == VoiceCloneStatus.FAILED
|
||||
assert result.error_message == "超时错误"
|
||||
|
||||
def test_process_clone_failure_not_found(self) -> None:
|
||||
"""profile 不存在时抛出异常。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
with pytest.raises(VoiceCloneNotFoundError):
|
||||
service.process_clone_failure("nonexistent", "错误")
|
||||
|
||||
|
||||
# ── retry_clone ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestRetryClone:
|
||||
"""测试 retry_clone 方法。"""
|
||||
|
||||
def test_retry_clone_with_async_task(self) -> None:
|
||||
"""重试成功，异步模式。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
profile = _make_profile(status=VoiceCloneStatus.FAILED, retry_count=1, max_retries=3)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
mock_cosyvoice.submit_clone_task.return_value = {
|
||||
"voice_id": "voice-retry",
|
||||
"status": "DEPLOYING",
|
||||
"request_id": "req-retry",
|
||||
}
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
result = service.retry_clone(profile.id, "user-123")
|
||||
|
||||
assert result.status == VoiceCloneStatus.PROCESSING
|
||||
assert result.metadata["cosyvoice_task_id"] == "voice-retry"
|
||||
assert result.retry_count == 2 # prepare_retry 增加了一次
|
||||
|
||||
def test_retry_clone_with_ok_status(self) -> None:
|
||||
"""重试成功，直接返回 OK 状态。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
profile = _make_profile(status=VoiceCloneStatus.FAILED, retry_count=0, max_retries=3)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
mock_cosyvoice.submit_clone_task.return_value = {
|
||||
"voice_id": "voice-retry-sync",
|
||||
"status": "OK",
|
||||
"request_id": "req-retry",
|
||||
}
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
result = service.retry_clone(profile.id, "user-123")
|
||||
|
||||
assert result.status == VoiceCloneStatus.READY
|
||||
assert result.voice_id == "voice-retry-sync"
|
||||
|
||||
def test_retry_clone_not_found(self) -> None:
|
||||
"""profile 不存在时抛出异常。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
with pytest.raises(VoiceCloneNotFoundError):
|
||||
service.retry_clone("nonexistent", "user-123")
|
||||
|
||||
def test_retry_clone_not_retryable(self) -> None:
|
||||
"""不可重试时抛出异常。"""
|
||||
mock_repo = MagicMock()
|
||||
profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
|
||||
mock_repo.get.return_value = profile
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
with pytest.raises(VoiceCloneNotRetryableError):
|
||||
service.retry_clone(profile.id, "user-123")
|
||||
|
||||
def test_retry_clone_cosyvoice_error(self) -> None:
|
||||
"""重试时 CosyVoice 失败，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
profile = _make_profile(status=VoiceCloneStatus.FAILED, retry_count=0, max_retries=3)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceError("重试失败")
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
result = service.retry_clone(profile.id, "user-123")
|
||||
|
||||
assert result.status == VoiceCloneStatus.FAILED
|
||||
assert "重试失败" in result.error_message
|
||||
|
||||
def test_retry_clone_ssrf_internal_url_rejected(self) -> None:
|
||||
"""重试时 SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
profile = _make_profile(
|
||||
status=VoiceCloneStatus.FAILED,
|
||||
source_audio_url="http://10.0.0.1/secret.wav",
|
||||
retry_count=0,
|
||||
max_retries=3,
|
||||
)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
result = service.retry_clone(profile.id, "user-123")
|
||||
|
||||
assert result.status == VoiceCloneStatus.FAILED
|
||||
assert result.status == VoiceCloneStatus.FAILED.value
|
||||
assert "安全校验失败" in result.error_message
|
||||
mock_cosyvoice.submit_clone_task.assert_not_called()
|
||||
# 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 ───────────────────────────────
|
||||
# ── poll_and_process_clone tests ────────────────────────
|
||||
|
||||
|
||||
class TestPollAndProcessClone:
|
||||
"""测试 poll_and_process_clone 方法（P2-2 修复）。"""
|
||||
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)
|
||||
|
||||
def test_poll_and_process_clone_success(self) -> None:
|
||||
"""轮询成功：调用 poll_clone_task → process_clone_result。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
result = svc.poll_and_process_clone("profile-123", timeout=60.0)
|
||||
|
||||
profile = _make_profile(
|
||||
status=VoiceCloneStatus.PROCESSING,
|
||||
metadata={"cosyvoice_task_id": "task-abc"},
|
||||
)
|
||||
mock_repo.get.return_value = profile
|
||||
mock_repo.update.side_effect = lambda p: p
|
||||
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
|
||||
|
||||
mock_cosyvoice.poll_clone_task.return_value = {"voice_id": "voice-poll-xyz"}
|
||||
def test_profile_not_found_raises(self):
|
||||
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
result = service.poll_and_process_clone(profile.id)
|
||||
repo = FakeProfileRepository()
|
||||
cosy = FakeCosyVoiceService()
|
||||
svc = VoiceCloneWorkflowService(repository=repo, cosyvoice_service=cosy)
|
||||
|
||||
assert result.status == VoiceCloneStatus.READY
|
||||
assert result.voice_id == "voice-poll-xyz"
|
||||
mock_cosyvoice.poll_clone_task.assert_called_once_with("task-abc", timeout=300)
|
||||
|
||||
def test_poll_and_process_clone_no_task_id(self) -> None:
|
||||
"""metadata 中没有 task_id 时抛出 ValueError。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
|
||||
profile = _make_profile(status=VoiceCloneStatus.PROCESSING, metadata={})
|
||||
mock_repo.get.return_value = profile
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
with pytest.raises(ValueError, match="cosyvoice_task_id"):
|
||||
service.poll_and_process_clone(profile.id)
|
||||
|
||||
def test_poll_and_process_clone_not_found(self) -> None:
|
||||
"""profile 不存在时抛出 VoiceCloneNotFoundError。"""
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
service = _make_service(repo=mock_repo)
|
||||
with pytest.raises(VoiceCloneNotFoundError):
|
||||
service.poll_and_process_clone("nonexistent")
|
||||
svc.poll_and_process_clone("nonexistent")
|
||||
|
||||
def test_poll_and_process_clone_timeout(self) -> None:
|
||||
"""超时时透传 CosyVoiceTimeoutError（由 Celery task 捕获重试）。"""
|
||||
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
|
||||
|
||||
mock_repo = MagicMock()
|
||||
mock_cosyvoice = MagicMock(spec=CosyVoiceService)
|
||||
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)
|
||||
|
||||
profile = _make_profile(
|
||||
status=VoiceCloneStatus.PROCESSING,
|
||||
metadata={"cosyvoice_task_id": "task-abc"},
|
||||
)
|
||||
mock_repo.get.return_value = profile
|
||||
|
||||
mock_cosyvoice.poll_clone_task.side_effect = CosyVoiceTimeoutError("超时")
|
||||
|
||||
service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
|
||||
with pytest.raises(CosyVoiceTimeoutError):
|
||||
service.poll_and_process_clone(profile.id)
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user