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查询

apps/worker/video_processing/unified_render_service.py

@@ -52,15 +52,6 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr
 from video_processing.tts_engine import TtsEngine
 from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
 
-from packages.domain.render_layer_utils import (
-    LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX,
-    can_pass_through as _can_pass_through_pure,
-    clip_adjusted_duration as _clip_adjusted_duration_pure,
-    clip_effective_duration as _clip_effective_duration_pure,
-    clip_playback_speed as _clip_playback_speed_pure,
-    estimate_total_duration as _estimate_total_duration_pure,
-    resolve_layer_role as _resolve_layer_role_pure,
-)
 from packages.domain.tts_config import TtsConfig
 
 logger = logging.getLogger(__name__)
@@ -116,16 +107,47 @@ class RenderResult:
 
 
 def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
-    """根据 clip_type 和 config.role 确定图层角色（向后兼容别名）。
+    """根据 clip_type 和 config.role 确定图层角色。
 
-    实际实现移至 packages.domain.render_layer_utils.resolve_layer_role。
+    映射规则：
+      intro / outro         → "main"（按 order 排在首/尾）
+      overlay               → "overlay"（画中画叠加，z=1）
+      corner_voice          → "corner_voice"（右上角小窗，z=1）
+      background            → "background"（全屏底图，z=0）
+      b_roll                → "broll"（z=0）
+      main + config.role=b_roll → "broll"
+      main (default)        → "main"
     """
-    return _resolve_layer_role_pure(clip_type, config)
+    role = config.get("role", "")
+
+    if clip_type in ("intro", "outro"):
+        return "main"
+    if clip_type == "overlay":
+        return "overlay"
+    if clip_type == "corner_voice":
+        return "corner_voice"
+    if clip_type == "background":
+        return "background"
+    if clip_type == "b_roll":
+        return "broll"
+    # main type
+    if role == "b_roll":
+        return "broll"
+    if role == "audio":
+        return "audio"
+    return "main"
 
 
 # ── 图层默认 z_index ─────────────────────────────────────────────────────────
 
-_LAYER_Z_INDEX: dict[str, int] = _IMPORTED_LAYER_Z_INDEX
+_LAYER_Z_INDEX: dict[str, int] = {
+    "background": -1,
+    "broll": 0,
+    "main": 0,
+    "overlay": 1,
+    "corner_voice": 1,
+    "audio": 2,
+}
 
 # 图层默认 PiP 位置（相对输出画布的偏移）
 _PIP_SCALE = 0.25  # PiP 占主画面的比例
@@ -467,9 +489,29 @@ class UnifiedRenderService:
     def _estimate_total_duration(self, layers: list[RenderLayer]) -> float:
         """估算视频总时长（用于字幕等需要）。
 
-        实际实现移至 packages.domain.render_layer_utils.estimate_total_duration。
+        取主图层（main/broll/background）的总时长，转场重叠按 transition_duration 估算。
         """
-        return _estimate_total_duration_pure(layers, self.transition_duration)
+        # 找主图层（第一个有视频内容的图层）
+        main_layer = None
+        for role in ("main", "broll", "background"):
+            for layer in layers:
+                if layer.role == role:
+                    main_layer = layer
+                    break
+            if main_layer:
+                break
+
+        if not main_layer or not main_layer.clips:
+            return 0.0
+
+        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
+
+        # 减去转场重叠时间（粗略估算）
+        n_clips = len(main_layer.clips)
+        if n_clips > 1:
+            total -= (n_clips - 1) * self.transition_duration
+
+        return max(0.1, total)
 
     def _maybe_generate_ass(self, video_duration: float) -> Path | None:
         """根据 plan.config 生成 ASS 字幕文件。
