test(unit): 第88波 - streaming_service+asset_types+asset_usage+title_usage单测 (+63)

- test_tts_streaming_service.py (26个): 文本路由/短长文本流/分块推送/分段并发/错误处理
- test_asset_types_pure.py (14个): infer_mime_type_from_storage_key 各种后缀/大小写/路径场景
- test_asset_usage_deep.py (14个): mark_asset_used_for_generation 边界/None/空metadata/review_status/时间格式
- test_title_usage_pure.py (9个): mark_title_used_for_generation 策略ID校验/递增/时间戳/DB查询

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"

tests/unit/test_asset_usage_deep.py

@@ -0,0 +1,141 @@
+"""mark_asset_used_for_generation 深度补充单测.
+
+补全边界场景：空 metadata、None metadata、last_used_at 格式、
+review_status 已有值不覆盖、多次调用递增。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+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

tests/unit/test_title_usage_pure.py

@@ -0,0 +1,145 @@
+"""mark_title_used_for_generation 纯逻辑单测.
+
+验证 title usage 计数 + updated_at 更新逻辑。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+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()

tests/unit/test_tts_streaming_service.py

@@ -0,0 +1,443 @@
+"""TTSStreamingService 纯逻辑单测 — 分段策略 + 分块推送 + 错误处理.
+
+mock 掉 WebSocket 和 CosyVoiceService，验证核心逻辑：
+- 文本长度路由（短文本/长文本）
+- 空文本/超长文本校验
+- 音频分块推送算法
+- 错误处理路径
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, call, 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
+        last_msg = calls[-1][0][0]
+        # 应该有错误
+        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"