xiaoxia-saas/tests/unit/test_reselect_plan_variant_1743.py

"""#1743 EditPlanService.reselect_plan_for_variant 服务层测试。

与 clone_plan_for_variant（只重算起点、素材/顺序不变）不同，reselect 完整重跑
单视频选片：素材池 shuffle + main 片段顺序洗牌 + 起点重选 + 批次 20% 重叠避让。

覆盖：
- 独立 plan：新 plan_id 与源不同、命名带后缀、模板/config 复制
- 新片段经 replace_all_clips_transactional 落库，素材/起点与源 plan 存在差异
- 源 plan 区间作为批次避让初始对象；record_used_segments 随新片段写回
- 源 plan 无片段 → ValueError（不创建同源变体）
"""

from __future__ import annotations

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

import pytest

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

REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO_ROOT / "apps" / "api"))
sys.path.insert(0, str(Path(__file__).resolve().parent))  # tests/unit

from test_edit_plan_service import StubEditPlanClipRepository, StubEditPlanRepository, _make_service  # noqa: E402

from packages.domain.edit_plan_clip import EditPlanClip  # noqa: E402


@pytest.fixture
def svc_with_source():
    """带源 plan（4 个 main 片段，素材 a1/a2/a3/a4）+ 2 个批次候选素材的 service。"""
    svc = _make_service()
    svc._clip_repo.session = MagicMock()

    source = svc.create_plan(template_id="tpl-001", name="9/6-草稿", total_duration=20.0, config={"title": "源配置"})
    for i, aid in enumerate(["a1", "a2", "a3", "a4"]):
        clip = EditPlanClip.create(
            plan_id=source.id,
            clip_type="main",
            order=i,
            asset_id=aid,
            start_time=float(i * 5),
            duration=5.0,
            text_content=f"文案{i}",
        )
        svc._clip_repo.create(clip)
    return svc, source


def _patch_deps(svc, durations):
    """统一 patch reselect 的 DB/素材/历史区间依赖。返回 (patches, mock_replace)。"""
    asset_models = []
    for aid, dur in durations.items():
        m = MagicMock(id=aid)
        m.duration = dur
        m.metadata = None  # extract_scene_points_from_metadata(None) → 无场景点
        asset_models.append(m)

    mock_replace = patch.object(svc, "replace_all_clips_transactional", return_value=4)
    patches = [
        patch("app.services.edit_plan_service.get_used_segments", return_value={}),
        patch("app.services.edit_plan_service.record_used_segments", return_value=None),
        patch("packages.domain.plan_generator_utils.extract_scene_points_from_metadata", return_value=[]),
        patch("packages.adapters.sqlalchemy_impl.models.AssetModel", create=True),
        mock_replace,
    ]
    started = []
    for p in patches:
        started.append(p.start())
    # started[-1] 是 replace_all 的 MagicMock
    mock_replace_obj = started[-1]
    # db.query(AssetModel).filter(...).all() → 带 duration 的素材 mock
    svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = asset_models
    return patches, mock_replace_obj