@@ -1827,11 +1869,10 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长（原速 trim 后时长）。
-
-        实际实现移至 packages.domain.render_layer_utils.clip_effective_duration。
-        """
-        return _clip_effective_duration_pure(clip.duration, clip.actual_duration)
+        """计算 clip 的有效时长（原速 trim 后时长）."""
+        if clip.duration > 0:
+            return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+        return clip.actual_duration if clip.actual_duration > 0 else 0.0
 
     # ── 画中画（PiP）相关方法 ──────────────────────────────────────────────────
 
@@ -1928,20 +1969,17 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_speed(clip: ResolvedClip) -> float:
-        """获取 clip 的播放速度，无效值回退到 1.0。
-
-        实际实现移至 packages.domain.render_layer_utils.clip_playback_speed。
-        """
-        return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))
+        """获取 clip 的播放速度，无效值回退到 1.0."""
+        speed = getattr(clip, "playback_speed", 1.0)
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return 1.0
+        return float(speed)
 
     @staticmethod
     def _clip_adjusted_duration(clip: ResolvedClip) -> float:
-        """计算调速后的 clip 实际时长（用于拼接计算）。
-
-        实际实现移至 packages.domain.render_layer_utils.clip_adjusted_duration。
-        """
-        return _clip_adjusted_duration_pure(
-            clip.duration,
-            clip.actual_duration,
-            getattr(clip, "playback_speed", 1.0),
-        )
+        """计算调速后的 clip 实际时长（用于拼接计算）."""
+        base = UnifiedRenderService._clip_effective_duration(clip)
+        speed = UnifiedRenderService._clip_speed(clip)
+        if abs(speed - 1.0) < 1e-6:
+            return base
+        return base / speed

packages/domain/render_layer_utils.py

@@ -1,241 +0,0 @@
-"""渲染图层工具函数 — 纯函数集合.
-
-从 unified_render_service.py 抽离的纯逻辑，负责：
-- clip 时长计算（有效时长、调速后时长）
-- clip_type → layer_role 映射
-- 总时长估算
-- 图层默认属性（z_index 等）
-
-所有函数均为纯函数，不依赖 FFmpeg、数据库或外部 IO。
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-# ── 图层角色定义 ─────────────────────────────────────────────────────────────
-
-# 图层默认 z_index 映射
-LAYER_Z_INDEX: dict[str, int] = {
-    "background": -1,
-    "broll": 0,
-    "main": 0,
-    "overlay": 1,
-    "corner_voice": 1,
-    "audio": 2,
-}
-
-# 图层默认 PiP 缩放比例（相对于主画面）
-PIP_DEFAULT_SCALE = 0.25
-
-# 主视频图层角色（用于总时长计算、直通判断等）
-MAIN_LAYER_ROLES = frozenset({"main", "broll", "background"})
-
-
-# ── clip_type → layer_role 映射 ──────────────────────────────────────────────
-
-
-def resolve_layer_role(clip_type: str, config: dict[str, Any] | None = None) -> str:
-    """根据 clip_type 和 config.role 确定图层角色。
-
-    映射规则：
-      intro / outro         → "main"（按 order 排在首/尾）
-      overlay                → "overlay"（画中画叠加，z=1）
-      corner_voice           → "corner_voice"（右上角小窗，z=1）
-      background             → "background"（全屏底图，z=0）
-      b_roll                 → "broll"（z=0）
-      main + config.role=b_roll → "broll"
-      main + config.role=audio  → "audio"
-      main (default)         → "main"
-
-    Args:
-        clip_type: 片段类型字符串
-        config: 片段配置字典（可选）
-
-    Returns:
-        图层角色字符串
-    """
-    role = (config or {}).get("role", "") if config else ""
-
-    if clip_type in ("intro", "outro"):
-        return "main"
-    if clip_type == "overlay":
-        return "overlay"
-    if clip_type == "corner_voice":
-        return "corner_voice"
-    if clip_type == "background":
-        return "background"
-    if clip_type == "b_roll":
-        return "broll"
-    # main type
-    if role == "b_roll":
-        return "broll"
-    if role == "audio":
-        return "audio"
-    return "main"
-
-
-def get_layer_z_index(role: str) -> int:
-    """获取图层角色的默认 z_index。
-
-    Args:
-        role: 图层角色
-
-    Returns:
-        z_index 值，未知角色返回 0
-    """
-    return LAYER_Z_INDEX.get(role, 0)
-
-
-# ── clip 时长计算 ──────────────────────────────────────────────────────────
-
-
-def clip_effective_duration(
-    duration: float,
-    actual_duration: float = 0.0,
-) -> float:
-    """计算 clip 的有效时长（原速 trim 后时长）。
-
-    规则：
-    - duration > 0: min(duration, actual_duration)，actual=0 时用 duration
-    - duration <= 0: actual_duration，actual=0 时返回 0
-
-    Args:
-        duration: 配置的时长（0 表示使用完整素材）
-        actual_duration: 素材实际时长（probe 后的结果）
-
-    Returns:
-        有效时长（秒）
-    """
-    if duration > 0:
-        return min(duration, actual_duration) if actual_duration > 0 else duration
-    return actual_duration if actual_duration > 0 else 0.0
-
-
-def clip_playback_speed(playback_speed: Any) -> float:
-    """获取 clip 的播放速度，无效值回退到 1.0。
-
-    Args:
-        playback_speed: 播放速度（可为任意类型
-
-    Returns:
-        有效的播放速度（正数）
-    """
-    if not isinstance(playback_speed, (int, float)):
-        return 1.0
-    if playback_speed <= 0:
-        return 1.0
-    return float(playback_speed)
-
-
-def clip_adjusted_duration(
-    duration: float,
-    actual_duration: float = 0.0,
-    playback_speed: Any = 1.0,
-) -> float:
-    """计算调速后的 clip 实际时长（用于拼接计算）。
-
-    Args:
-        duration: 配置的时长
-        actual_duration: 素材实际时长
-        playback_speed: 播放速度
-
-    Returns:
-        调速后的时长
-    """
-    base = clip_effective_duration(duration, actual_duration)
-    speed = clip_playback_speed(playback_speed)
-    if abs(speed - 1.0) < 1e-6:
-        return base
-    return base / speed
-
-
-# ── 总时长估算 ────────────────────────────────────────────────────────────
-
-
-def estimate_total_duration(
-    layers: list[Any],
-    transition_duration: float = 0.0,
-) -> float:
-    """估算视频总时长。
-
-    取主图层（main/broll/background）的总调整后时长，减去转场重叠时间。
-
-    Args:
-        layers: 图层列表（每个元素需有 role 和 clips 属性，
-                clips 中元素需有 duration/actual_duration/playback_speed 属性）
-        transition_duration: 转场时长（秒），用于估算重叠时间
-
-    Returns:
-        估算的总时长（秒），最小 0.1
-    """
-    # 找主图层（第一个有视频内容的图层）
-    main_layer = None
-    for role in ("main", "broll", "background"):
-        for layer in layers:
-            if getattr(layer, "role", None) == role and getattr(layer, "clips", None):
-                main_layer = layer
-                break
-        if main_layer:
-            break
-
-    if not main_layer or not getattr(main_layer, "clips", None):
-        return 0.0
-
-    clips = getattr(main_layer, "clips", [])
-    total = sum(
-        clip_adjusted_duration(
-            duration=getattr(c, "duration", 0),
-            actual_duration=getattr(c, "actual_duration", 0.0),
-            playback_speed=getattr(c, "playback_speed", 1.0),
-        )
-        for c in clips
-    )
-
-    # 减去转场重叠时间（粗略估算）
-    n_clips = len(clips)
-    if n_clips > 1 and transition_duration > 0:
-        total -= (n_clips - 1) * transition_duration
-
-    return max(0.1, total)
-
-
-# ── 直通 / Stream Copy 判断辅助 ──────────────────────────────────────────────
-
-
-def can_pass_through(
-    layers: list[Any],
-    has_stickers: bool = False,
-    has_watermark: bool = False,
-) -> bool:
-    """判断是否可以走直通优化路径（单 clip 简单场景）。
-
-    条件：
-    1. 只有 1 个图层
-    2. 该图层是视频图层（main/broll/background）
-    3. 该图层只有 1 个 clip（无转场需求）
-    4. 没有贴纸
-    5. 没有水印
-
-    Args:
-        layers: 图层列表
-        has_stickers: 是否有贴纸
-        has_watermark: 是否有水印
-
-    Returns:
-        是否可以走直通
-    """
-    if len(layers) != 1:
-        return False
-    layer = layers[0]
-    role = getattr(layer, "role", "")
-    if role not in MAIN_LAYER_ROLES:
-        return False
-    clips = getattr(layer, "clips", [])
-    if len(clips) != 1:
-        return False
-    if has_stickers:
-        return False
-    if has_watermark:
-        return False
-    return True

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_render_layer_utils.py

