xiaoxia-saas/tests/unit/test_editor_clips_random_start.py

"""测试编辑器 from-assets 端点：按模板segment创建片段 + 事务性替换 + 随机起始.

覆盖：
- 片段数量 = segment 数量（required_clips_count 被忽略）
- 素材不足时同一素材轮询切多个片段
- 随机 start_time + used_segments 去重
- 素材时长不足时 clip duration 缩短
- 素材时长全部为 0/缺失时抛 400「素材可切区间不足」；混合池中零时长素材被跳过
- 使用 replace_all_clips_transactional 原子性替换
- order 从 0 开始
- start_time=None 时抛出 400
- mark_clips_ready 在事务方法内部完成
"""

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
from fastapi import HTTPException

TEST_PLAN_ID = "plan-draft-001"
TEST_USER_ID = "user-001"

# 默认测试用 segments：4 个片段，每个 3~5 秒
DEFAULT_SEGMENTS = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)]


def _segments(count: int, dur_min: float = 3.0, dur_max: float = 5.0):
    return [(i, dur_min, dur_max) for i in range(count)]


def _patch_segments(segments=None):
    return patch(
        "app.api.routes.templates_editor.clips._get_template_segments",
        return_value=segments if segments is not None else DEFAULT_SEGMENTS,
    )


def _make_auth_user():
    auth = MagicMock()
    auth.user.id = TEST_USER_ID
    auth.user.email = "test@example.com"
    auth.user.display_name = "测试用户"
    auth.user_id = TEST_USER_ID
    return auth


def _make_mock_asset(asset_id, duration):
    asset = MagicMock()
    asset.id = asset_id
    asset.duration = duration
    return asset


def _make_plan_svc(replace_return_count=None):
    svc = MagicMock()
    # replace_all_clips_transactional 返回创建的片段数量
    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 _get_clips_data_from_call(mock_plan_svc):
    """从 replace_all_clips_transactional 的调用中获取 clips_data。"""
    assert mock_plan_svc.replace_all_clips_transactional.called, "replace_all_clips_transactional 未被调用"
    call_args = mock_plan_svc.replace_all_clips_transactional.call_args
    # call_args = ((plan_id, clips_data), kwargs)
    if len(call_args.args) >= 2:
        return call_args.args[1]
    return call_args.kwargs.get("clips_data", [])


@pytest.fixture(autouse=True)
def _mock_segment_tracker():
    """from-assets 现在会读/写素材 metadata 的片段区间记录，测试中 mock 掉避免依赖真实 DB。

    get_used_segments 返回空 dict（等价历史行为：无历史区间）；
    record/remove/reset 回调均无副作用。
    """
    with (
        patch(
            "app.api.routes.templates_editor.clips.get_used_segments",
            return_value={},
        ),
        patch(
            "app.api.routes.templates_editor.clips.record_used_segments",
            return_value=None,
        ),
        patch(
            "app.api.routes.templates_editor.clips.make_reuse_callback",
            return_value=lambda asset_id, clip_duration: None,
        ),
        patch(
            "app.api.routes.templates_editor.clips.remove_used_segment",
            return_value=False,
        ),
    ):
        yield


