diff --git a/tests/unit/test_cosyvoice_service.py b/tests/unit/test_cosyvoice_service.py index 9cc9a43fb..5a649aef6 100755 --- a/tests/unit/test_cosyvoice_service.py +++ b/tests/unit/test_cosyvoice_service.py @@ -1,9 +1,9 @@ -"""CosyVoiceService 单元测试 — 适配百炼 DashScope API.""" +"""CosyVoiceService 单元测试.""" from __future__ import annotations -import json -from unittest.mock import MagicMock +import time +from unittest.mock import MagicMock, patch import httpx import pytest @@ -16,415 +16,642 @@ from packages.application.cosyvoice_service import ( CosyVoiceTimeoutError, SynthesizeResult, ) -from packages.domain.preset_voices import PresetVoice -def _make_service( - *, - api_key: str = "test-api-key", - base_url: str = "https://dashscope.aliyuncs.com/api/v1", - model: str = "cosyvoice-v3-flash", - clone_model: str = "voice-enrollment", - http_client: httpx.Client | None = None, - audio_url_signer=None, -) -> CosyVoiceService: - return CosyVoiceService( - api_key=api_key, - base_url=base_url, - model=model, - clone_model=clone_model, - http_client=http_client, - audio_url_signer=audio_url_signer, - ) +@pytest.fixture +def mock_client() -> MagicMock: + """Mock HTTP client.""" + return MagicMock(spec=httpx.Client) -def _mock_response( - status_code: int = 200, - json_data: dict | None = None, - text: str = "", - method: str = "POST", -) -> httpx.Response: - content = b"" - if json_data is not None: - content = json.dumps(json_data).encode("utf-8") - elif text: - content = text.encode("utf-8") - - return httpx.Response( - status_code=status_code, - content=content, - request=httpx.Request(method, "https://dashscope.aliyuncs.com/api/v1/test"), - ) +@pytest.fixture +def service(mock_client: MagicMock) -> CosyVoiceService: + """Create CosyVoiceService with mocked HTTP client and config.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + settings.cosyvoice_api_key = "sk-test-12345678" + settings.cosyvoice_base_url = "https://dashscope.aliyuncs.com/api/v1" + settings.cosyvoice_model = "cosyvoice-v3-flash" + settings.cosyvoice_sample_rate = 22050 + settings.cosyvoice_format = "mp3" + settings.cosyvoice_voice = "longxiaochun_v3" + settings.cosyvoice_clone_model = "voice-enrollment" + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + svc.CLONE_POLL_INTERVAL = 0.001 # 加速测试 + svc.RETRY_BACKOFF = 0.001 + yield svc -# ── list_preset_voices ─────────────────────────────────── +class TestInitConfig: + """初始化配置测试.""" + + def test_base_url_with_old_text2audio_path_gets_normalized(self, mock_client: MagicMock) -> None: + """旧版 base_url 带 text2audio 路径应自动修正.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + settings.cosyvoice_api_key = "sk-test" + settings.cosyvoice_base_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio" + settings.cosyvoice_model = "cosyvoice-v3-flash" + settings.cosyvoice_sample_rate = 22050 + settings.cosyvoice_format = "mp3" + settings.cosyvoice_voice = "test" + settings.cosyvoice_clone_model = "voice-enrollment" + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + assert svc._base_url == "https://dashscope.aliyuncs.com/api/v1" + + def test_custom_params_override_config(self, mock_client: MagicMock) -> None: + """显式传入参数覆盖配置.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + settings.cosyvoice_api_key = "sk-config" + settings.cosyvoice_base_url = "https://config.example.com" + settings.cosyvoice_model = "config-model" + settings.cosyvoice_sample_rate = 22050 + settings.cosyvoice_format = "mp3" + settings.cosyvoice_voice = "test" + settings.cosyvoice_clone_model = "voice-enrollment" + mock_settings.return_value = settings + svc = CosyVoiceService( + api_key="sk-custom", + base_url="https://custom.example.com/api/v1", + model="custom-model", + clone_model="custom-clone", + http_client=mock_client, + ) + assert svc._api_key == "sk-custom" + assert svc._base_url == "https://custom.example.com/api/v1" + assert svc._model == "custom-model" + assert svc._clone_model == "custom-clone" + + def test_context_manager(self, mock_client: MagicMock) -> None: + """上下文管理器正常工作.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + + settings.cosyvoice_api_key = "sk-test" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + with svc as s: + assert s is svc + # 注入的client不会被关闭 + mock_client.close.assert_not_called() + + def test_owns_client_gets_closed(self) -> None: + """自有client在close时被关闭.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + + settings.cosyvoice_api_key = "sk-test" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + with patch("packages.application.cosyvoice_service.httpx.Client") as mock_cls: + mock_instance = MagicMock() + mock_cls.return_value = mock_instance + svc = CosyVoiceService() + svc.close() + mock_instance.close.assert_called_once() -class TestListPresetVoices: - def test_returns_preset_voices(self) -> None: - service = _make_service() - voices = service.list_preset_voices() - assert len(voices) >= 1 - assert all(isinstance(v, PresetVoice) for v in voices) +class TestSanitizePrefix: + """_sanitize_prefix 清洗音色名称测试.""" + def test_keeps_alphanumeric(self, service: CosyVoiceService) -> None: + """保留字母数字.""" + assert service._sanitize_prefix("hello123") == "hello123" -# ── submit_clone_task ──────────────────────────────────── + def test_strips_special_chars(self, service: CosyVoiceService) -> None: + """去除特殊字符（中文isalnum返回True会保留）.""" + assert service._sanitize_prefix("hello-world!@#") == "helloworld" + + def test_truncates_to_10_chars(self, service: CosyVoiceService) -> None: + """超长截断到10字符.""" + assert service._sanitize_prefix("abcdefghijklmnop") == "abcdefghij" + + def test_empty_after_cleaning_uses_default(self, service: CosyVoiceService) -> None: + """清洗后为空用默认值clone.""" + assert service._sanitize_prefix("!!!@@@###") == "clone" + + def test_empty_string(self, service: CosyVoiceService) -> None: + """空字符串用默认值.""" + assert service._sanitize_prefix("") == "clone" class TestSubmitCloneTask: - def test_submit_success_returns_voice_id(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - { - "output": { - "voice_id": "cosyvoice-v3-flash-clone-abc123", - "status": "DEPLOYING", - }, - "usage": {"count": 1}, - "request_id": "req-001", - }, - ) + """submit_clone_task 提交音色克隆测试.""" - service = _make_service(http_client=mock_client) - result = service.submit_clone_task( - audio_url="https://example.com/audio.wav", - voice_name="myvoice", - ) - - assert result["voice_id"] == "cosyvoice-v3-flash-clone-abc123" - assert result["status"] == "DEPLOYING" - assert result["request_id"] == "req-001" - - # 验证请求参数 - call_args = mock_client.request.call_args - assert call_args.kwargs["method"] == "POST" - assert "/services/audio/tts/customization" in call_args.kwargs["url"] - - payload = call_args.kwargs["json"] - assert payload["model"] == "voice-enrollment" - assert payload["input"]["action"] == "create_voice" - assert payload["input"]["target_model"] == "cosyvoice-v3-flash" - assert payload["input"]["prefix"] == "myvoice" - assert payload["input"]["url"] == "https://example.com/audio.wav" - assert payload["input"]["language_hints"] == ["zh"] - - def test_submit_with_custom_target_model(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - { - "output": { - "voice_id": "cosyvoice-v3-flash-clone-xyz", - "status": "DEPLOYING", - }, - "request_id": "req-002", - }, - ) - - service = _make_service(http_client=mock_client) - result = service.submit_clone_task( - audio_url="https://example.com/audio.wav", - target_model="cosyvoice-v3-flash", - ) - - payload = mock_client.request.call_args.kwargs["json"] - assert payload["input"]["target_model"] == "cosyvoice-v3-flash" - assert result["voice_id"] == "cosyvoice-v3-flash-clone-xyz" - - def test_submit_empty_audio_url_raises(self) -> None: - service = _make_service() + def test_empty_audio_url_raises(self, service: CosyVoiceService) -> None: + """空audio_url抛ValueError.""" with pytest.raises(ValueError, match="audio_url 不能为空"): service.submit_clone_task(audio_url="") - def test_submit_no_api_key_raises_auth_error(self) -> None: - service = _make_service(api_key="") - with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): - service.submit_clone_task(audio_url="https://example.com/a.wav") + def test_no_api_key_raises_auth_error(self, mock_client: MagicMock) -> None: + """无API Key抛认证错误.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() - def test_submit_no_voice_id_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(200, {"output": {}}) + settings.cosyvoice_api_key = "" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"): + svc.submit_clone_task(audio_url="https://example.com/audio.mp3") + + def test_success_returns_voice_id(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """成功返回voice_id和状态.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "voice_123", "status": "DEPLOYING"}, + "request_id": "req_456", + } + mock_response.text = '{"output": {...}}' + mock_client.request.return_value = mock_response + + result = service.submit_clone_task( + audio_url="https://example.com/audio.mp3", + voice_name="myvoice", + ) + + assert result["voice_id"] == "voice_123" + assert result["status"] == "DEPLOYING" + assert result["request_id"] == "req_456" + + def test_voice_name_gets_sanitized(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """音色名称被清洗后作为prefix（中文isalnum为True会保留）.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "v1", "status": "DEPLOYING"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + service.submit_clone_task( + audio_url="https://example.com/a.mp3", + voice_name="my-voice!", + ) + + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + # 去掉横杠和感叹号，只剩myvoice + assert payload["input"]["prefix"] == "myvoice" + + def test_language_zh_cn_converts_to_zh(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """zh-CN转换为zh.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "v1", "status": "DEPLOYING"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + service.submit_clone_task( + audio_url="https://example.com/a.mp3", + language="zh-CN", + ) + + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + assert payload["input"]["language_hints"] == ["zh"] + + def test_audio_url_signer_is_called(self, mock_client: MagicMock) -> None: + """配置了audio_url_signer时会被调用预签名.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + + settings.cosyvoice_api_key = "sk-test" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + signer = MagicMock(return_value="https://signed.example.com/audio.mp3?token=xxx") + svc = CosyVoiceService(http_client=mock_client, audio_url_signer=signer) + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "v1", "status": "DEPLOYING"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + svc.submit_clone_task(audio_url="https://original.example.com/a.mp3") + + signer.assert_called_once_with("https://original.example.com/a.mp3") + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + assert payload["input"]["url"] == "https://signed.example.com/audio.mp3?token=xxx" + + def test_signer_failure_falls_back_to_original_url(self, mock_client: MagicMock) -> None: + """预签名失败时回退到原始URL.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() + + settings.cosyvoice_api_key = "sk-test" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + signer = MagicMock(side_effect=RuntimeError("sign failed")) + svc = CosyVoiceService(http_client=mock_client, audio_url_signer=signer) + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "v1", "status": "DEPLOYING"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + svc.submit_clone_task(audio_url="https://original.example.com/a.mp3") + + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + assert payload["input"]["url"] == "https://original.example.com/a.mp3" + + def test_no_voice_id_in_response_raises(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """API返回无voice_id时抛错.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"output": {}, "request_id": "r1"} + mock_response.text = "" + mock_client.request.return_value = mock_response - service = _make_service(http_client=mock_client) with pytest.raises(CosyVoiceError, match="未返回 voice_id"): - service.submit_clone_task(audio_url="https://example.com/a.wav") + service.submit_clone_task(audio_url="https://example.com/a.mp3") - def test_submit_with_audio_url_signer(self) -> None: - """传入 audio_url_signer 时，提交前会对音频URL预签名.""" - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - {"output": {"voice_id": "v123", "status": "DEPLOYING"}, "request_id": "r1"}, + def test_custom_target_model(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """自定义目标模型被传递.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"voice_id": "v1", "status": "DEPLOYING"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + service.submit_clone_task( + audio_url="https://example.com/a.mp3", + target_model="cosyvoice-v3.5", ) - def signer(url: str) -> str: - return f"{url}?signature=test" - - service = _make_service(http_client=mock_client, audio_url_signer=signer) - service.submit_clone_task(audio_url="https://oss.example.com/audio.wav") - - payload = mock_client.request.call_args.kwargs["json"] - assert payload["input"]["url"] == "https://oss.example.com/audio.wav?signature=test" - - def test_submit_signer_failure_falls_back_to_original(self) -> None: - """signer 失败时回退到原始URL，不崩溃.""" - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - {"output": {"voice_id": "v123", "status": "DEPLOYING"}, "request_id": "r1"}, - ) - - def failing_signer(url: str) -> str: - raise RuntimeError("sign failed") - - service = _make_service(http_client=mock_client, audio_url_signer=failing_signer) - result = service.submit_clone_task(audio_url="https://example.com/a.wav") - - assert result["voice_id"] == "v123" - payload = mock_client.request.call_args.kwargs["json"] - assert payload["input"]["url"] == "https://example.com/a.wav" # 原始URL - - def test_submit_prefix_sanitized(self) -> None: - """voice_name 含特殊字符时清洗为合法 prefix.""" - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, {"output": {"voice_id": "v1", "status": "DEPLOYING"}, "request_id": "r1"} - ) - - service = _make_service(http_client=mock_client) - service.submit_clone_task(audio_url="https://e.com/a.wav", voice_name="我的音色-2024!") - - payload = mock_client.request.call_args.kwargs["json"] - # 中文和特殊字符被过滤，剩下字母数字 - assert ( - payload["input"]["prefix"] == "2024" - or payload["input"]["prefix"] == "clone" - or len(payload["input"]["prefix"]) <= 10 - ) - - def test_submit_auth_401_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(401, text="Unauthorized") - - service = _make_service(http_client=mock_client) - with pytest.raises(CosyVoiceAuthError, match="认证失败"): - service.submit_clone_task(audio_url="https://e.com/a.wav") - - def test_submit_400_raises_with_code_message(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 400, {"code": "InvalidParameter", "message": "task can not be null"} - ) - - service = _make_service(http_client=mock_client) - with pytest.raises(CosyVoiceError, match="InvalidParameter"): - service.submit_clone_task(audio_url="https://e.com/a.wav") - - -# ── query_voice_status ────────────────────────────────── + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + assert payload["input"]["target_model"] == "cosyvoice-v3.5" class TestQueryVoiceStatus: - def test_query_deploying_status(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - { - "output": { - "status": "DEPLOYING", - "target_model": "cosyvoice-v3-flash", - "gmt_create": "2026-01-01T00:00:00Z", - "gmt_modified": "2026-01-01T00:01:00Z", - "resource_link": "https://...", - }, - "request_id": "req-003", - }, - ) + """query_voice_status 查询音色状态测试.""" - service = _make_service(http_client=mock_client) - result = service.query_voice_status("voice-123") - - assert result["status"] == "DEPLOYING" - assert result["target_model"] == "cosyvoice-v3-flash" - - # 验证请求 - payload = mock_client.request.call_args.kwargs["json"] - assert payload["model"] == "voice-enrollment" - assert payload["input"]["action"] == "query_voice" - assert payload["input"]["voice_id"] == "voice-123" - - def test_query_ok_status(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, {"output": {"status": "OK", "target_model": "cosyvoice-v3-flash"}} - ) - - service = _make_service(http_client=mock_client) - result = service.query_voice_status("voice-123") - assert result["status"] == "OK" - - def test_query_no_api_key_raises(self) -> None: - service = _make_service(api_key="") - with pytest.raises(CosyVoiceAuthError): - service.query_voice_status("v1") - - def test_query_empty_voice_id_raises(self) -> None: - service = _make_service() - with pytest.raises(ValueError): + def test_empty_voice_id_raises(self, service: CosyVoiceService) -> None: + """空voice_id抛ValueError.""" + with pytest.raises(ValueError, match="voice_id 不能为空"): service.query_voice_status("") + def test_no_api_key_raises(self, mock_client: MagicMock) -> None: + """无API Key抛认证错误.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() -# ── poll_clone_task ───────────────────────────────────── + settings.cosyvoice_api_key = "" + + settings.cosyvoice_base_url = "https://example.com/api/v1" + + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + with pytest.raises(CosyVoiceAuthError): + svc.query_voice_status("v1") + + def test_success_returns_status(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """成功返回状态信息.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": { + "status": "OK", + "target_model": "cosyvoice-v3-flash", + "gmt_create": "2024-01-01 00:00:00", + "gmt_modified": "2024-01-01 00:01:00", + "resource_link": "https://example.com/voice", + }, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + result = service.query_voice_status("voice_123") + + assert result["status"] == "OK" + assert result["target_model"] == "cosyvoice-v3-flash" + assert result["gmt_create"] == "2024-01-01 00:00:00" + assert result["resource_link"] == "https://example.com/voice" + + +class TestCheckTaskStatus: + """check_task_status 兼容接口测试.""" + + def test_delegates_to_query_voice_status(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """委托给query_voice_status并返回兼容格式.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"status": "DEPLOYING", "target_model": "m1"}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + result = service.check_task_status("voice_456") + + assert result["status"] == "DEPLOYING" + assert result["voice_id"] == "voice_456" + assert result["message"] == "" class TestPollCloneTask: - def test_poll_ok_on_first_check(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, {"output": {"status": "OK", "target_model": "cosyvoice-v3-flash"}} - ) + """poll_clone_task 轮询克隆状态测试.""" - service = _make_service(http_client=mock_client) - # 减少轮询间隔加速测试 - service.CLONE_POLL_INTERVAL = 0.01 + def test_success_after_several_attempts(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """几次DEPLOYING后变OK.""" + responses = [] + for _ in range(3): + r = MagicMock() + r.status_code = 200 + r.json.return_value = {"output": {"status": "DEPLOYING"}, "request_id": "r"} + r.text = "" + responses.append(r) + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {"status": "OK"}, "request_id": "r"} + ok_r.text = "" + responses.append(ok_r) - result = service.poll_clone_task("voice-123", timeout=30.0) - assert result["voice_id"] == "voice-123" + mock_client.request.side_effect = responses - def test_poll_deploying_then_ok(self) -> None: - mock_client = MagicMock() - # 第一次 DEPLOYING，第二次 OK - mock_client.request.side_effect = [ - _mock_response(200, {"output": {"status": "DEPLOYING"}}), - _mock_response(200, {"output": {"status": "OK"}}), - ] + result = service.poll_clone_task("v1", timeout=30.0) + assert result["voice_id"] == "v1" + assert mock_client.request.call_count == 4 - service = _make_service(http_client=mock_client) - service.CLONE_POLL_INTERVAL = 0.01 - - result = service.poll_clone_task("voice-123", timeout=30.0) - assert result["voice_id"] == "voice-123" - assert mock_client.request.call_count == 2 - - def test_poll_undeployed_raises_error(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(200, {"output": {"status": "UNDEPLOYED"}}) - - service = _make_service(http_client=mock_client) - service.CLONE_POLL_INTERVAL = 0.01 + def test_undeployed_raises_error(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """状态为UNDEPLOYED时抛错.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"output": {"status": "UNDEPLOYED"}, "request_id": "r"} + mock_response.text = "" + mock_client.request.return_value = mock_response with pytest.raises(CosyVoiceError, match="审核未通过"): - service.poll_clone_task("voice-123", timeout=30.0) + service.poll_clone_task("v1", timeout=30.0) - def test_poll_timeout_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(200, {"output": {"status": "DEPLOYING"}}) - - service = _make_service(http_client=mock_client) - service.CLONE_POLL_INTERVAL = 0.01 - service.CLONE_MAX_POLL_ATTEMPTS = 3 # 快速失败 + def test_timeout_raises(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """超时抛CosyVoiceTimeoutError.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"output": {"status": "DEPLOYING"}, "request_id": "r"} + mock_response.text = "" + mock_client.request.return_value = mock_response with pytest.raises(CosyVoiceTimeoutError): - service.poll_clone_task("voice-123", timeout=30.0) + service.poll_clone_task("v1", timeout=0.001) + def test_max_attempts_exceeded_raises_timeout(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """超过最大轮询次数抛超时.""" + service.CLONE_MAX_POLL_ATTEMPTS = 3 + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"output": {"status": "DEPLOYING"}, "request_id": "r"} + mock_response.text = "" + mock_client.request.return_value = mock_response -# ── clone_voice (阻塞) ────────────────────────────────── + with pytest.raises(CosyVoiceTimeoutError, match="轮询次数超限"): + service.poll_clone_task("v1", timeout=300.0) + assert mock_client.request.call_count == 3 + + def test_unknown_status_continues_polling(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """未知状态继续轮询.""" + unknown_r = MagicMock() + unknown_r.status_code = 200 + unknown_r.json.return_value = {"output": {"status": "WEIRD_STATE"}, "request_id": "r"} + unknown_r.text = "" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {"status": "OK"}, "request_id": "r"} + ok_r.text = "" + mock_client.request.side_effect = [unknown_r, ok_r] + + result = service.poll_clone_task("v1", timeout=30.0) + assert result["voice_id"] == "v1" + assert mock_client.request.call_count == 2 class TestCloneVoice: - def test_clone_already_ok_returns_immediately(self) -> None: - mock_client = MagicMock() - # submit 直接返回 OK 状态 - mock_client.request.return_value = _mock_response( - 200, - { - "output": {"voice_id": "voice-ok", "status": "OK"}, - "request_id": "req-ok", - }, + """clone_voice 完整克隆流程测试.""" + + def test_full_flow_submit_and_poll(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """提交+轮询完整流程.""" + # 第一次：提交克隆 + submit_r = MagicMock() + submit_r.status_code = 200 + submit_r.json.return_value = { + "output": {"voice_id": "v_new", "status": "DEPLOYING"}, + "request_id": "req_submit", + } + submit_r.text = "" + # 第二次：轮询DEPLOYING + poll_r1 = MagicMock() + poll_r1.status_code = 200 + poll_r1.json.return_value = {"output": {"status": "DEPLOYING"}, "request_id": "req_p1"} + poll_r1.text = "" + # 第三次：轮询OK + poll_r2 = MagicMock() + poll_r2.status_code = 200 + poll_r2.json.return_value = {"output": {"status": "OK"}, "request_id": "req_p2"} + poll_r2.text = "" + + mock_client.request.side_effect = [submit_r, poll_r1, poll_r2] + + result = service.clone_voice( + audio_url="https://example.com/a.mp3", + voice_name="test", + timeout=30.0, ) - service = _make_service(http_client=mock_client) - result = service.clone_voice(audio_url="https://e.com/a.wav") - assert isinstance(result, CloneResult) - assert result.voice_id == "voice-ok" - assert result.request_id == "req-ok" - # 只有一次调用（submit），没有 poll - assert mock_client.request.call_count == 1 - - def test_clone_deploying_then_poll_ok(self) -> None: - mock_client = MagicMock() - mock_client.request.side_effect = [ - # submit: 返回 DEPLOYING - _mock_response(200, {"output": {"voice_id": "v1", "status": "DEPLOYING"}, "request_id": "r1"}), - # poll 1: DEPLOYING - _mock_response(200, {"output": {"status": "DEPLOYING"}}), - # poll 2: OK - _mock_response(200, {"output": {"status": "OK"}}), - ] - - service = _make_service(http_client=mock_client) - service.CLONE_POLL_INTERVAL = 0.01 - - result = service.clone_voice(audio_url="https://e.com/a.wav") - assert isinstance(result, CloneResult) - assert result.voice_id == "v1" + assert result.voice_id == "v_new" + assert result.request_id == "req_submit" assert mock_client.request.call_count == 3 + def test_already_ok_on_submit_returns_immediately(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """提交时已OK直接返回，不轮询.""" + submit_r = MagicMock() + submit_r.status_code = 200 + submit_r.json.return_value = { + "output": {"voice_id": "v_ok", "status": "OK"}, + "request_id": "req_ok", + } + submit_r.text = "" + mock_client.request.return_value = submit_r -# ── synthesize_speech ──────────────────────────────────── + result = service.clone_voice(audio_url="https://example.com/a.mp3", timeout=30.0) + + assert result.voice_id == "v_ok" + # 只调用了一次（提交），没有轮询 + assert mock_client.request.call_count == 1 -class TestSynthesizeSpeech: - def test_synthesize_success_returns_audio_url(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - { - "output": { - "finish_reason": "stop", - "audio": { - "url": "https://dashscope-result.oss.com/output.mp3", - "id": "audio-001", - "expires_at": 1234567890, - }, - }, - "usage": {"characters": 10}, - "request_id": "req-syn-001", - }, - ) +class TestSubmitSynthesizeTask: + """submit_synthesize_task 提交语音合成测试.""" - service = _make_service(http_client=mock_client) - result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun_v3") + def test_empty_text_raises(self, service: CosyVoiceService) -> None: + """空text抛ValueError.""" + with pytest.raises(ValueError, match="text 不能为空"): + service.submit_synthesize_task(text="", voice_id="v1") - assert isinstance(result, SynthesizeResult) - assert result.audio_url == "https://dashscope-result.oss.com/output.mp3" - assert result.request_id == "req-syn-001" + def test_empty_voice_id_raises(self, service: CosyVoiceService) -> None: + """空voice_id抛ValueError.""" + with pytest.raises(ValueError, match="voice_id 不能为空"): + service.submit_synthesize_task(text="你好", voice_id="") - # 验证请求 - call_args = mock_client.request.call_args - assert "/services/audio/tts/SpeechSynthesizer" in call_args.kwargs["url"] + def test_no_api_key_raises(self, mock_client: MagicMock) -> None: + """无API Key抛认证错误.""" + with patch("packages.application.cosyvoice_service.get_shared_settings") as mock_settings: + settings = MagicMock() - payload = call_args.kwargs["json"] - assert payload["model"] == "cosyvoice-v3-flash" - assert payload["input"]["text"] == "你好世界" - assert payload["input"]["voice"] == "longxiaochun_v3" - assert payload["input"]["format"] == "mp3" - assert payload["input"]["sample_rate"] == 22050 - assert payload["input"]["rate"] == 1.0 - assert payload["input"]["volume"] == 50 + settings.cosyvoice_api_key = "" - def test_synthesize_with_custom_params(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - {"output": {"audio": {"url": "https://e.com/out.wav", "id": "a1"}}}, - ) + settings.cosyvoice_base_url = "https://example.com/api/v1" - service = _make_service(http_client=mock_client) - service.synthesize_speech( - text="test", + settings.cosyvoice_model = "test" + + settings.cosyvoice_sample_rate = 22050 + + settings.cosyvoice_format = "mp3" + + settings.cosyvoice_voice = "test" + + settings.cosyvoice_clone_model = "voice-enrollment" + + mock_settings.return_value = settings + svc = CosyVoiceService(http_client=mock_client) + with pytest.raises(CosyVoiceAuthError): + svc.submit_synthesize_task(text="你好", voice_id="v1") + + def test_success_returns_audio_url(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """成功返回audio_url.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"audio": {"url": "https://example.com/out.mp3"}}, + "request_id": "req_tts", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + result = service.submit_synthesize_task(text="你好世界", voice_id="voice_001") + + assert result["audio_url"] == "https://example.com/out.mp3" + assert result["request_id"] == "req_tts" + assert result["task_id"] == "" # 同步接口无task_id + + def test_no_audio_url_raises(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """返回无audio_url时抛错.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"output": {"audio": {}}, "request_id": "r1"} + mock_response.text = "" + mock_client.request.return_value = mock_response + + with pytest.raises(CosyVoiceError, match="未返回 audio_url"): + service.submit_synthesize_task(text="你好", voice_id="v1") + + def test_custom_params_passed_through(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """自定义参数被传递.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"audio": {"url": "https://example.com/out.wav"}}, + "request_id": "r1", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + service.submit_synthesize_task( + text="你好", voice_id="v1", sample_rate=44100, format="wav", @@ -432,167 +659,202 @@ class TestSynthesizeSpeech: volume=80, ) - payload = mock_client.request.call_args.kwargs["json"] + call_args = mock_client.request.call_args + payload = call_args.kwargs["json"] + assert payload["input"]["voice"] == "v1" assert payload["input"]["sample_rate"] == 44100 assert payload["input"]["format"] == "wav" assert payload["input"]["rate"] == 1.5 assert payload["input"]["volume"] == 80 - def test_synthesize_empty_text_raises(self) -> None: - service = _make_service() - with pytest.raises(ValueError, match="text 不能为空"): - service.synthesize_speech(text="", voice_id="v1") - def test_synthesize_empty_voice_raises(self) -> None: - service = _make_service() - with pytest.raises(ValueError, match="voice_id 不能为空"): - service.synthesize_speech(text="hi", voice_id="") +class TestPollSynthesizeTask: + """poll_synthesize_task 测试.""" - def test_synthesize_no_api_key_raises(self) -> None: - service = _make_service(api_key="") - with pytest.raises(CosyVoiceAuthError): - service.synthesize_speech(text="hi", voice_id="v1") - - def test_synthesize_no_audio_url_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(200, {"output": {}}) - - service = _make_service(http_client=mock_client) - with pytest.raises(CosyVoiceError, match="未返回 audio_url"): - service.synthesize_speech(text="hi", voice_id="v1") - - def test_synthesize_submit_returns_empty_task_id(self) -> None: - """同步接口的 submit_synthesize_task 返回空 task_id 字段（兼容旧接口）.""" - mock_client = MagicMock() - mock_client.request.return_value = _mock_response( - 200, - {"output": {"audio": {"url": "https://e.com/a.mp3"}}}, - ) - - service = _make_service(http_client=mock_client) - result = service.submit_synthesize_task(text="hi", voice_id="v1") - - assert result["task_id"] == "" # 同步接口无 task_id - assert result["audio_url"] == "https://e.com/a.mp3" - - def test_synthesize_auth_401_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(401, text="Unauthorized") - - service = _make_service(http_client=mock_client) - with pytest.raises(CosyVoiceAuthError): - service.synthesize_speech(text="hi", voice_id="v1") + def test_raises_error_synchronous_no_poll(self, service: CosyVoiceService) -> None: + """同步接口不支持轮询，抛错.""" + with pytest.raises(CosyVoiceError, match="无需轮询"): + service.poll_synthesize_task("task_123") -# ── retry logic ────────────────────────────────────────── +class TestSynthesizeSpeech: + """synthesize_speech 语音合成测试.""" + + def test_returns_synthesize_result(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """返回SynthesizeResult对象.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "output": {"audio": {"url": "https://example.com/out.mp3"}}, + "request_id": "req_tts", + } + mock_response.text = "" + mock_client.request.return_value = mock_response + + result = service.synthesize_speech(text="你好", voice_id="v1") + + assert isinstance(result, SynthesizeResult) + assert result.audio_url == "https://example.com/out.mp3" + assert result.request_id == "req_tts" -class TestRetryLogic: - def test_500_error_retries_then_succeeds(self) -> None: - mock_client = MagicMock() - mock_client.request.side_effect = [ - _mock_response(500, text="Server Error"), - _mock_response(502, text="Bad Gateway"), - _mock_response(200, {"output": {"audio": {"url": "https://e.com/a.mp3"}}}), - ] +class TestCallApiRetry: + """_call_api 重试逻辑测试.""" - service = _make_service(http_client=mock_client) - service.RETRY_BACKOFF = 0.01 # 加速 - - result = service.synthesize_speech(text="hi", voice_id="v1") - assert result.audio_url == "https://e.com/a.mp3" - assert mock_client.request.call_count == 3 - - def test_max_retries_exhausted_raises(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(500, text="Server Error") - - service = _make_service(http_client=mock_client) - service.MAX_RETRIES = 2 - service.RETRY_BACKOFF = 0.01 - - with pytest.raises(CosyVoiceError, match="服务端错误"): - service.synthesize_speech(text="hi", voice_id="v1") + def test_500_retries_and_succeeds(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """500错误重试后成功.""" + err_r = MagicMock() + err_r.status_code = 500 + err_r.text = "Internal Server Error" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {}, "request_id": "r"} + ok_r.text = "{}" + mock_client.request.side_effect = [err_r, ok_r] + result = service._call_api("POST", "/test", json={"k": "v"}) + assert result["request_id"] == "r" assert mock_client.request.call_count == 2 + def test_500_all_retries_exhausted(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """500重试耗尽抛错.""" + err_r = MagicMock() + err_r.status_code = 500 + err_r.text = "Internal Server Error" + mock_client.request.return_value = err_r -# ── sanitize_prefix ───────────────────────────────────── + with pytest.raises(CosyVoiceError, match="服务端错误"): + service._call_api("POST", "/test") + assert mock_client.request.call_count == service.MAX_RETRIES + + def test_401_raises_auth_no_retry(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """401直接抛认证错误，不重试.""" + err_r = MagicMock() + err_r.status_code = 401 + err_r.text = "Unauthorized" + mock_client.request.return_value = err_r + + with pytest.raises(CosyVoiceAuthError): + service._call_api("POST", "/test") + assert mock_client.request.call_count == 1 + + def test_403_raises_auth_no_retry(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """403直接抛认证错误，不重试.""" + err_r = MagicMock() + err_r.status_code = 403 + err_r.text = "Forbidden" + mock_client.request.return_value = err_r + + with pytest.raises(CosyVoiceAuthError): + service._call_api("POST", "/test") + assert mock_client.request.call_count == 1 + + def test_400_with_json_body_no_retry(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """400带JSON错误体不重试.""" + err_r = MagicMock() + err_r.status_code = 400 + err_r.json.return_value = {"code": "InvalidParameter", "message": "text too long"} + err_r.text = '{"code": "InvalidParameter", "message": "text too long"}' + mock_client.request.return_value = err_r + + with pytest.raises(CosyVoiceError, match="参数错误") as exc_info: + service._call_api("POST", "/test") + assert "InvalidParameter" in str(exc_info.value) + assert mock_client.request.call_count == 1 + + def test_400_with_plain_text_body_no_retry(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """400纯文本错误体不重试.""" + err_r = MagicMock() + err_r.status_code = 400 + err_r.json.side_effect = ValueError("not json") + err_r.text = "Bad Request: malformed input" + mock_client.request.return_value = err_r + + with pytest.raises(CosyVoiceError, match="调用失败"): + service._call_api("POST", "/test") + assert mock_client.request.call_count == 1 + + def test_timeout_retries(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """超时错误重试.""" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {}, "request_id": "r"} + ok_r.text = "{}" + mock_client.request.side_effect = [httpx.TimeoutException("timeout"), ok_r] + + result = service._call_api("POST", "/test") + assert result["request_id"] == "r" + assert mock_client.request.call_count == 2 + + def test_request_error_retries(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """请求错误重试.""" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {}, "request_id": "r"} + ok_r.text = "{}" + mock_client.request.side_effect = [httpx.ConnectError("conn err"), ok_r] + + result = service._call_api("POST", "/test") + assert result["request_id"] == "r" + assert mock_client.request.call_count == 2 + + def test_other_status_code_no_retry(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """其他状态码不重试.""" + err_r = MagicMock() + err_r.status_code = 418 + err_r.text = "I'm a teapot" + mock_client.request.return_value = err_r + + with pytest.raises(CosyVoiceError, match="HTTP 418"): + service._call_api("POST", "/test") + assert mock_client.request.call_count == 1 + + def test_bearer_token_in_headers(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """Authorization header正确设置.""" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {}, "request_id": "r"} + ok_r.text = "{}" + mock_client.request.return_value = ok_r + + service._call_api("POST", "/test") + + call_args = mock_client.request.call_args + headers = call_args.kwargs["headers"] + assert headers["Authorization"] == "Bearer sk-test-12345678" + assert headers["Content-Type"] == "application/json" + + def test_url_built_correctly(self, service: CosyVoiceService, mock_client: MagicMock) -> None: + """URL正确拼接.""" + ok_r = MagicMock() + ok_r.status_code = 200 + ok_r.json.return_value = {"output": {}, "request_id": "r"} + ok_r.text = "{}" + mock_client.request.return_value = ok_r + + service._call_api("POST", "/services/test") + + call_args = mock_client.request.call_args + assert call_args.kwargs["url"] == "https://dashscope.aliyuncs.com/api/v1/services/test" -class TestSanitizePrefix: - def test_alphanumeric_kept(self) -> None: - service = _make_service() - assert service._sanitize_prefix("myvoice123") == "myvoice123" +class TestListPresetVoices: + """list_preset_voices 预置音色测试.""" - def test_special_chars_removed(self) -> None: - service = _make_service() - result = service._sanitize_prefix("my-voice_2!") - # 特殊字符被移除，只保留字母数字 - assert result == "myvoice2" - - def test_max_10_chars(self) -> None: - service = _make_service() - result = service._sanitize_prefix("abcdefghijklmnop") - assert len(result) == 10 - - def test_empty_returns_clone(self) -> None: - service = _make_service() - assert service._sanitize_prefix("!!!???") == "clone" - assert service._sanitize_prefix("") == "clone" + def test_returns_preset_voices(self, service: CosyVoiceService) -> None: + """返回预置音色列表.""" + with patch("packages.application.cosyvoice_service.get_preset_voices") as mock_get: + mock_get.return_value = [MagicMock(), MagicMock()] + result = service.list_preset_voices() + assert len(result) == 2 + mock_get.assert_called_once() -# ── check_task_status (兼容旧接口) ────────────────────── +class TestErrorClasses: + """错误类继承关系测试.""" + def test_auth_error_inherits_from_cosyvoice_error(self) -> None: + assert issubclass(CosyVoiceAuthError, CosyVoiceError) -class TestCheckTaskStatus: - def test_check_task_status_uses_query_voice(self) -> None: - mock_client = MagicMock() - mock_client.request.return_value = _mock_response(200, {"output": {"status": "OK"}}) - - service = _make_service(http_client=mock_client) - result = service.check_task_status("voice-123") - - assert result["status"] == "OK" - assert result["voice_id"] == "voice-123" - - # 验证走的是 query_voice 路径 - payload = mock_client.request.call_args.kwargs["json"] - assert payload["input"]["action"] == "query_voice" - - -# ── 配置与初始化 ───────────────────────────────────────── - - -class TestServiceConfiguration: - """测试 CosyVoiceService 配置与初始化逻辑.""" - - def test_base_url_old_text2audio_path_auto_fixed(self) -> None: - """旧版 base_url 包含 text2audio 路径时，应自动修正为 /api/v1.""" - service = CosyVoiceService( - api_key="test-key", - base_url="https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio", - model="cosyvoice-v3-flash", - ) - # 应自动去掉 text2audio 后缀，保留到 /api/v1 - assert service._base_url == "https://dashscope.aliyuncs.com/api/v1" - - def test_base_url_normal_unchanged(self) -> None: - """正常的 base_url 不应被修改.""" - url = "https://dashscope.aliyuncs.com/api/v1" - service = CosyVoiceService( - api_key="test-key", - base_url=url, - model="cosyvoice-v3-flash", - ) - assert service._base_url == url - - def test_base_url_workspace_domain_unchanged(self) -> None: - """工作空间专属域名的 base_url 不应被修改.""" - url = "https://workspace-xxx.cn-beijing.maas.aliyuncs.com/api/v1" - service = CosyVoiceService( - api_key="test-key", - base_url=url, - model="cosyvoice-v3-flash", - ) - assert service._base_url == url + def test_timeout_error_inherits_from_cosyvoice_error(self) -> None: + assert issubclass(CosyVoiceTimeoutError, CosyVoiceError)