@@ -1,361 +0,0 @@
-"""render_layer_utils 模块单元测试."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any
-
-import pytest
-
-from packages.domain.render_layer_utils import (
-    LAYER_Z_INDEX,
-    MAIN_LAYER_ROLES,
-    PIP_DEFAULT_SCALE,
-    can_pass_through,
-    clip_adjusted_duration,
-    clip_effective_duration,
-    clip_playback_speed,
-    estimate_total_duration,
-    get_layer_z_index,
-    resolve_layer_role,
-)
-
-# ── 辅助数据类 ──────────────────────────────────────────────────────────────
-
-
-@dataclass
-class FakeClip:
-    duration: float = 0.0
-    actual_duration: float = 0.0
-    playback_speed: Any = 1.0
-
-
-@dataclass
-class FakeLayer:
-    role: str = "main"
-    clips: list[FakeClip] = field(default_factory=list)
-
-
-# ── 常量验证 ────────────────────────────────────────────────────────────────
-
-
-class TestConstants:
-    def test_layer_z_index_has_expected_keys(self):
-        assert set(LAYER_Z_INDEX.keys()) == {
-            "background",
-            "broll",
-            "main",
-            "overlay",
-            "corner_voice",
-            "audio",
-        }
-
-    def test_layer_z_index_ordering(self):
-        assert LAYER_Z_INDEX["background"] < LAYER_Z_INDEX["main"]
-        assert LAYER_Z_INDEX["main"] == LAYER_Z_INDEX["broll"]
-        assert LAYER_Z_INDEX["overlay"] > LAYER_Z_INDEX["main"]
-        assert LAYER_Z_INDEX["corner_voice"] > LAYER_Z_INDEX["main"]
-        assert LAYER_Z_INDEX["audio"] > LAYER_Z_INDEX["overlay"]
-
-    def test_pip_default_scale_positive(self):
-        assert 0 < PIP_DEFAULT_SCALE < 1
-
-    def test_main_layer_roles(self):
-        assert "main" in MAIN_LAYER_ROLES
-        assert "broll" in MAIN_LAYER_ROLES
-        assert "background" in MAIN_LAYER_ROLES
-        assert "overlay" not in MAIN_LAYER_ROLES
-
-
-# ── resolve_layer_role ──────────────────────────────────────────────────────
-
-
-class TestResolveLayerRole:
-    def test_intro_maps_to_main(self):
-        assert resolve_layer_role("intro") == "main"
-
-    def test_outro_maps_to_main(self):
-        assert resolve_layer_role("outro") == "main"
-
-    def test_overlay_maps_to_overlay(self):
-        assert resolve_layer_role("overlay") == "overlay"
-
-    def test_corner_voice_maps_to_corner_voice(self):
-        assert resolve_layer_role("corner_voice") == "corner_voice"
-
-    def test_background_maps_to_background(self):
-        assert resolve_layer_role("background") == "background"
-
-    def test_b_roll_maps_to_broll(self):
-        assert resolve_layer_role("b_roll") == "broll"
-
-    def test_main_defaults_to_main(self):
-        assert resolve_layer_role("main") == "main"
-
-    def test_main_with_b_roll_role(self):
-        assert resolve_layer_role("main", {"role": "b_roll"}) == "broll"
-
-    def test_main_with_audio_role(self):
-        assert resolve_layer_role("main", {"role": "audio"}) == "audio"
-
-    def test_main_with_other_role_stays_main(self):
-        assert resolve_layer_role("main", {"role": "overlay"}) == "main"
-
-    def test_none_config(self):
-        assert resolve_layer_role("main", None) == "main"
-
-    def test_empty_config(self):
-        assert resolve_layer_role("main", {}) == "main"
-
-    def test_unknown_type_defaults_to_main(self):
-        assert resolve_layer_role("unknown_type") == "main"
-
-
-# ── get_layer_z_index ──────────────────────────────────────────────────────
-
-
-class TestGetLayerZIndex:
-    def test_known_roles(self):
-        for role, expected in LAYER_Z_INDEX.items():
-            assert get_layer_z_index(role) == expected
-
-    def test_unknown_role_returns_zero(self):
-        assert get_layer_z_index("nonexistent") == 0
-
-    def test_empty_string_returns_zero(self):
-        assert get_layer_z_index("") == 0
-
-
-# ── clip_effective_duration ────────────────────────────────────────────────
-
-
-class TestClipEffectiveDuration:
-    def test_explicit_duration_no_actual(self):
-        assert clip_effective_duration(5.0) == 5.0
-
-    def test_explicit_duration_with_shorter_actual(self):
-        assert clip_effective_duration(5.0, 3.0) == 3.0
-
-    def test_explicit_duration_with_longer_actual(self):
-        assert clip_effective_duration(5.0, 10.0) == 5.0
-
-    def test_zero_duration_uses_actual(self):
-        assert clip_effective_duration(0, 8.0) == 8.0
-
-    def test_negative_duration_uses_actual(self):
-        assert clip_effective_duration(-1.0, 8.0) == 8.0
-
-    def test_zero_duration_zero_actual(self):
-        assert clip_effective_duration(0, 0) == 0.0
-
-    def test_no_args_returns_zero(self):
-        assert clip_effective_duration(0) == 0.0
-
-    def test_equal_duration_and_actual(self):
-        assert clip_effective_duration(5.0, 5.0) == 5.0
-
-
-# ── clip_playback_speed ────────────────────────────────────────────────────
-
-
-class TestClipPlaybackSpeed:
-    def test_normal_speed(self):
-        assert clip_playback_speed(1.0) == 1.0
-
-    def test_fast_speed(self):
-        assert clip_playback_speed(2.0) == 2.0
-
-    def test_slow_speed(self):
-        assert clip_playback_speed(0.5) == 0.5
-
-    def test_zero_speed_defaults_to_one(self):
-        assert clip_playback_speed(0) == 1.0
-
-    def test_negative_speed_defaults_to_one(self):
-        assert clip_playback_speed(-1.0) == 1.0
-
-    def test_none_defaults_to_one(self):
-        assert clip_playback_speed(None) == 1.0
-
-    def test_string_defaults_to_one(self):
-        assert clip_playback_speed("fast") == 1.0
-
-    def test_int_speed(self):
-        assert clip_playback_speed(2) == 2.0
-
-
-# ── clip_adjusted_duration ─────────────────────────────────────────────────
-
-
-class TestClipAdjustedDuration:
-    def test_normal_speed_same_as_effective(self):
-        assert clip_adjusted_duration(5.0, 10.0, 1.0) == 5.0
-
-    def test_double_speed_half_duration(self):
-        assert clip_adjusted_duration(10.0, 10.0, 2.0) == pytest.approx(5.0)
-
-    def test_half_speed_double_duration(self):
-        assert clip_adjusted_duration(5.0, 10.0, 0.5) == pytest.approx(10.0)
-
-    def test_invalid_speed_uses_default(self):
-        assert clip_adjusted_duration(5.0, 10.0, 0) == 5.0
-
-    def test_zero_duration(self):
-        assert clip_adjusted_duration(0, 0, 1.0) == 0.0
-
-    def test_actual_duration_only(self):
-        assert clip_adjusted_duration(0, 8.0, 1.0) == 8.0
-
-    def test_actual_duration_only_with_speed(self):
-        assert clip_adjusted_duration(0, 8.0, 2.0) == pytest.approx(4.0)
-
-    def test_very_close_to_normal_speed(self):
-        # 1.0000001 应该被认为接近 1.0，不做除法
-        result = clip_adjusted_duration(5.0, 10.0, 1.0 + 1e-10)
-        assert result == 5.0
-
-
-# ── estimate_total_duration ────────────────────────────────────────────────
-
-
-class TestEstimateTotalDuration:
-    def test_empty_layers(self):
-        assert estimate_total_duration([]) == 0.0
-
-    def test_no_main_layer(self):
-        layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])]
-        assert estimate_total_duration(layers) == 0.0
-
-    def test_single_clip_main_layer(self):
-        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
-        assert estimate_total_duration(layers) == pytest.approx(5.0)
-
-    def test_multiple_clips_no_transition(self):
-        layers = [
-            FakeLayer(
-                role="main",
-                clips=[
-                    FakeClip(duration=3.0),
-                    FakeClip(duration=2.0),
-                    FakeClip(duration=5.0),
-                ],
-            )
-        ]
-        assert estimate_total_duration(layers) == pytest.approx(10.0)
-
-    def test_multiple_clips_with_transition(self):
-        layers = [
-            FakeLayer(
-                role="main",
-                clips=[
-                    FakeClip(duration=3.0),
-                    FakeClip(duration=2.0),
-                    FakeClip(duration=5.0),
-                ],
-            )
-        ]
-        # 3 + 2 + 5 - 2 * 0.5 = 9.0
-        assert estimate_total_duration(layers, transition_duration=0.5) == pytest.approx(9.0)
-
-    def test_prefers_main_over_broll(self):
-        layers = [
-            FakeLayer(role="broll", clips=[FakeClip(duration=10.0)]),
-            FakeLayer(role="main", clips=[FakeClip(duration=5.0)]),
-        ]
-        assert estimate_total_duration(layers) == pytest.approx(5.0)
-
-    def test_prefers_broll_over_background(self):
-        layers = [
-            FakeLayer(role="background", clips=[FakeClip(duration=10.0)]),
-            FakeLayer(role="broll", clips=[FakeClip(duration=5.0)]),
-        ]
-        assert estimate_total_duration(layers) == pytest.approx(5.0)
-
-    def test_main_layer_empty_clips(self):
-        layers = [FakeLayer(role="main", clips=[])]
-        assert estimate_total_duration(layers) == 0.0
-
-    def test_minimum_duration(self):
-        layers = [
-            FakeLayer(
-                role="main",
-                clips=[
-                    FakeClip(duration=0.01),
-                    FakeClip(duration=0.01),
-                ],
-            )
-        ]
-        result = estimate_total_duration(layers, transition_duration=0.5)
-        assert result >= 0.1
-
-    def test_with_playback_speed(self):
-        layers = [
-            FakeLayer(
-                role="main",
-                clips=[
-                    FakeClip(duration=10.0, playback_speed=2.0),
-                    FakeClip(duration=10.0, playback_speed=0.5),
-                ],
-            )
-        ]
-        # 5 + 20 = 25
-        assert estimate_total_duration(layers) == pytest.approx(25.0)
-
-
-# ── can_pass_through ──────────────────────────────────────────────────────
-
-
-class TestCanPassThrough:
-    def test_single_main_clip_no_effects(self):
-        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers) is True
-
-    def test_single_broll_clip(self):
-        layers = [FakeLayer(role="broll", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers) is True
-
-    def test_single_background_clip(self):
-        layers = [FakeLayer(role="background", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers) is True
-
-    def test_multiple_layers(self):
-        layers = [
-            FakeLayer(role="main", clips=[FakeClip(duration=5.0)]),
-            FakeLayer(role="overlay", clips=[FakeClip(duration=3.0)]),
-        ]
-        assert can_pass_through(layers) is False
-
-    def test_overlay_layer(self):
-        layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers) is False
-
-    def test_multiple_clips_in_layer(self):
-        layers = [
-            FakeLayer(
-                role="main",
-                clips=[
-                    FakeClip(duration=3.0),
-                    FakeClip(duration=2.0),
-                ],
-            )
-        ]
-        assert can_pass_through(layers) is False
-
-    def test_with_stickers(self):
-        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers, has_stickers=True) is False
-
-    def test_with_watermark(self):
-        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers, has_watermark=True) is False
-
-    def test_with_stickers_and_watermark(self):
-        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers, has_stickers=True, has_watermark=True) is False
-
-    def test_empty_layer_list(self):
-        assert can_pass_through([]) is False
-
-    def test_audio_layer_only(self):
-        layers = [FakeLayer(role="audio", clips=[FakeClip(duration=5.0)])]
-        assert can_pass_through(layers) is False

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"