class TestEditorClipsBySegments:
    """测试按 segment 数量创建片段 + 素材轮询。"""

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_creates_clips_matching_segment_count(self, mock_storage):
        """4 个 segment 即使只有2个素材也创建4个片段，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

        mock_plan_svc = _make_plan_svc(replace_return_count=4)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0, "a2": 20.0}[aid]))

        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2)

        with _patch_segments(DEFAULT_SEGMENTS):
            result = create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), 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_from_call(mock_plan_svc)
        assert len(clips_data) == 4

        # 验证轮询分配：a1, a2, a1, a2
        assert clips_data[0]["asset_id"] == "a1"
        assert clips_data[1]["asset_id"] == "a2"
        assert clips_data[2]["asset_id"] == "a1"
        assert clips_data[3]["asset_id"] == "a2"

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_orders_start_at_zero(self, mock_storage):
        """片段 order 从 0 开始递增。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=3)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)

        with _patch_segments(_segments(3)):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert clips_data[0]["order"] == 0
        assert clips_data[1]["order"] == 1
        assert clips_data[2]["order"] == 2

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_uses_transactional_replace(self, mock_storage):
        """使用 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

        mock_plan_svc = _make_plan_svc(replace_return_count=2)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1"])

        with _patch_segments(_segments(2)):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), 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()
        # 不应调用单独的 delete 或 create
        assert not hasattr(mock_plan_svc, "create_clip") or not mock_plan_svc.create_clip.called

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

        mock_plan_svc = _make_plan_svc()
        mock_asset_repo = MagicMock()

        body = ClipsFromAssetsRequest(asset_ids=["a1"])

        with _patch_segments([]):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    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()


class TestEditorClipsDurationAndStartTime:
    """测试素材时长获取、clip duration 缩短、start_time 传入。"""

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_clip_duration_shortened_for_short_assets(self, mock_storage):
        """素材只有 3s 时 clip duration 缩短到不超过 3.0。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=1)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0))

        body = ClipsFromAssetsRequest(asset_ids=["short"])

        with _patch_segments(_segments(1, dur_min=5.0, dur_max=10.0)):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert clips_data[0]["duration"] <= 3.0

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_zero_duration_asset_raises_400(self, mock_storage):
        """所有素材时长均为 0 时轮询无可用素材，抛出 400「素材可切区间不足」。

        新轮询逻辑下零时长素材被跳过（而非立即报错）；全部素材都被跳过时
        返回 400，不创建无效片段。
        """
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc()
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("bad", 0.0))

        body = ClipsFromAssetsRequest(asset_ids=["bad"])

        with _patch_segments(_segments(1)):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )

        assert exc_info.value.status_code == 400
        assert "素材可切区间不足" in exc_info.value.detail

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_zero_duration_asset_skipped_in_mixed_pool(self, mock_storage):
        """素材池混合零时长与正常素材时，零时长素材被跳过、正常素材承担片段。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=2)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(
            side_effect=lambda aid: {
                "zero": _make_mock_asset("zero", 0.0),
                "good": _make_mock_asset("good", 30.0),
            }[aid]
        )

        body = ClipsFromAssetsRequest(asset_ids=["zero", "good"], required_clips_count=2)

        with (
            _patch_segments(_segments(2)),
            patch(
                "app.api.routes.templates_editor.clips._calc_random_start_time",
                side_effect=[5.0, 12.0],
            ),
        ):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert len(clips_data) == 2
        # 所有片段都分配给正常素材，零时长素材被跳过
        assert all(c["asset_id"] == "good" for c in clips_data)

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_missing_duration_asset_raises_400(self, mock_storage):
        """素材时长缺失（asset_repo.get 返回 None）时抛出 400。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc()
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=None)

        body = ClipsFromAssetsRequest(asset_ids=["missing"])

        with _patch_segments(_segments(1)):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )

        assert exc_info.value.status_code == 400

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_start_time_passed_to_create(self, mock_storage):
        """_calc_random_start_time 返回值被传入 clips_data。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=2)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2)

        with (
            _patch_segments(_segments(2)),
            patch(
                "app.api.routes.templates_editor.clips._calc_random_start_time",
                side_effect=[12.5, 18.0],
            ) as mock_calc,
        ):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        assert mock_calc.call_count == 2
        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert clips_data[0]["start_time"] == 12.5
        assert clips_data[1]["start_time"] == 18.0

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_asset_durations_deduped(self, mock_storage):
        """asset_ids 有重复时只查询一次素材时长。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=3)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"])

        with _patch_segments(_segments(3)):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        # 去重后只调用 1 次获取素材时长
        assert mock_asset_repo.get.call_count == 1

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_used_segments_maintained_across_clips(self, mock_storage):
        """同一素材切多个片段时，used_segments 应被维护。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=3)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)

        captured_used_segments = []

        def fake_calc(asset_id, clip_duration, asset_durations, used_segments, on_exhausted=None):
            captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()})
            return (len(captured_used_segments) - 1) * 5.0

        with (
            _patch_segments(_segments(3)),
            patch(
                "app.api.routes.templates_editor.clips._calc_random_start_time",
                side_effect=fake_calc,
            ),
        ):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        # 第一次没有已使用时间段
        assert captured_used_segments[0] == {}
        # 第二次有第一次的记录
        assert len(captured_used_segments[1]["a1"]) == 1
        # 第三次有前两次的记录
        assert len(captured_used_segments[2]["a1"]) == 2


class TestEditorClipsErrorHandling:
    """测试异常处理。"""

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_none_start_time_raises_400(self, mock_storage):
        """所有素材 calc 均返回 None（区间耗尽且复用被拒）→ 轮询失败抛 400。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc()
        mock_asset_repo = MagicMock()
        # 素材有 duration 但 calc 返回 None（模拟可用区间耗尽、复用被闸门拒绝）
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1"])

        with (
            _patch_segments(_segments(1)),
            patch(
                "app.api.routes.templates_editor.clips._calc_random_start_time",
                return_value=None,
            ),
        ):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )

        assert exc_info.value.status_code == 400
        assert "素材可切区间不足" in exc_info.value.detail
        # 复用被拒导致无起点时，不应创建任何片段
        assert not mock_plan_svc.replace_all_clips_transactional.called

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_transactional_replace_exception_propagates(self, mock_storage):
        """replace_all_clips_transactional 抛异常时应向上传播（事务已回滚）。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc()
        mock_plan_svc.replace_all_clips_transactional = MagicMock(side_effect=RuntimeError("DB connection lost"))
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1"])

        with _patch_segments(_segments(1)):
            with pytest.raises(RuntimeError, match="DB connection lost"):
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )


class TestReuseRatioGate:
    """素材区间耗尽后的受控复用与 15% 占比闸门（路由级）。"""

    @staticmethod
    def _make_calc_with_reuse(normal_starts, reused_durations):
        """构造模拟「区间耗尽后受控复用」的 _calc_random_start_time。

        normal_starts: list[float | None]，前 N 次调用返回的空闲起点；
            返回 None 表示随机找不到空闲 → 触发 on_exhausted 复用回调。
        回调被调用时返回复用区间（固定 0.0 起点），复用片段时长由路由累加到
        reused_durations；回调内部占比预判超 15% 时返回 None（calc 随之 None）。
        """
        calls = {"i": 0}

        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
            i = calls["i"]
            calls["i"] += 1
            if i < len(normal_starts) and normal_starts[i] is not None:
                return normal_starts[i]
            # 空闲耗尽 → 走受控复用回调（回调返回 (start, end) 元组，calc 取起点）
            if on_exhausted is not None:
                result = on_exhausted(asset_id, clip_duration)
                return result[0] if result else None
            return None

        return fake_calc, calls

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_reused_clip_ratio_within_threshold(self, mock_storage):
        """素材 60s、片段 5s：前 12 个用空闲区间，第 13 个复用，
        复用占比 5/(12*5+5)=7.7% ≤ 15%，正常创建 13 个片段。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=13)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0))
        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=13)

        reused: dict = {}

        def reuse_cb(aid, dur):
            # 模拟真实回调：返回复用区间前记录复用时长
            reused[aid] = reused.get(aid, 0.0) + dur
            return (0.0, dur)

        # 前 12 次分配空闲起点；第 13 次 calc 直接走回调（normal_starts 越界 → None → 回调）
        normal_starts = [float(i * 5) for i in range(12)]
        fake_calc, _ = self._make_calc_with_reuse(normal_starts, reused)
        with (
            _patch_segments(_segments(13, dur_min=5.0, dur_max=5.0)),
            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
            patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=reuse_cb),
        ):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )
        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert len(clips_data) == 13
        # 1 个复用片段，占比 1/13 ≈ 7.7% ≤ 15%
        assert reused.get("a1", 0.0) == 5.0

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_reuse_ratio_exceeded_returns_400(self, mock_storage):
        """复用占比将超 15% 时回调拒绝复用 → 无可用素材 → 400「素材可切区间不足」。

        60s 素材、5s 片段：前 12 个空闲、随后复用占比累计；当 (reused+d)/(assigned+d)
        超过 15% 时回调返回 None，calc 返回 None，轮询无素材 → 400。
        """
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=0)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0))
        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=20)

        # 模拟真实回调：累计复用时长，预判超 15% 拒绝
        reused: dict = {}
        assigned: dict = {}

        def fake_reuse_cb(aid, clip_duration):
            a = assigned.get(aid, 0.0)
            r = reused.get(aid, 0.0)
            if a > 0 and (r + clip_duration) / (a + clip_duration) > 0.15:
                return None  # 占比闸门拒绝
            reused[aid] = r + clip_duration
            return (0.0, clip_duration)

        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
            a = assigned.get(asset_id, 0.0)
            # 前 12 个片段（60s/5s）有空闲区间
            if a < 60.0:
                start = a
                assigned[asset_id] = a + clip_duration
                return start
            # 之后空闲耗尽 → 复用
            if on_exhausted is not None:
                result = on_exhausted(asset_id, clip_duration)
                if result is not None:
                    assigned[asset_id] = assigned.get(asset_id, 0.0) + clip_duration
                return result[0] if result else None
            return None

        with (
            _patch_segments(_segments(20, dur_min=5.0, dur_max=5.0)),
            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
            patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=fake_reuse_cb),
        ):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )

        assert exc_info.value.status_code == 400
        assert "素材可切区间不足" in exc_info.value.detail
        # 闸门在复用占比达上限时拒绝：60s 空闲 + 至多 ~15% 复用
        assert reused.get("a1", 0.0) <= 12.0  # 10.0 或 15.0 以内，不会无限复用
        # 未创建任何片段（整批失败）
        assert not mock_plan_svc.replace_all_clips_transactional.called

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_calc_none_falls_through_to_next_asset(self, mock_storage):
        """一个素材区间耗尽且复用被拒（calc 返回 None）时，轮询到下一个可用素材。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=2)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(
            side_effect=lambda aid: {
                "exhausted": _make_mock_asset("exhausted", 60.0),
                "fresh": _make_mock_asset("fresh", 60.0),
            }[aid]
        )
        body = ClipsFromAssetsRequest(asset_ids=["exhausted", "fresh"], required_clips_count=2)

        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
            if asset_id == "exhausted":
                # 空闲耗尽 + 回调拒绝 → None
                return on_exhausted(asset_id, clip_duration) if on_exhausted else None
            return 8.0  # 新鲜素材正常返回

        with (
            _patch_segments(_segments(2, dur_min=5.0, dur_max=5.0)),
            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
            patch(
                "app.api.routes.templates_editor.clips.make_reuse_callback",
                return_value=lambda aid, d: None,  # 复用始终被拒
            ),
        ):
            create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )
        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert len(clips_data) == 2
        # 耗尽素材被跳过，两个片段都分配给新鲜素材
        assert all(c["asset_id"] == "fresh" for c in clips_data)


# ── P0 回归：from-assets 对 undefined/null/空串 asset_ids 容错 ──────────────
# 线上事故：前端 smart-match 拿到 {asset: {...}} 包装层后 items.map(a=>a.id)
# 全为 undefined，POST /clips/from-assets 携带 null → 422，自动模式预览断裂。


class TestClipsFromAssetsInvalidIds:
    def test_schema_filters_null_and_empty_ids(self):
        """请求 schema 在 pre 阶段剔除 null/空串/空白 id，不触发 422。"""
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        body = ClipsFromAssetsRequest(asset_ids=["a1", None, "", "  ", "a2"])  # type: ignore[list-item]
        assert body.asset_ids == ["a1", "a2"]

    def test_schema_all_invalid_raises(self):
        """全部为非法 id 时 min_length=1 兜底报校验错误（前端得到 422 而非脏数据）。"""
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            ClipsFromAssetsRequest(asset_ids=[None, "", "  "])  # type: ignore[list-item]

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_route_filters_invalid_ids_and_uses_valid(self, mock_storage):
        """路由层二次兜底：混有 null/空串时只用合法 id 正常创建片段。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_plan_svc = _make_plan_svc(replace_return_count=2)
        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0))

        body = ClipsFromAssetsRequest(asset_ids=["a1", None, "", "a2"])  # type: ignore[list-item]

        with _patch_segments(_segments(2, dur_min=3.0, dur_max=5.0)):
            result = create_clips_from_assets_editor(
                template_id="tpl-001",
                body=body,
                background_tasks=MagicMock(),
                plan_id=TEST_PLAN_ID,
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        assert result.created_count == 2
        clips_data = _get_clips_data_from_call(mock_plan_svc)
        assert {c["asset_id"] for c in clips_data} == {"a1", "a2"}

    @patch("app.api.routes.templates_editor.clips.get_storage_service")
    def test_route_all_empty_ids_raises_400(self, mock_storage):
        """schema 被绕过直接调路由、且 id 全非法时，路由 400 而非 500/422。"""
        from app.api.routes.templates_editor.clips import (
            create_clips_from_assets_editor,
        )

        mock_plan_svc = _make_plan_svc()
        mock_asset_repo = MagicMock()

        body = MagicMock()
        body.asset_ids = [None, "", "  "]

        with _patch_segments(DEFAULT_SEGMENTS):
            with pytest.raises(HTTPException) as exc_info:
                create_clips_from_assets_editor(
                    template_id="tpl-001",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=TEST_PLAN_ID,
                    services=(MagicMock(), 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()