"""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="变体")