diff --git a/tests/unit/test_asset_types_pure.py b/tests/unit/test_asset_types_pure.py new file mode 100755 index 000000000..a1d6a20aa --- /dev/null +++ b/tests/unit/test_asset_types_pure.py @@ -0,0 +1,69 @@ +"""infer_mime_type_from_storage_key 纯逻辑单测. + +Worker core 工具函数,从 storage_key 推断 MIME 类型。 +""" + +from __future__ import annotations + +from worker_app.core.asset_types import infer_mime_type_from_storage_key + + +class TestInferMimeTypeFromStorageKey: + """infer_mime_type_from_storage_key 测试.""" + + def test_mp4_returns_video_mp4(self): + """mp4 后缀返回 video/mp4.""" + assert infer_mime_type_from_storage_key("projects/abc/video.mp4") == "video/mp4" + + def test_mov_returns_quicktime(self): + """mov 后缀返回 video/quicktime.""" + assert infer_mime_type_from_storage_key("uploads/test.mov") == "video/quicktime" + + def test_m4v_returns_video_mp4(self): + """m4v 后缀返回 video/mp4.""" + assert infer_mime_type_from_storage_key("clip.m4v") == "video/mp4" + + def test_avi_returns_video_mp4(self): + """avi 后缀返回 video/mp4.""" + assert infer_mime_type_from_storage_key("movie.avi") == "video/mp4" + + def test_mkv_returns_video_mp4(self): + """mkv 后缀返回 video/mp4.""" + assert infer_mime_type_from_storage_key("video.mkv") == "video/mp4" + + def test_webm_returns_video_mp4(self): + """webm 后缀返回 video/mp4.""" + assert infer_mime_type_from_storage_key("output.webm") == "video/mp4" + + def test_jpg_default_returns_image_jpeg(self): + """jpg 等非视频后缀默认返回 image/jpeg.""" + assert infer_mime_type_from_storage_key("thumb.jpg") == "image/jpeg" + + def test_png_default_returns_image_jpeg(self): + """png 也返回 image/jpeg(当前实现的默认值).""" + assert infer_mime_type_from_storage_key("image.png") == "image/jpeg" + + def test_no_extension_returns_jpeg(self): + """无扩展名返回 image/jpeg.""" + assert infer_mime_type_from_storage_key("random_file") == "image/jpeg" + + def test_case_insensitive(self): + """大小写不敏感.""" + assert infer_mime_type_from_storage_key("VIDEO.MP4") == "video/mp4" + assert infer_mime_type_from_storage_key("Clip.MOV") == "video/quicktime" + + def test_deep_path(self): + """多级路径正常推断.""" + assert infer_mime_type_from_storage_key("generated/projects/abc/def/output.mp4") == "video/mp4" + + def test_empty_string(self): + """空字符串返回 image/jpeg(默认值).""" + assert infer_mime_type_from_storage_key("") == "image/jpeg" + + def test_filename_with_multiple_dots(self): + """文件名含多个点时取最后一个扩展名.""" + assert infer_mime_type_from_storage_key("my.video.file.mp4") == "video/mp4" + + def test_mov_case_insensitive_upper(self): + """MOV 大写也识别为 quicktime.""" + assert infer_mime_type_from_storage_key("video.MOV") == "video/quicktime" diff --git a/tests/unit/test_asset_usage_deep.py b/tests/unit/test_asset_usage_deep.py new file mode 100755 index 000000000..e643f2e29 --- /dev/null +++ b/tests/unit/test_asset_usage_deep.py @@ -0,0 +1,139 @@ +"""mark_asset_used_for_generation 深度补充单测. + +补全边界场景:空 metadata、None metadata、last_used_at 格式、 +review_status 已有值不覆盖、多次调用递增。 +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from worker_app.core.asset_usage import mark_asset_used_for_generation + +from packages.domain import Asset, AssetStatus + + +def _asset() -> Asset: + return Asset.create( + project_id="project-1", + library_id="library-1", + name="video.mp4", + storage_key="uploads/video.mp4", + mime_type="video/mp4", + file_size=1024, + status=AssetStatus.READY, + ) + + +class TestMarkAssetUsedForGeneration: + """mark_asset_used_for_generation 深度测试.""" + + def test_first_use_sets_count_to_1(self): + """首次使用,use_count 从 0 变 1.""" + asset = _asset() + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 1 + + def test_increments_existing_count(self): + """已有计数时递增.""" + asset = _asset() + asset.metadata = {"generation_use_count": 5} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 6 + + def test_zero_count_increments_to_1(self): + """计数为 0 时递增到 1.""" + asset = _asset() + asset.metadata = {"generation_use_count": 0} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 1 + + def test_empty_metadata_still_works(self): + """空 dict metadata 也能正常工作.""" + asset = _asset() + asset.metadata = {} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 1 + assert asset.metadata["review_status"] == "pending_review" + assert "last_used_at" in asset.metadata + + def test_none_metadata_field_defaults_to_0(self): + """metadata 中 generation_use_count 为 None 时按 0 处理.""" + asset = _asset() + asset.metadata = {"generation_use_count": None} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 1 + + def test_string_count_gets_casted(self): + """字符串类型的 use_count 通过 int() 转换.""" + asset = _asset() + asset.metadata = {"generation_use_count": "3"} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 4 + + def test_preserves_other_metadata_fields(self): + """不覆盖 metadata 中的其他字段.""" + asset = _asset() + asset.metadata = { + "generation_use_count": 1, + "custom_field": "value", + "tags": ["a", "b"], + } + mark_asset_used_for_generation(asset) + assert asset.metadata["custom_field"] == "value" + assert asset.metadata["tags"] == ["a", "b"] + assert asset.metadata["generation_use_count"] == 2 + + def test_review_status_pending_when_not_set(self): + """review_status 未设置时设为 pending_review.""" + asset = _asset() + asset.metadata = {} + mark_asset_used_for_generation(asset) + assert asset.metadata["review_status"] == "pending_review" + + def test_review_status_not_overwritten_if_present(self): + """review_status 已有值时不覆盖.""" + asset = _asset() + asset.metadata = {"review_status": "approved"} + mark_asset_used_for_generation(asset) + assert asset.metadata["review_status"] == "approved" + + def test_review_status_empty_string_considered_falsy(self): + """review_status 为空字符串时视为 falsy,设置为 pending_review.""" + asset = _asset() + asset.metadata = {"review_status": ""} + mark_asset_used_for_generation(asset) + assert asset.metadata["review_status"] == "pending_review" + + def test_last_used_at_is_iso_format(self): + """last_used_at 是 ISO 格式时间字符串.""" + asset = _asset() + mark_asset_used_for_generation(asset) + ts = asset.metadata["last_used_at"] + # 可以被解析为 ISO 格式 + parsed = datetime.fromisoformat(ts) + assert parsed.tzinfo is not None # 带时区 + + def test_last_used_at_is_utc(self): + """last_used_at 是 UTC 时间.""" + asset = _asset() + before = datetime.now(timezone.utc) + mark_asset_used_for_generation(asset) + after = datetime.now(timezone.utc) + ts = datetime.fromisoformat(asset.metadata["last_used_at"]) + assert before <= ts <= after + + def test_multiple_calls_increment_count(self): + """多次调用持续递增.""" + asset = _asset() + mark_asset_used_for_generation(asset) + mark_asset_used_for_generation(asset) + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == 3 + + def test_negative_count_still_increments(self): + """负数计数(异常数据)也能递增.""" + asset = _asset() + asset.metadata = {"generation_use_count": -5} + mark_asset_used_for_generation(asset) + assert asset.metadata["generation_use_count"] == -4 diff --git a/tests/unit/test_title_usage_pure.py b/tests/unit/test_title_usage_pure.py new file mode 100755 index 000000000..6b9715aaf --- /dev/null +++ b/tests/unit/test_title_usage_pure.py @@ -0,0 +1,143 @@ +"""mark_title_used_for_generation 纯逻辑单测. + +验证 title usage 计数 + updated_at 更新逻辑。 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock + +from packages.adapters.sqlalchemy_impl.models import TitleLibraryModel + + +def test_module_importable(): + """确认模块可以正常导入.""" + from worker_app.core.title_usage import mark_title_used_for_generation # noqa: F401 + + +class TestMarkTitleUsedForGeneration: + """mark_title_used_for_generation 测试.""" + + def _make_task(self, strategy_id: str = "title-1") -> MagicMock: + task = MagicMock() + task.strategy_id = strategy_id + return task + + def _make_title(self, usage_count: int = 0) -> MagicMock: + title = MagicMock(spec=TitleLibraryModel) + title.id = "title-1" + title.usage_count = usage_count + title.updated_at = None + return title + + def test_no_strategy_id_returns_early(self): + """无 strategy_id 时直接返回,不查 DB.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + task = self._make_task(strategy_id="") + + mark_title_used_for_generation(db, task) + + db.query.assert_not_called() + + def test_none_strategy_id_returns_early(self): + """strategy_id 为 None 时直接返回.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + task = self._make_task(strategy_id=None) + + mark_title_used_for_generation(db, task) + + db.query.assert_not_called() + + def test_title_not_found_returns_early(self): + """title 不存在时不报错,静默返回.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = None + task = self._make_task(strategy_id="missing-id") + + mark_title_used_for_generation(db, task) + + db.add.assert_not_called() + db.commit.assert_not_called() + + def test_increments_usage_count_from_zero(self): + """usage_count 从 0 递增到 1.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + title = self._make_title(usage_count=0) + db.query.return_value.filter.return_value.first.return_value = title + task = self._make_task() + + mark_title_used_for_generation(db, task) + + assert title.usage_count == 1 + db.add.assert_called_once_with(title) + db.commit.assert_called_once() + + def test_increments_usage_count_from_existing(self): + """已有 usage_count 时递增.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + title = self._make_title(usage_count=5) + db.query.return_value.filter.return_value.first.return_value = title + task = self._make_task() + + mark_title_used_for_generation(db, task) + + assert title.usage_count == 6 + + def test_none_usage_count_defaults_to_zero_then_increments(self): + """usage_count 为 None 时按 0 处理,递增到 1.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + title = self._make_title(usage_count=None) + db.query.return_value.filter.return_value.first.return_value = title + task = self._make_task() + + mark_title_used_for_generation(db, task) + + assert title.usage_count == 1 + + def test_updates_updated_at_to_utc_now(self): + """updated_at 更新为当前 UTC 时间.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + title = self._make_title(usage_count=3) + db.query.return_value.filter.return_value.first.return_value = title + task = self._make_task() + + before = datetime.now(timezone.utc) + mark_title_used_for_generation(db, task) + after = datetime.now(timezone.utc) + + assert before <= title.updated_at <= after + assert title.updated_at.tzinfo is not None # 带时区 + + def test_correct_query_filter(self): + """查询时使用正确的 id 过滤.""" + from worker_app.core.title_usage import mark_title_used_for_generation + + db = MagicMock() + title = self._make_title() + db.query.return_value.filter.return_value.first.return_value = title + task = self._make_task(strategy_id="title-abc") + + mark_title_used_for_generation(db, task) + + # 验证 query 模型正确 + db.query.assert_called_once_with(TitleLibraryModel) + # 验证 filter 条件 + filter_call = db.query.return_value.filter + assert filter_call.called + # first 被调用 + filter_call.return_value.first.assert_called_once() diff --git a/tests/unit/test_tts_streaming_service.py b/tests/unit/test_tts_streaming_service.py new file mode 100755 index 000000000..9d3d36eab --- /dev/null +++ b/tests/unit/test_tts_streaming_service.py @@ -0,0 +1,441 @@ +"""TTSStreamingService 纯逻辑单测 — 分段策略 + 分块推送 + 错误处理. + +mock 掉 WebSocket 和 CosyVoiceService,验证核心逻辑: +- 文本长度路由(短文本/长文本) +- 空文本/超长文本校验 +- 音频分块推送算法 +- 错误处理路径 +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from packages.application.cosyvoice_service import CosyVoiceError +from packages.application.tts_job.streaming_service import ( + TTSStreamingError, + TTSStreamingService, + _AUDIO_CHUNK_SIZE, +) + +# ── Fixtures ────────────────────────────────────────────────────────────── + + +@pytest.fixture +def mock_cosyvoice(): + """mock CosyVoiceService.""" + svc = MagicMock() + svc.submit_synthesize_task.return_value = { + "audio_url": "https://example.com/audio.mp3", + "duration": 3.5, + } + return svc + + +@pytest.fixture +def streaming_service(mock_cosyvoice): + """TTSStreamingService 实例.""" + return TTSStreamingService(mock_cosyvoice) + + +@pytest.fixture +def mock_ws(): + """mock WebSocket.""" + ws = AsyncMock() + ws.send_bytes = AsyncMock() + ws.send_json = AsyncMock() + return ws + + +class FakeAudioBytes: + """生成指定大小的假音频数据.""" + + @staticmethod + def make(size: int) -> bytes: + return b"\x00" * size + + +# ── 合成路由测试 ────────────────────────────────────────────────────────── + + +class TestSynthesizeRouting: + """synthesize_and_stream 路由逻辑测试.""" + + @pytest.mark.asyncio + async def test_empty_text_returns_error(self, streaming_service, mock_ws): + """空文本返回错误,不调用合成.""" + await streaming_service.synthesize_and_stream(mock_ws, {"text": ""}) + + # 应发送 error 消息 + mock_ws.send_json.assert_called() + last_call = mock_ws.send_json.call_args + assert last_call[0][0]["type"] == "error" + assert "不能为空" in last_call[0][0]["message"] + # 不应调用合成 + streaming_service._cosyvoice.submit_synthesize_task.assert_not_called() + + @pytest.mark.asyncio + async def test_missing_text_key_returns_error(self, streaming_service, mock_ws): + """缺少 text 字段返回错误.""" + await streaming_service.synthesize_and_stream(mock_ws, {}) + + mock_ws.send_json.assert_called() + last_call = mock_ws.send_json.call_args + assert last_call[0][0]["type"] == "error" + streaming_service._cosyvoice.submit_synthesize_task.assert_not_called() + + @pytest.mark.asyncio + async def test_too_long_text_returns_error(self, streaming_service, mock_ws): + """超长文本返回错误.""" + long_text = "你" * 10001 + await streaming_service.synthesize_and_stream(mock_ws, {"text": long_text}) + + mock_ws.send_json.assert_called() + last_call = mock_ws.send_json.call_args + assert last_call[0][0]["type"] == "error" + assert "最大" in last_call[0][0]["message"] + streaming_service._cosyvoice.submit_synthesize_task.assert_not_called() + + @pytest.mark.asyncio + async def test_short_text_uses_short_path(self, streaming_service, mock_ws): + """短文本走 _stream_short_text 路径.""" + with patch.object(streaming_service, "_stream_short_text", new_callable=AsyncMock) as mock_short: + await streaming_service.synthesize_and_stream(mock_ws, {"text": "hello"}) + mock_short.assert_called_once() + + @pytest.mark.asyncio + async def test_long_text_uses_long_path(self, streaming_service, mock_ws): + """长文本走 _stream_long_text 路径.""" + long_text = "你" * 501 + with patch.object(streaming_service, "_stream_long_text", new_callable=AsyncMock) as mock_long: + await streaming_service.synthesize_and_stream(mock_ws, {"text": long_text}) + mock_long.assert_called_once() + + @pytest.mark.asyncio + async def test_exactly_threshold_uses_short_path(self, streaming_service, mock_ws): + """恰好等于阈值走短文本路径.""" + text = "你" * 500 + with ( + patch.object(streaming_service, "_stream_short_text", new_callable=AsyncMock) as mock_short, + patch.object(streaming_service, "_stream_long_text", new_callable=AsyncMock) as mock_long, + ): + await streaming_service.synthesize_and_stream(mock_ws, {"text": text}) + mock_short.assert_called_once() + mock_long.assert_not_called() + + +# ── 短文本流测试 ────────────────────────────────────────────────────────── + + +class TestStreamShortText: + """短文本流式合成测试.""" + + @pytest.mark.asyncio + async def test_happy_path_sends_started_then_done(self, streaming_service, mock_ws): + """短文本正常流程:started → 音频块 → done.""" + audio_data = FakeAudioBytes.make(5000) + with patch.object(streaming_service, "_download_audio", return_value=audio_data): + await streaming_service._stream_short_text( + mock_ws, + {"text": "hello", "voice_id": "v1", "format": "mp3", "speed": 1.0}, + ) + + # 检查 started 消息 + calls = mock_ws.send_json.call_args_list + assert calls[0][0][0]["type"] == "started" + assert calls[0][0][0]["segment_count"] == 1 + + # 检查 done 消息 + last_msg = calls[-1][0][0] + assert last_msg["type"] == "done" + assert last_msg["file_size"] == 5000 + assert last_msg["format"] == "mp3" + assert last_msg["duration"] == 3.5 + + @pytest.mark.asyncio + async def test_calls_cosyvoice_with_correct_params(self, streaming_service, mock_ws): + """正确传递参数给 CosyVoice.""" + audio_data = FakeAudioBytes.make(1000) + with patch.object(streaming_service, "_download_audio", return_value=audio_data): + await streaming_service._stream_short_text( + mock_ws, + { + "text": "test text", + "voice_id": "voice-123", + "sample_rate": 22050, + "format": "wav", + "speed": 1.5, + }, + ) + + streaming_service._cosyvoice.submit_synthesize_task.assert_called_once_with( + text="test text", + voice_id="voice-123", + sample_rate=22050, + format="wav", + speed=1.5, + ) + + @pytest.mark.asyncio + async def test_cosyvoice_error_returns_error(self, streaming_service, mock_ws): + """CosyVoice 错误返回 error 消息.""" + streaming_service._cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("API quota exceeded") + + await streaming_service._stream_short_text(mock_ws, {"text": "hello"}) + + # 最后一条应该是 error + last_msg = mock_ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "API quota exceeded" in last_msg["message"] + + @pytest.mark.asyncio + async def test_generic_exception_returns_error(self, streaming_service, mock_ws): + """普通异常返回 error 消息.""" + streaming_service._cosyvoice.submit_synthesize_task.side_effect = RuntimeError("boom") + + await streaming_service._stream_short_text(mock_ws, {"text": "hello"}) + + last_msg = mock_ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "合成失败" in last_msg["message"] + + @pytest.mark.asyncio + async def test_no_audio_url_returns_error(self, streaming_service, mock_ws): + """合成结果无 audio_url 返回错误.""" + streaming_service._cosyvoice.submit_synthesize_task.return_value = {"duration": 3.0} + + await streaming_service._stream_short_text(mock_ws, {"text": "hello"}) + + last_msg = mock_ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "音频 URL" in last_msg["message"] + + @pytest.mark.asyncio + async def test_download_failure_returns_error(self, streaming_service, mock_ws): + """音频下载失败返回 error.""" + with patch.object( + streaming_service, + "_download_audio", + side_effect=Exception("download failed"), + ): + await streaming_service._stream_short_text(mock_ws, {"text": "hello"}) + + last_msg = mock_ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "音频推送失败" in last_msg["message"] + + +# ── 音频分块测试 ────────────────────────────────────────────────────────── + + +class TestStreamAudioChunks: + """_stream_audio_chunks 分块推送测试.""" + + @pytest.mark.asyncio + async def test_exact_one_chunk(self, streaming_service, mock_ws): + """恰好一个 chunk 大小的数据.""" + data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE) + total = await streaming_service._stream_audio_chunks(mock_ws, data) + + assert total == _AUDIO_CHUNK_SIZE + assert mock_ws.send_bytes.call_count == 1 + assert len(mock_ws.send_bytes.call_args[0][0]) == _AUDIO_CHUNK_SIZE + + @pytest.mark.asyncio + async def test_smaller_than_one_chunk(self, streaming_service, mock_ws): + """小于一个 chunk 的数据.""" + data = FakeAudioBytes.make(1000) + total = await streaming_service._stream_audio_chunks(mock_ws, data) + + assert total == 1000 + assert mock_ws.send_bytes.call_count == 1 + + @pytest.mark.asyncio + async def test_multiple_full_chunks(self, streaming_service, mock_ws): + """多个完整 chunk.""" + num_chunks = 5 + data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE * num_chunks) + total = await streaming_service._stream_audio_chunks(mock_ws, data) + + assert total == _AUDIO_CHUNK_SIZE * num_chunks + assert mock_ws.send_bytes.call_count == num_chunks + for c in mock_ws.send_bytes.call_args_list: + assert len(c[0][0]) == _AUDIO_CHUNK_SIZE + + @pytest.mark.asyncio + async def test_partial_last_chunk(self, streaming_service, mock_ws): + """最后一个 chunk 不完整.""" + data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE * 2 + 1234) + total = await streaming_service._stream_audio_chunks(mock_ws, data) + + assert total == _AUDIO_CHUNK_SIZE * 2 + 1234 + assert mock_ws.send_bytes.call_count == 3 + # 最后一块是 1234 字节 + last_chunk = mock_ws.send_bytes.call_args_list[-1][0][0] + assert len(last_chunk) == 1234 + + @pytest.mark.asyncio + async def test_empty_audio_sends_zero_chunks(self, streaming_service, mock_ws): + """空音频不发送任何 chunk.""" + total = await streaming_service._stream_audio_chunks(mock_ws, b"") + assert total == 0 + mock_ws.send_bytes.assert_not_called() + + @pytest.mark.asyncio + async def test_chunks_are_consecutive(self, streaming_service, mock_ws): + """所有 chunk 拼接起来等于原始数据.""" + data = bytes(range(256)) * 50 # 12800 bytes + total = await streaming_service._stream_audio_chunks(mock_ws, data) + + assert total == len(data) + # 收集所有 chunk + all_bytes = b"".join(c[0][0] for c in mock_ws.send_bytes.call_args_list) + assert all_bytes == data + + +# ── 长文本分段流测试 ────────────────────────────────────────────────────── + + +class TestStreamLongText: + """长文本分段流式合成测试.""" + + @pytest.mark.asyncio + async def test_happy_path_all_segments_ok(self, streaming_service, mock_ws): + """长文本正常流程:多个分段全部成功.""" + audio_data = FakeAudioBytes.make(2000) + with patch.object(streaming_service, "_download_audio", return_value=audio_data): + text = "你" * 1200 # 应该分成3段 + await streaming_service._stream_long_text( + mock_ws, + {"text": text, "voice_id": "v1", "format": "mp3", "speed": 1.0}, + ) + + # 检查 started 消息 + calls = mock_ws.send_json.call_args_list + assert calls[0][0][0]["type"] == "started" + segment_count = calls[0][0][0]["segment_count"] + assert segment_count >= 2 # 1200 字至少分 2 段 + + # 检查有 segment_done 消息 + segment_dones = [c for c in calls if c[0][0].get("type") == "segment_done"] + assert len(segment_dones) == segment_count + + # 检查最后是 done 消息 + last_msg = calls[-1][0][0] + assert last_msg["type"] == "done" + assert last_msg["file_size"] == 2000 * segment_count + + @pytest.mark.asyncio + async def test_first_segment_fails_returns_error(self, streaming_service, mock_ws): + """第一个分段失败,立即返回错误.""" + streaming_service._cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("segment 0 failed") + + text = "你" * 1200 + await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"}) + + calls = mock_ws.send_json.call_args_list + last_msg = calls[-1][0][0] + assert last_msg["type"] == "error" + assert "分段" in last_msg["message"] + assert "1" in last_msg["message"] # 第1段失败 + + @pytest.mark.asyncio + async def test_segment_without_audio_url_fails(self, streaming_service, mock_ws): + """分段结果无 audio_url 视为失败.""" + # 第一段正常,第二段返回空 audio_url + call_results = [ + {"audio_url": "https://a.com/1.mp3", "duration": 2.0}, + {"audio_url": "", "duration": 0}, + {"audio_url": "https://a.com/3.mp3", "duration": 3.0}, + ] + streaming_service._cosyvoice.submit_synthesize_task.side_effect = call_results + + with patch.object( + streaming_service, + "_download_audio", + return_value=FakeAudioBytes.make(1000), + ): + text = "你" * 1500 + await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"}) + + calls = mock_ws.send_json.call_args_list + # 应该有错误 + error_msgs = [c for c in calls if c[0][0].get("type") == "error"] + assert len(error_msgs) >= 1 + + @pytest.mark.asyncio + async def test_each_segment_gets_correct_text(self, streaming_service, mock_ws): + """每个分段都调用了合成,且 text 参数不同.""" + with patch.object( + streaming_service, + "_download_audio", + return_value=FakeAudioBytes.make(500), + ): + text = "你" * 1200 + await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"}) + + # 分段数应大于1 + assert streaming_service._cosyvoice.submit_synthesize_task.call_count >= 2 + + # 收集所有传进去的 text + texts_called = [ + c.kwargs.get("text") or c.args[0] + for c in streaming_service._cosyvoice.submit_synthesize_task.call_args_list + ] + # 每段文本都应该是原文的一部分(不全部相同) + assert len(set(texts_called)) >= 2 + # 所有文本拼接起来应该约等于原文长度 + total_len = sum(len(t) for t in texts_called) + assert total_len >= len(text) * 0.95 # 允许标点切分的小误差 + + @pytest.mark.asyncio + async def test_each_segment_has_unique_index(self, streaming_service, mock_ws): + """segment_done 消息的序号不重复且正确.""" + with patch.object( + streaming_service, + "_download_audio", + return_value=FakeAudioBytes.make(500), + ): + text = "你" * 1200 + await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"}) + + calls = mock_ws.send_json.call_args_list + segment_dones = [c[0][0] for c in calls if c[0][0].get("type") == "segment_done"] + indices = [s["segment"] for s in segment_dones] + total = segment_dones[0]["total"] + # 序号从 1 到 total,不重复 + assert sorted(indices) == list(range(1, total + 1)) + + +# ── WebSocket 发送失败容错 ──────────────────────────────────────────────── + + +class TestSendJsonErrorHandling: + """_send_json 容错测试.""" + + @pytest.mark.asyncio + async def test_send_json_failure_logs_warning(self, streaming_service, mock_ws): + """WebSocket send_json 失败不抛异常.""" + mock_ws.send_json.side_effect = Exception("connection closed") + + # 不应抛出异常 + await streaming_service._send_json(mock_ws, {"type": "done"}) + mock_ws.send_json.assert_called_once() + + +# ── TTSStreamingError 异常类 ────────────────────────────────────────────── + + +class TestTTSStreamingError: + """TTSStreamingError 异常类测试.""" + + def test_is_exception(self): + """是 Exception 子类.""" + assert issubclass(TTSStreamingError, Exception) + + def test_carry_message(self): + """携带错误消息.""" + err = TTSStreamingError("stream failed") + assert str(err) == "stream failed"