xiaoxia-saas/tests/unit/test_mediakit_smart_clips.py

"""测试 MediaKit 智能选片 + from-assets 按模板 segment 创建片段。

覆盖：
- _recommended_time_conflicts 冲突检测
- _get_mediakit_recommendations 解析与降级
- _get_template_segments 查询模板片段配置
- from-assets 端点：按模板 segment 数量和时长创建片段
- from-assets 端点：事务性原子替换
- from-assets 端点：素材轮询分配
- from-assets 端点：MediaKit 推荐时间首片段使用
- from-assets 端点：同一素材多片段时后续用随机
- from-assets 端点：无 segment 配置时报错
- from-assets 端点：素材时长为0时报400
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

import pytest

# ── _recommended_time_conflicts 单元测试 ─────────────────────────────────────


class TestRecommendedTimeConflicts:
    """测试推荐时间与已使用时间段的冲突检测。"""

    def test_no_conflict_when_empty(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        assert _recommended_time_conflicts(5.0, 5.0, []) is False

    def test_no_conflict_when_before(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        assert _recommended_time_conflicts(5.0, 5.0, [(15.0, 20.0)]) is False

    def test_no_conflict_when_after(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 10.0)]) is False

    def test_conflict_overlap_start(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        # 推荐 [5, 10]，已用 [0, 7]
        assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 7.0)]) is True

    def test_conflict_overlap_end(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        # 推荐 [5, 10]，已用 [8, 15]
        assert _recommended_time_conflicts(5.0, 5.0, [(8.0, 15.0)]) is True

    def test_conflict_contained(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        # 推荐 [5, 10]，已用 [0, 20]
        assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 20.0)]) is True

    def test_conflict_exact_boundary_no_overlap(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        # 新语义：默认 0.3s 边缘间隙扩边，推荐 [10, 15] 与已用 [0, 10] 首尾相接
        # 落在扩边范围内 → 判为冲突（避免观感重复）
        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is True
        # 显式 edge_gap=0 时退回纯区间重叠判定：相接不算重叠
        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)], edge_gap=0.0) is False
        # 间隙大于边缘间隙（0.5 > 0.3）→ 不冲突
        assert _recommended_time_conflicts(10.5, 5.0, [(0.0, 10.0)]) is False

    def test_conflict_multiple_used(self):
        from app.api.routes.templates_editor.clips import _recommended_time_conflicts

        used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
        # 推荐 [6, 11] 与 [10, 15] 冲突
        assert _recommended_time_conflicts(6.0, 5.0, used) is True
        # 推荐 [15, 20] 与 [10, 15] 首尾相接：0.3s 扩边内 → 冲突
        assert _recommended_time_conflicts(15.0, 5.0, used) is True
        # 空闲段 [5.3, 9.7] 长 4.4s：推荐 [5.5, 9.5]（dur=4）与三区间扩边均不接触
        assert _recommended_time_conflicts(5.5, 4.0, used) is False


# ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────


class TestGetMediakitRecommendations:
    """测试 MediaKit 推荐结果解析和降级。"""

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_returns_parsed_recommendations(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client.analyze_videos.return_value = [
            '[{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "动作场景"}]'
        ]
        mock_client_fn.return_value = mock_client

        mock_storage_svc = MagicMock()
        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
        mock_storage.return_value = mock_storage_svc

        asset = MagicMock()
        asset.storage_key = "v.mp4"
        asset.mime_type = "video/mp4"

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = asset

        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
        assert result == {"a1": 12.5}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_client_not_available_returns_empty(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = False
        mock_client_fn.return_value = mock_client

        result = _get_mediakit_recommendations(["a1"], MagicMock())
        assert result == {}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_empty_contents_returns_empty(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client.analyze_videos.return_value = []
        mock_client_fn.return_value = mock_client

        mock_storage_svc = MagicMock()
        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
        mock_storage.return_value = mock_storage_svc

        asset = MagicMock()
        asset.storage_key = "v.mp4"
        asset.mime_type = "video/mp4"
        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = asset

        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
        assert result == {}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_unparseable_response_returns_empty(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client.analyze_videos.return_value = ["这是一段自然语言描述，没有JSON"]
        mock_client_fn.return_value = mock_client

        mock_storage_svc = MagicMock()
        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
        mock_storage.return_value = mock_storage_svc

        asset = MagicMock()
        asset.storage_key = "v.mp4"
        asset.mime_type = "video/mp4"
        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = asset

        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
        assert result == {}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_storage_failure_returns_empty(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client_fn.return_value = mock_client
        mock_storage.side_effect = RuntimeError("storage unavailable")

        result = _get_mediakit_recommendations(["a1"], MagicMock())
        assert result == {}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_exception_returns_empty(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client_fn.side_effect = RuntimeError("unexpected error")

        result = _get_mediakit_recommendations(["a1"], MagicMock())
        assert result == {}

    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_skips_non_video_assets(self, mock_storage, mock_client_fn):
        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client_fn.return_value = mock_client

        mock_asset_repo = MagicMock()
        mock_asset = MagicMock()
        mock_asset.storage_key = "images/test.jpg"
        mock_asset.mime_type = "image/jpeg"
        mock_asset_repo.get.return_value = mock_asset

        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
        assert result == {}
        mock_client.analyze_videos.assert_not_called()


# ── _get_template_segments 单元测试 ─────────────────────────────────────────


class TestGetTemplateSegments:
    """测试模板片段配置查询。"""

    def test_returns_segments_from_new_template_system(self):
        """新模板系统（clip_configs）有数据时优先使用。"""
        from app.api.routes.templates_editor.clips import _get_template_segments

        mock_tpl_svc = MagicMock()
        cc1 = MagicMock()
        cc1.order = 0
        cc1.min_duration = 3.0
        cc1.max_duration = 5.0
        cc2 = MagicMock()
        cc2.order = 1
        cc2.min_duration = 4.0
        cc2.max_duration = 8.0
        mock_tpl_svc.list_clip_configs.return_value = [cc2, cc1]  # 乱序返回

        result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
        assert len(result) == 2
        assert result[0] == (0, 3.0, 5.0)
        assert result[1] == (1, 4.0, 8.0)

    def test_falls_back_to_old_template_segments(self):
        """新模板系统无数据时回退到旧系统。"""
        from app.api.routes.templates_editor.clips import _get_template_segments

        mock_tpl_svc = MagicMock()
        mock_tpl_svc.list_clip_configs.return_value = []

        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
            mock_repo = MagicMock()
            seg1 = MagicMock()
            seg1.segment_order = 0
            seg1.duration_min = 2.0
            seg1.duration_max = 4.0
            mock_repo.list_segments.return_value = [seg1]
            MockRepo.return_value = mock_repo

            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
            assert len(result) == 1
            assert result[0] == (0, 2.0, 4.0)

    def test_returns_empty_when_no_segments(self):
        """两套系统都没有片段配置时返回空列表。"""
        from app.api.routes.templates_editor.clips import _get_template_segments

        mock_tpl_svc = MagicMock()
        mock_tpl_svc.list_clip_configs.return_value = []

        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
            mock_repo = MagicMock()
            mock_repo.list_segments.return_value = []
            MockRepo.return_value = mock_repo

            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
            assert result == []

    def test_new_system_exception_falls_back(self):
        """新模板系统异常时回退到旧系统。"""
        from app.api.routes.templates_editor.clips import _get_template_segments

        mock_tpl_svc = MagicMock()
        mock_tpl_svc.list_clip_configs.side_effect = RuntimeError("db error")

        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
            mock_repo = MagicMock()
            seg = MagicMock()
            seg.segment_order = 0
            seg.duration_min = 1.0
            seg.duration_max = 3.0
            mock_repo.list_segments.return_value = [seg]
            MockRepo.return_value = mock_repo

            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
            assert len(result) == 1


# ── from-assets 端点集成测试 ────────────────────────────────────────────────


def _make_auth_user():
    auth = MagicMock()
    auth.user.id = "user-001"
    auth.user.email = "test@example.com"
    auth.user.display_name = "test"
    auth.user_id = "user-001"
    return auth


def _make_clip_config(order, min_dur, max_dur):
    cc = MagicMock()
    cc.order = order
    cc.min_duration = min_dur
    cc.max_duration = max_dur
    return cc


def _make_plan_svc(replace_return_count=None):
    svc = MagicMock()
    svc.get_plan_or_raise = MagicMock()
    if replace_return_count is not None:
        svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count)
    else:
        svc.replace_all_clips_transactional = MagicMock(return_value=0)
    return svc


def _make_tpl_svc_with_segments(segments):
    """segments: list of (order, min_dur, max_dur)"""
    svc = MagicMock()
    clip_configs = [_make_clip_config(o, mn, mx) for o, mn, mx in segments]
    svc.list_clip_configs.return_value = clip_configs
    return svc


def _make_rich_asset(asset_id, duration, storage_key="v.mp4", mime="video/mp4"):
    asset = MagicMock()
    asset.id = asset_id
    asset.duration = duration
    asset.storage_key = storage_key
    asset.mime_type = mime
    return asset


def _get_clips_data(mock_plan_svc):
    """从 replace_all_clips_transactional 调用中提取 clips_data。"""
    call_args = mock_plan_svc.replace_all_clips_transactional.call_args
    if len(call_args.args) >= 2:
        return call_args.args[1]
    return call_args.kwargs.get("clips_data", [])


class TestFromAssetsByTemplateSegments:
    """测试 from-assets 按模板 segment 创建片段（V2 事务性替换）。"""

    def test_creates_clips_matching_segment_count(self):
        """片段数量 = segment 数量，忽略 required_clips_count。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0), (2, 2.0, 6.0), (3, 5.0, 10.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=4)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)

        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
        result = create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        assert result.created_count == 4
        clips_data = _get_clips_data(mock_plan_svc)
        assert len(clips_data) == 4

    def test_uses_transactional_replace(self):
        """使用 replace_all_clips_transactional 而不是分别 delete + create。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=2)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)

        body = ClipsFromAssetsRequest(asset_ids=["a1"])
        create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        mock_plan_svc.replace_all_clips_transactional.assert_called_once()
        assert not mock_plan_svc.delete_all_clips.called
        assert not mock_plan_svc.create_clip.called

    def test_duration_within_segment_range(self):
        """每个片段时长在 segment 的 min~max 范围内。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=2)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)

        body = ClipsFromAssetsRequest(asset_ids=["a1"])
        create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        clips_data = _get_clips_data(mock_plan_svc)
        assert 3.0 <= clips_data[0]["duration"] <= 5.0
        assert 4.0 <= clips_data[1]["duration"] <= 8.0

    def test_assets_round_robin_assignment(self):
        """素材按片段顺序轮询分配。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=4)

        mock_asset_repo = MagicMock()

        def get_asset(aid):
            return _make_rich_asset(aid, 60.0)

        mock_asset_repo.get.side_effect = get_asset

        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"])
        create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        clips_data = _get_clips_data(mock_plan_svc)
        asset_ids = [c["asset_id"] for c in clips_data]
        assert asset_ids == ["a1", "a2", "a1", "a2"]

    def test_orders_start_from_zero(self):
        """片段 order 从 0 开始递增。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=3)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)

        body = ClipsFromAssetsRequest(asset_ids=["a1"])
        create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        clips_data = _get_clips_data(mock_plan_svc)
        orders = [c["order"] for c in clips_data]
        assert orders == [0, 1, 2]

    def test_no_segments_raises_400(self):
        """模板没有 segment 配置时返回 400。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
        from fastapi import HTTPException

        mock_tpl_svc = MagicMock()
        mock_tpl_svc.list_clip_configs.return_value = []
        mock_plan_svc = _make_plan_svc()

        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
            mock_repo = MagicMock()
            mock_repo.list_segments.return_value = []
            MockRepo.return_value = mock_repo

            body = ClipsFromAssetsRequest(asset_ids=["a1"])
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tmpl-1",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id="plan-1",
                    services=(mock_tpl_svc, mock_plan_svc),
                    asset_repo=MagicMock(),
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )
            assert exc_info.value.status_code == 400

        mock_plan_svc.replace_all_clips_transactional.assert_not_called()

    def test_duration_capped_by_asset_duration(self):
        """素材时长不足时 clip duration 被缩短。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        segments = [(0, 10.0, 20.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc(replace_return_count=1)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("a1", 5.0)

        body = ClipsFromAssetsRequest(asset_ids=["a1"])
        create_clips_from_assets_editor(
            template_id="tmpl-1",
            body=body,
            background_tasks=MagicMock(),
            plan_id="plan-1",
            services=(mock_tpl_svc, mock_plan_svc),
            asset_repo=mock_asset_repo,
            db=MagicMock(),
            current_user=_make_auth_user(),
        )

        clips_data = _get_clips_data(mock_plan_svc)
        assert clips_data[0]["duration"] <= 5.0

    def test_zero_duration_asset_raises_400(self):
        """素材时长为 0 时抛出 400。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
        from fastapi import HTTPException

        segments = [(0, 3.0, 5.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc()

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = _make_rich_asset("bad", 0.0)

        body = ClipsFromAssetsRequest(asset_ids=["bad"])
        with pytest.raises(HTTPException) as exc_info:
            create_clips_from_assets_editor(
                template_id="tmpl-1",
                body=body,
                background_tasks=MagicMock(),
                plan_id="plan-1",
                services=(mock_tpl_svc, mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        assert exc_info.value.status_code == 400
        mock_plan_svc.replace_all_clips_transactional.assert_not_called()

    def test_empty_asset_ids_raises_400(self):
        """asset_ids 为空列表时返回 400（defense-in-depth，schema 层也有 min_length=1）。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from fastapi import HTTPException

        segments = [(0, 3.0, 5.0)]
        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
        mock_plan_svc = _make_plan_svc()

        # 用 MagicMock 模拟 body，绕过 Pydantic schema 的 min_length 校验
        mock_body = MagicMock()
        mock_body.asset_ids = []
        mock_body.required_clips_count = None

        with pytest.raises(HTTPException) as exc_info:
            create_clips_from_assets_editor(
                template_id="tmpl-1",
                body=mock_body,
                background_tasks=MagicMock(),
                plan_id="plan-1",
                services=(mock_tpl_svc, mock_plan_svc),
                asset_repo=MagicMock(),
                db=MagicMock(),
                current_user=_make_auth_user(),
            )
        assert exc_info.value.status_code == 400
        assert "素材" in exc_info.value.detail
        mock_plan_svc.replace_all_clips_transactional.assert_not_called()


# ── _safe_segment_duration 单元测试 ─────────────────────────────────────────


class TestSafeSegmentDuration:
    """测试片段时长安全转换。"""

    def test_normal_float(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration(3.5, 1.0) == 3.5

    def test_none_returns_default(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration(None, 5.0) == 5.0

    def test_string_number(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration("4.2", 1.0) == 4.2

    def test_invalid_string_returns_default(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration("abc", 5.0) == 5.0

    def test_negative_returns_default(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration(-1.0, 5.0) == 5.0

    def test_zero_returns_default(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration(0, 5.0) == 5.0

    def test_integer_value(self):
        from app.api.routes.templates_editor.clips import _safe_segment_duration

        assert _safe_segment_duration(10, 1.0) == 10.0