xiaoxia-saas/tests/unit/test_clone_plan_variant.py

"""clone_plan_for_variant 单元测试（Task G 验收项：批量 N 条视频片段独立）。

验证：
- 同一源 plan 克隆 3 次产出 3 个不同 plan_id，各自片段起点不同
- 源 plan 的片段不被修改
- 模板/config/时长结构被复制
- 复用占比闸门触发时保留原起点（不重复抽取）
- 源 plan 无片段时抛出 ValueError
"""

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")

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
sys.path.insert(0, str(Path(__file__).resolve().parent))  # tests/unit，便于复用同目录 stub

# 复用 test_edit_plan_service 里的内存 stub 仓储
from test_edit_plan_service import (  # noqa: E402
    StubEditPlanClipRepository,
    StubEditPlanRepository,
    _make_service,
)

from packages.domain.edit_plan_clip import EditPlanClip


@pytest.fixture
def svc_with_source():
    """构造带源 plan + 3 个片段的 service（stub 仓储）。"""
    svc = _make_service()
    # clone 用 self._clip_repo.session 拿 db；stub 无 session，补一个 MagicMock
    svc._clip_repo.session = MagicMock()

    source = svc.create_plan(template_id="tpl-001", name="源计划", total_duration=15.0)

    for i in range(3):
        clip = EditPlanClip.create(
            plan_id=source.id,
            clip_type="main",
            order=i,
            asset_id=f"a{i % 2 + 1}",  # a1, a2, a1
            start_time=float(i * 5),
            duration=5.0,
        )
        svc._clip_repo.create(clip)
    return svc, source


def _clone_with_fake_calc(svc, source, starts, *, used=None):
    """用受控的 calc 起点列表执行一次克隆。

    starts: 每次 _calc_random_start_time 返回的起点（按片段顺序）。
    返回 (new_plan, replace_all 调用的 clips_data, calc 调用记录)。
    """
    calc_calls: list[dict] = []

    def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
        idx = len(calc_calls)
        calc_calls.append({"asset_id": asset_id, "clip_duration": clip_duration, "on_exhausted": on_exhausted})
        return starts[idx]

    with (
        patch(
            "app.services.edit_plan_service.get_used_segments",
            return_value=used or {},
        ),
        patch(
            "app.services.edit_plan_service.make_reuse_callback",
            return_value=lambda aid, d: None,
        ),
        patch(
            "app.services.edit_plan_service.record_used_segments",
            return_value=None,
        ),
        patch(
            "packages.domain.plan_generator_utils._calc_random_start_time",
            side_effect=fake_calc,
        ),
        patch.object(svc, "replace_all_clips_transactional", return_value=3) as mock_replace,
        patch(
            "packages.adapters.sqlalchemy_impl.models.AssetModel",
            create=True,
        ) as mock_asset_model,
    ):
        # db.query(AssetModel).filter(...).all() → 返回带 duration 的 mock 素材
        m1 = MagicMock(id="a1")
        m1.duration = 60.0
        m2 = MagicMock(id="a2")
        m2.duration = 60.0
        svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = [m1, m2]
        new_plan = svc.clone_plan_for_variant(source.id, created_by_user_id="u1", name_suffix="变体")
        clips_data = mock_replace.call_args.args[1]
    return new_plan, clips_data, calc_calls


class TestClonePlanForVariant:
    def test_three_clones_produce_distinct_plans_and_starts(self, svc_with_source):
        """克隆 3 次：3 个不同 plan_id，片段起点互不相同（Task G 验收）。"""
        svc, source = svc_with_source
        start_sets = [
            [10.0, 20.0, 30.0],
            [11.0, 21.0, 31.0],
            [12.0, 22.0, 32.0],
        ]
        plans = []
        all_clips = []
        for starts in start_sets:
            new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, starts)
            plans.append(new_plan)
            all_clips.append(clips_data)

        # 3 个不同 plan_id，且都不等于源 plan
        plan_ids = {p.id for p in plans}
        assert len(plan_ids) == 3
        assert source.id not in plan_ids

        # 每次克隆的起点各自不同
        for clips_data, starts in zip(all_clips, start_sets, strict=True):
            assert [c["start_time"] for c in clips_data] == starts

        # 三次克隆的起点集合互不相同
        assert {tuple(c["start_time"] for c in clips) for clips in all_clips} == {
            (10.0, 20.0, 30.0),
            (11.0, 21.0, 31.0),
            (12.0, 22.0, 32.0),
        }

    def test_source_plan_not_modified(self, svc_with_source):
        """克隆不修改源 plan 及其片段（保留用户手动编辑）。"""
        svc, source = svc_with_source
        source_clips_before = sorted(
            [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)]
        )
        source_name_before = source.name

        _clone_with_fake_calc(svc, source, [9.0, 19.0, 29.0])
        _clone_with_fake_calc(svc, source, [8.0, 18.0, 28.0])

        source_clips_after = sorted(
            [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)]
        )
        assert source_clips_after == source_clips_before
        assert svc._plan_repo.get(source.id).name == source_name_before

    def test_clone_copies_structure(self, svc_with_source):
        """克隆复制 template_id / config / total_duration / 片段素材与时长。"""
        svc, source = svc_with_source
        source.config = {"mode": "ONE_TAKE"}
        new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0])

        assert new_plan.template_id == source.template_id
        assert new_plan.total_duration == source.total_duration
        assert new_plan.config == {"mode": "ONE_TAKE"}
        assert "变体" in new_plan.name
        # 片段素材与时长结构保持
        assert [c["asset_id"] for c in clips_data] == ["a1", "a2", "a1"]
        assert all(c["duration"] == 5.0 for c in clips_data)
        assert [c["order"] for c in clips_data] == [0, 1, 2]

    def test_clone_uses_reuse_callback(self, svc_with_source):
        """克隆时 calc 传入了 on_exhausted 受控复用回调（耗尽时复用而非清空历史）。"""
        svc, source = svc_with_source
        _, _, calc_calls = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0])
        assert len(calc_calls) == 3
        for call in calc_calls:
            assert call["on_exhausted"] is not None

    def test_clone_ratio_blocked_keeps_original_start(self, svc_with_source):
        """复用占比闸门触发（calc 返回 None）时保留源片段原起点。"""
        svc, source = svc_with_source
        # 第 3 个片段 calc 返回 None（模拟复用占比超 15% 拒绝复用）
        new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, None])  # type: ignore[list-item]
        starts = [c["start_time"] for c in clips_data]
        assert starts[0] == 10.0
        assert starts[1] == 20.0
        # 第 3 片段保留源起点（源 order=2 → start_time=10.0）
        assert starts[2] == 10.0

    def test_clone_empty_source_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="无片段"):
            svc.clone_plan_for_variant(empty.id, name_suffix="变体")

    def test_clone_nonexistent_source_raises(self):
        """源 plan 不存在时抛出 ValueError。"""
        svc = _make_service()
        svc._clip_repo.session = MagicMock()
        with pytest.raises(ValueError, match="不存在"):
            svc.clone_plan_for_variant("no-such-plan", name_suffix="变体")