class TestReselectPlanForVariant:
    def test_creates_independent_plan_with_different_clips(self, svc_with_source):
        """reselect 产出新 plan（id/名称不同），片段素材或起点与源 plan 存在差异。"""
        import random

        svc, source = svc_with_source
        durations = {"a1": 300.0, "a2": 300.0, "a3": 300.0, "a4": 300.0, "a5": 300.0, "a6": 300.0}
        patches, mock_replace = _patch_deps(svc, durations)
        try:
            new_plan = svc.reselect_plan_for_variant(
                source.id,
                ["a5", "a6"],  # 批次素材并入素材池
                created_by_user_id="u-1",
                name_suffix="批量2",
                rng=random.Random(42),
            )
        finally:
            for p in reversed(patches):
                p.stop()

        # 新 plan 独立、归属/模板/config 复制
        assert new_plan.id != source.id
        assert "批量2" in new_plan.name
        assert new_plan.template_id == "tpl-001"
        assert new_plan.config == {"title": "源配置"}
        assert new_plan.created_by_user_id == "u-1"

        # 落库片段数 == 源片段数，且 order 对齐
        clips_data = mock_replace.call_args.args[1]
        assert len(clips_data) == 4
        assert [c["order"] for c in clips_data] == [0, 1, 2, 3]

        # 与源 plan 对比：素材序列或起点必须存在差异（降重核心——不是克隆）
        source_pairs = [(c.asset_id, round(float(c.start_time), 2)) for c in svc._clip_repo.list_by_plan(source.id)]
        new_pairs = [(c["asset_id"], round(float(c["start_time"]), 2)) for c in clips_data]
        assert new_pairs != source_pairs, f"reselect 片段应与源 plan 不同，实际相同: {new_pairs}"
        # 素材全部来自素材池（源 a1-a4 ∪ 批次 a5-a6）
        for c in clips_data:
            assert c["asset_id"] in durations

    def test_record_used_segments_called_per_clip(self, svc_with_source):
        """每个新片段区间调用 record_used_segments 写回（跨变体/跨任务避让链路）。"""
        import random

        svc, source = svc_with_source
        durations = {"a1": 300.0, "a2": 300.0, "a3": 300.0, "a4": 300.0, "a5": 300.0}

        asset_models = []
        for aid in durations:
            m = MagicMock(id=aid)
            m.duration = durations[aid]
            m.metadata = None
            asset_models.append(m)
        svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = asset_models

        started = [
            patch("app.services.edit_plan_service.get_used_segments", return_value={}).start(),
            patch("packages.domain.plan_generator_utils.extract_scene_points_from_metadata", return_value=[]).start(),
            patch("packages.adapters.sqlalchemy_impl.models.AssetModel", create=True).start(),
            patch.object(svc, "replace_all_clips_transactional", return_value=4).start(),
        ]
        mock_record = patch("app.services.edit_plan_service.record_used_segments", return_value=None).start()
        try:
            svc.reselect_plan_for_variant(
                source.id, ["a5"], created_by_user_id="u-1", name_suffix="批量2", rng=random.Random(5)
            )
        finally:
            patch.stopall()

        assert mock_record.call_count == 4, "4 个片段应各写一次 used_segment"
        for call in mock_record.call_args_list:
            args = call.args
            assert args[1] in durations, f"asset_id {args[1]} 不在素材池"  # asset_id
            assert args[3] > args[2], "区间 end 应大于 start"  # end > start
            assert args[4], "new plan_id 应非空"

    def test_source_clips_seed_batch_avoidance(self, svc_with_source):
        """源 plan 片段区间进入批次避让集：与源完全同区间的起点重叠率应超限被避开。"""
        import random

        svc, source = svc_with_source
        # 素材池只有源素材（极端小池），时长充足
        durations = {"a1": 600.0, "a2": 600.0, "a3": 600.0, "a4": 600.0}
        patches, mock_replace = _patch_deps(svc, durations)
        try:
            new_plan = svc.reselect_plan_for_variant(
                source.id, [], created_by_user_id="u-1", name_suffix="批量2", rng=random.Random(99)
            )
        finally:
            for p in reversed(patches):
                p.stop()

        clips_data = mock_replace.call_args.args[1]
        source_clips = svc._clip_repo.list_by_plan(source.id)
        source_by_asset = {}
        for c in source_clips:
            source_by_asset.setdefault(c.asset_id, []).append(
                (float(c.start_time), float(c.start_time) + float(c.duration))
            )

        # 同素材新片段与源区间的重叠占比均 ≤20%
        from packages.domain.variant_plan_selector import _clip_overlap_ratio

        for c in clips_data:
            ratio = _clip_overlap_ratio(c["asset_id"], float(c["start_time"]), float(c["duration"]), source_by_asset)
            assert (
                ratio <= 0.20 + 1e-6
            ), f"变体片段与源 plan 同素材区间重叠超限: asset={c['asset_id']} ratio={ratio:.2%}"
        assert new_plan.id != source.id

    def test_source_plan_without_clips_raises(self):
        """源 plan 无片段 → ValueError（明确报错，不产出同源变体）。"""
        svc = _make_service()
        svc._clip_repo.session = MagicMock()
        empty = svc.create_plan(template_id="tpl-x", name="空计划")

        with pytest.raises(ValueError, match="源 plan 无片段"):
            svc.reselect_plan_for_variant(empty.id, ["a1"], created_by_user_id="u-1", name_suffix="变体")

    def test_missing_source_plan_raises(self):
        """源 plan 不存在 → get_plan_or_raise 抛错。"""
        svc = _make_service()
        svc._clip_repo.session = MagicMock()
        with pytest.raises((ValueError, KeyError, LookupError)):  # get_plan_or_raise 抛错
            svc.reselect_plan_for_variant("not-exist-plan", ["a1"], created_by_user_id="u-1", name_suffix="变体")