""" PlanGeneratorService 单元测试 覆盖（6 组测试）： - ONE_TAKE 模式：素材顺序分配给 main clips - PIP 模式：第1个素材→main，其余→overlay - VOICE_OVER 模式：素材→main clips (B-roll) - VOICE_PIP 模式：第1个→background, 第2个→corner_voice, 其余→b_roll - 无 clip_configs 时自动生成默认结构 - 空素材列表时 clips 创建但无素材分配 """ from __future__ import annotations import os import sys from datetime import datetime, timezone from pathlib import Path from typing import List, Optional 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")) from packages.domain.edit_plan import EditPlan, EditPlanStatus from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus from packages.domain.edit_template import EditTemplate, EditTemplateStatus from packages.domain.editing_mode import EditingMode from packages.domain.template_clip_config import ClipType, TemplateClipConfig, TransitionEffect # --------------------------------------------------------------------------- # Stub Repositories # --------------------------------------------------------------------------- class StubEditPlanRepository: """内存中的 EditPlan 仓储 stub""" def __init__(self) -> None: self._plans: dict[str, EditPlan] = {} self._counter = 0 def _next_id(self) -> str: self._counter += 1 return f"plan-{self._counter:03d}" def get(self, plan_id: str) -> Optional[EditPlan]: return self._plans.get(plan_id) def create(self, plan: EditPlan) -> EditPlan: if not plan.id: plan = EditPlan( id=self._next_id(), template_id=plan.template_id, name=plan.name, status=plan.status, total_duration=plan.total_duration, source_edit_plan_id=plan.source_edit_plan_id, project_id=plan.project_id, created_by_user_id=plan.created_by_user_id, config=plan.config, created_at=plan.created_at, updated_at=plan.updated_at, ) self._plans[plan.id] = plan return plan def update(self, plan: EditPlan) -> EditPlan: self._plans[plan.id] = plan return plan def list_all(self, **kwargs) -> List[EditPlan]: return list(self._plans.values()) def count(self, **kwargs) -> int: return len(self._plans) def delete(self, plan_id: str) -> bool: return self._plans.pop(plan_id, None) is not None class StubEditPlanClipRepository: """内存中的 EditPlanClip 仓储 stub""" def __init__(self) -> None: self._clips: dict[str, EditPlanClip] = {} self._counter = 0 def _next_id(self) -> str: self._counter += 1 return f"clip-{self._counter:03d}" def get(self, clip_id: str) -> Optional[EditPlanClip]: return self._clips.get(clip_id) def create(self, clip: EditPlanClip) -> EditPlanClip: if not clip.id: clip = EditPlanClip( id=self._next_id(), plan_id=clip.plan_id, template_clip_config_id=clip.template_clip_config_id, clip_type=clip.clip_type, order=clip.order, asset_id=clip.asset_id, text_content=clip.text_content, start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect, status=clip.status, config=clip.config, created_at=clip.created_at, updated_at=clip.updated_at, ) self._clips[clip.id] = clip return clip def update(self, clip: EditPlanClip) -> EditPlanClip: self._clips[clip.id] = clip return clip def list_by_plan( self, plan_id: str, *, status: Optional[EditPlanClipStatus] = None, skip: int = 0, limit: int = 100, ) -> List[EditPlanClip]: items = [c for c in self._clips.values() if c.plan_id == plan_id] items.sort(key=lambda c: c.order) if status: items = [c for c in items if c.status == status] return items[skip : skip + limit] def delete(self, clip_id: str) -> bool: return self._clips.pop(clip_id, None) is not None # --------------------------------------------------------------------------- # Helper: 构建 PlanGeneratorService（patch 仓储） # --------------------------------------------------------------------------- def _make_generator(): """创建使用 stub 仓储的 PlanGeneratorService""" from apps.api.app.services.plan_generator_service import PlanGeneratorService plan_repo = StubEditPlanRepository() clip_repo = StubEditPlanClipRepository() with ( patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository", return_value=plan_repo, ), patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository", return_value=clip_repo, ), ): db = MagicMock() svc = PlanGeneratorService(db) # 替换为 stub svc._plan_repo = plan_repo svc._clip_repo = clip_repo return svc, plan_repo, clip_repo def _make_template( editing_mode: str = "one_take", config: Optional[dict] = None, ) -> EditTemplate: """创建测试用 EditTemplate""" return EditTemplate( id="tpl-001", name="测试模板", description="", template_type="default", editing_mode=editing_mode, config=config or {}, preview_url="", sort_weight=0, status=EditTemplateStatus.ACTIVE, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) def _make_clip_configs( template_id: str = "tpl-001", specs: Optional[List[dict]] = None, ) -> List[TemplateClipConfig]: """创建测试用 TemplateClipConfig 列表 specs 示例: [{"clip_type": ClipType.INTRO, "order": 0}, ...] """ if specs is None: specs = [ {"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0}, {"clip_type": ClipType.MAIN, "order": 1, "min_duration": 3.0, "max_duration": 7.0}, {"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0}, ] configs = [] for i, spec in enumerate(specs): cfg = TemplateClipConfig( id=f"cfg-{i:03d}", template_id=template_id, clip_type=spec.get("clip_type", ClipType.MAIN), order=spec.get("order", i), min_duration=spec.get("min_duration", 0.0), max_duration=spec.get("max_duration", 0.0), text_template=spec.get("text_template", ""), material_requirements=spec.get("material_requirements"), transition_effect=spec.get("transition_effect", TransitionEffect.CUT), config=spec.get("config"), created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) configs.append(cfg) return configs # --------------------------------------------------------------------------- # 测试：ONE_TAKE 模式 # --------------------------------------------------------------------------- class TestGenerateOneTakePlan: """ONE_TAKE 模式：素材顺序分配给 main clips""" def test_generate_one_take_plan(self): """3个clip_configs + 3个asset_ids → 按顺序分配""" svc, plan_repo, clip_repo = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs() asset_ids = ["asset-1", "asset-2", "asset-3"] result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=asset_ids, project_id="proj-001", created_by_user_id="user-001", ) plan = result["plan"] clips = result["clips"] assert plan.template_id == "tpl-001" assert plan.config["editing_mode"] == "one_take" assert plan.status == EditPlanStatus.EDITING assert len(clips) == 3 # 按 order 排序后检查素材分配 sorted_clips = sorted(clips, key=lambda c: c.order) # intro clip (order=0) 不是 main 类型，不分配素材 # main clip (order=1) → asset-2（ONE_TAKE 只分配给 main clips） # outro clip (order=2) 不是 main 类型 main_clips = [c for c in sorted_clips if c.clip_type == ClipType.MAIN.value] assert len(main_clips) == 1 assert main_clips[0].asset_id == "asset-1" def test_one_take_plan_name_from_template(self): """name 为空时自动取模板名""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["asset-1"], ) assert "测试模板" in result["plan"].name def test_one_take_custom_name(self): """指定 name 时使用自定义名称""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["asset-1"], name="我的剪辑", ) assert result["plan"].name == "我的剪辑" # --------------------------------------------------------------------------- # 测试：PIP 模式 # --------------------------------------------------------------------------- class TestGeneratePipPlan: """PIP 模式：第1个素材→main，其余→overlay""" def test_generate_pip_plan(self): """4个素材 → 第1个→main，其余→overlay""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.PIP.value) # 无 clip_configs，自动生成默认结构 asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"] result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=asset_ids, ) plan = result["plan"] clips = result["clips"] assert plan.config["editing_mode"] == "pip" # 自动生成: 1 main + 3 overlay assert len(clips) == 4 main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] overlay_clips = [c for c in clips if c.clip_type == "overlay"] assert len(main_clips) == 1 assert len(overlay_clips) == 3 # 第1个素材 → main assert main_clips[0].asset_id == "bg-asset" # 其余 → overlay assert overlay_clips[0].asset_id == "overlay-1" assert overlay_clips[1].asset_id == "overlay-2" assert overlay_clips[2].asset_id == "overlay-3" def test_pip_with_clip_configs(self): """PIP + 有 clip_configs（都是 MAIN 类型）→ 自动映射为 main + overlay，素材正确分配""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.PIP.value) # 4 个 MAIN 类型的 clip_config（模拟 PIP 模板配置） clip_configs = [ TemplateClipConfig.create( template_id=template.id, clip_type=ClipType.MAIN, order=i, min_duration=3.0, max_duration=5.0, ) for i in range(4) ] asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"] result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=asset_ids, ) plan = result["plan"] clips = result["clips"] assert plan.config["editing_mode"] == "pip" assert len(clips) == 4 # 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画） main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] overlay_clips = [c for c in clips if c.clip_type == "overlay"] assert len(main_clips) == 1 assert len(overlay_clips) == 3 # 素材分配正确 assert main_clips[0].asset_id == "bg-asset" assert overlay_clips[0].asset_id == "overlay-1" assert overlay_clips[1].asset_id == "overlay-2" assert overlay_clips[2].asset_id == "overlay-3" def test_voice_pip_with_clip_configs(self): """VOICE_PIP + 有 clip_configs（都是 MAIN 类型）→ 自动映射为 background + corner_voice + b_roll""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.VOICE_PIP.value) # 4 个 MAIN 类型的 clip_config clip_configs = [ TemplateClipConfig.create( template_id=template.id, clip_type=ClipType.MAIN, order=i, min_duration=3.0, max_duration=5.0, ) for i in range(4) ] asset_ids = ["bg-asset", "voice-asset", "broll-1", "broll-2"] result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=asset_ids, ) plan = result["plan"] clips = result["clips"] assert plan.config["editing_mode"] == "voice_pip" assert len(clips) == 4 bg_clips = [c for c in clips if c.clip_type == "background"] corner_clips = [c for c in clips if c.clip_type == "corner_voice"] broll_clips = [c for c in clips if c.clip_type == "b_roll"] assert len(bg_clips) == 1 assert len(corner_clips) == 1 assert len(broll_clips) == 2 # 素材分配正确 assert bg_clips[0].asset_id == "bg-asset" assert corner_clips[0].asset_id == "voice-asset" assert broll_clips[0].asset_id == "broll-1" assert broll_clips[1].asset_id == "broll-2" # --------------------------------------------------------------------------- # 测试：VOICE_OVER 模式 # --------------------------------------------------------------------------- class TestGenerateVoiceOverPlan: """VOICE_OVER 模式：素材→main clips (B-roll)""" def test_generate_voice_over_plan(self): """3个素材 → 3个 main clips，每个标记为 b_roll""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.VOICE_OVER.value) asset_ids = ["video-1", "video-2", "video-3"] result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=asset_ids, ) plan = result["plan"] clips = result["clips"] assert plan.config["editing_mode"] == "voice_over" assert len(clips) == 3 # 所有 clips 都是 main 类型 for clip in clips: assert clip.clip_type == ClipType.MAIN.value # 素材按顺序分配 assert clips[0].asset_id == "video-1" assert clips[1].asset_id == "video-2" assert clips[2].asset_id == "video-3" # 每个 clip 的 config 标记为 b_roll for clip in clips: assert clip.config.get("role") == "b_roll" # --------------------------------------------------------------------------- # 测试：VOICE_PIP 模式 # --------------------------------------------------------------------------- class TestGenerateVoicePipPlan: """VOICE_PIP 模式：第1个→background, 第2个→corner_voice, 其余→b_roll""" def test_generate_voice_pip_plan(self): """4个素材 → background + corner_voice + 2 b_roll""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.VOICE_PIP.value) asset_ids = ["bg-video", "corner-video", "broll-1", "broll-2"] result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=asset_ids, ) plan = result["plan"] clips = result["clips"] assert plan.config["editing_mode"] == "voice_pip" assert len(clips) == 4 bg_clips = [c for c in clips if c.clip_type == "background"] corner_clips = [c for c in clips if c.clip_type == "corner_voice"] broll_clips = [c for c in clips if c.clip_type == "b_roll"] assert len(bg_clips) == 1 assert len(corner_clips) == 1 assert len(broll_clips) == 2 # 素材分配 assert bg_clips[0].asset_id == "bg-video" assert corner_clips[0].asset_id == "corner-video" assert broll_clips[0].asset_id == "broll-1" assert broll_clips[1].asset_id == "broll-2" # --------------------------------------------------------------------------- # 测试：无 clip_configs 时自动生成默认结构 # --------------------------------------------------------------------------- class TestGenerateWithoutClipConfigs: """无 clip_configs 时根据 editing_mode 生成默认 clip 结构""" def test_one_take_default_clips(self): """ONE_TAKE + 3个素材 → 3个 main clips""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["a1", "a2", "a3"], ) clips = result["clips"] assert len(clips) == 3 for clip in clips: assert clip.clip_type == ClipType.MAIN.value def test_pip_default_clips(self): """PIP + 3个素材 → 1 main + 2 overlay""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.PIP.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["a1", "a2", "a3"], ) clips = result["clips"] assert len(clips) == 3 main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] overlay_clips = [c for c in clips if c.clip_type == "overlay"] assert len(main_clips) == 1 assert len(overlay_clips) == 2 def test_voice_pip_default_clips(self): """VOICE_PIP + 4个素材 → 1 background + 1 corner_voice + 2 b_roll""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.VOICE_PIP.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["a1", "a2", "a3", "a4"], ) clips = result["clips"] assert len(clips) == 4 # --------------------------------------------------------------------------- # 测试：空素材列表 # --------------------------------------------------------------------------- class TestGenerateEmptyAssets: """空素材列表时 clips 创建但无素材分配""" def test_empty_assets(self): """空 asset_ids → clips 创建但 asset_id 为空""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs() result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=[], ) plan = result["plan"] clips = result["clips"] assert plan.total_duration > 0 # clips 有默认时长 assert len(clips) == 3 for clip in clips: assert clip.asset_id == "" def test_empty_assets_pip(self): """PIP 模式空素材 → 1个 main clip（至少1个）""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.PIP.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=[], ) clips = result["clips"] # 至少1个 main clip（n = max(asset_count, 1) = 1） assert len(clips) == 1 assert clips[0].clip_type == ClipType.MAIN.value # --------------------------------------------------------------------------- # 测试：plan config 继承模板配置 # --------------------------------------------------------------------------- class TestPlanConfigInheritance: """plan config 继承模板的 cover/title/subtitle/bgm""" def test_inherit_template_config(self): """模板有 cover/title/bgm 配置 → plan 继承""" svc, _, _ = _make_generator() template_config = { "editing_mode": "one_take", "cover": {"type": "ai_frame"}, "title": {"text": "测试标题", "font": "思源黑体"}, "bgm": {"url": "https://example.com/bgm.mp3"}, } template = _make_template( editing_mode=EditingMode.ONE_TAKE.value, config=template_config, ) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["a1"], ) plan_config = result["plan"].config assert plan_config["editing_mode"] == "one_take" assert plan_config["cover"]["type"] == "ai_frame" assert plan_config["title"]["text"] == "测试标题" assert plan_config["bgm"]["url"] == "https://example.com/bgm.mp3" # --------------------------------------------------------------------------- # 测试：total_duration 计算 # --------------------------------------------------------------------------- class TestTotalDuration: """total_duration 正确计算""" def test_duration_from_clip_configs(self): """有 clip_configs 时，duration 取 min/max 中间值""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs( specs=[ {"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0}, {"clip_type": ClipType.MAIN, "order": 1, "min_duration": 4.0, "max_duration": 6.0}, {"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0}, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) # intro: (2+4)/2=3, main: (4+6)/2=5, outro: (2+4)/2=3 → total=11 assert result["plan"].total_duration == 11.0 def test_duration_from_default_clips(self): """无 clip_configs 时，每个 clip 默认 5 秒""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) result = svc.generate_from_template( template=template, clip_configs=[], asset_ids=["a1", "a2", "a3"], ) # 3 个 clips × 5 秒 = 15 秒 assert result["plan"].total_duration == 15.0 # --------------------------------------------------------------------------- # 测试：模板配置传递（export / filter / 片段级 speed_ratio） # --------------------------------------------------------------------------- class TestTemplateConfigPropagation: """模板配置到剪辑计划的传递验证""" def test_export_config_propagated(self): """模板 export 配置（watermark/resolution）应传递到 plan.config.export""" svc, _, _ = _make_generator() template = _make_template( editing_mode=EditingMode.ONE_TAKE.value, config={ "export": { "resolution": "1080x1920", "fps": 30, "watermark_enabled": True, "watermark_text": "小虾剪辑", }, }, ) clip_configs = _make_clip_configs( specs=[ {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) export_cfg = result["plan"].config.get("export", {}) assert export_cfg.get("watermark_enabled") is True assert export_cfg.get("watermark_text") == "小虾剪辑" assert export_cfg.get("resolution") == "1080x1920" assert export_cfg.get("fps") == 30 def test_filter_config_propagated(self): """模板 filter 配置应传递到 plan.config.filter""" svc, _, _ = _make_generator() template = _make_template( editing_mode=EditingMode.ONE_TAKE.value, config={ "filter": { "enabled": True, "preset_id": "filter_vivid", "intensity": 80, }, }, ) clip_configs = _make_clip_configs( specs=[ {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) filter_cfg = result["plan"].config.get("filter", {}) assert filter_cfg.get("enabled") is True assert filter_cfg.get("preset_id") == "filter_vivid" assert filter_cfg.get("intensity") == 80 def test_subtitle_asr_config_propagated(self): """模板 subtitle ASR 配置应传递到 plan.config.subtitle""" svc, _, _ = _make_generator() template = _make_template( editing_mode=EditingMode.ONE_TAKE.value, config={ "subtitle": { "enabled": True, "auto_generated": True, "position": "bottom", "source": "asr", }, }, ) clip_configs = _make_clip_configs( specs=[ {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) subtitle_cfg = result["plan"].config.get("subtitle", {}) assert subtitle_cfg.get("enabled") is True assert subtitle_cfg.get("auto_generated") is True assert subtitle_cfg.get("position") == "bottom" def test_speed_ratio_mapped_to_playback_speed(self): """clip_config.config.speed_ratio 应映射到 clip.playback_speed""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs( specs=[ { "clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0, "config": {"speed_ratio": 1.2, "name": "开场"}, }, { "clip_type": ClipType.MAIN, "order": 1, "min_duration": 3.0, "max_duration": 5.0, "config": {"speed_ratio": 0.8, "name": "结尾"}, }, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1", "a2"], ) clips = result["clips"] assert clips[0].playback_speed == 1.2 assert clips[1].playback_speed == 0.8 def test_playback_speed_takes_priority_over_speed_ratio(self): """clip_config.config.playback_speed 优先于 speed_ratio""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs( specs=[ { "clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0, "config": {"speed_ratio": 1.2, "playback_speed": 1.5}, }, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) assert result["clips"][0].playback_speed == 1.5 def test_default_playback_speed_is_1_0(self): """无 speed_ratio / playback_speed 时，默认 1.0""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs( specs=[ {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) assert result["clips"][0].playback_speed == 1.0 def test_clip_config_dict_propagated(self): """clip_config.config 整体应传递到 clip.config""" svc, _, _ = _make_generator() template = _make_template(editing_mode=EditingMode.ONE_TAKE.value) clip_configs = _make_clip_configs( specs=[ { "clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0, "config": {"speed_ratio": 1.2, "name": "开场", "custom_field": "value"}, }, ] ) result = svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) clip_cfg = result["clips"][0].config assert clip_cfg.get("speed_ratio") == 1.2 assert clip_cfg.get("name") == "开场" assert clip_cfg.get("custom_field") == "value" class TestAssetDurationsAlwaysFetched: """验证 asset_durations 不再受 random_preview 条件限制。 修复前：asset_durations 仅在 random_preview=True 时传入 distribute_assets 修复后：只要 _asset_repo 存在，就始终获取 asset_durations """ def _make_service_with_asset_repo(self): """创建带 mock asset_repo 的 PlanGeneratorService""" from apps.api.app.services.plan_generator_service import PlanGeneratorService plan_repo = StubEditPlanRepository() clip_repo = StubEditPlanClipRepository() # mock asset_repo: 返回带 duration 的素材 asset_repo = MagicMock() def fake_get(asset_id): mock_asset = MagicMock() mock_asset.duration = 30.0 # 每个素材 30 秒 # score_asset 所需的属性 mock_asset.quality_score = None mock_asset.created_at = None mock_asset.metadata = {} return mock_asset asset_repo.get = MagicMock(side_effect=fake_get) with ( patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository", return_value=plan_repo, ), patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository", return_value=clip_repo, ), ): db = MagicMock() svc = PlanGeneratorService(db, asset_repo=asset_repo) svc._plan_repo = plan_repo svc._clip_repo = clip_repo return svc, asset_repo def test_asset_durations_fetched_without_random_preview(self): """random_preview=False 时也应获取 asset_durations""" svc, asset_repo = self._make_service_with_asset_repo() template = _make_template("one_take") clip_configs = _make_clip_configs( template_id=template.id, specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}], ) # patch distribute_assets 以捕获传入的参数 with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute: svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1", "a2"], random_preview=False, # 关键：非随机预览模式 ) # 验证 asset_durations 被传入（不是 None） mock_distribute.assert_called_once() call_kwargs = mock_distribute.call_args asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations")) assert asset_durations is not None, "asset_durations should be fetched even when random_preview=False" assert "a1" in asset_durations assert "a2" in asset_durations assert asset_durations["a1"] == 30.0 # 验证 asset_repo.get 被调用（说明 _fetch_asset_durations 执行了） assert asset_repo.get.call_count >= 2 def test_asset_durations_fetched_with_random_preview(self): """random_preview=True 时仍正常获取 asset_durations（行为不变）""" svc, asset_repo = self._make_service_with_asset_repo() template = _make_template("one_take") clip_configs = _make_clip_configs( template_id=template.id, specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}], ) with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute: svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], random_preview=True, ) call_kwargs = mock_distribute.call_args asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations")) assert asset_durations is not None assert "a1" in asset_durations def test_no_asset_repo_means_no_durations(self): """_asset_repo 为 None 时 asset_durations 应为 None""" svc, _, _ = _make_generator() # 默认不带 asset_repo template = _make_template("one_take") clip_configs = _make_clip_configs( template_id=template.id, specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}], ) with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute: svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=["a1"], ) call_kwargs = mock_distribute.call_args asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations")) assert asset_durations is None # --------------------------------------------------------------------------- # 测试：正式生成片段随机重排（Issue #1663） # --------------------------------------------------------------------------- class TestFormalGenerationShuffle: """验证正式生成时片段顺序随机化。 Issue #1663: 正式生成时 smart_match 排序后对 asset_ids 做 random.shuffle， 使得同一批素材每次生成的视频片段顺序不同，有利于查重降重。 """ def _make_service_with_asset_repo(self): """创建带 mock asset_repo 的 PlanGeneratorService（复用 TestAssetDurationsAlwaysFetched 模式）""" from apps.api.app.services.plan_generator_service import PlanGeneratorService plan_repo = StubEditPlanRepository() clip_repo = StubEditPlanClipRepository() asset_repo = MagicMock() def fake_get(asset_id): mock_asset = MagicMock() mock_asset.duration = 30.0 mock_asset.quality_score = None mock_asset.created_at = None mock_asset.metadata = {} return mock_asset asset_repo.get = MagicMock(side_effect=fake_get) with ( patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository", return_value=plan_repo, ), patch( "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository", return_value=clip_repo, ), ): db = MagicMock() svc = PlanGeneratorService(db, asset_repo=asset_repo) svc._plan_repo = plan_repo svc._clip_repo = clip_repo return svc, asset_repo def test_formal_generation_shuffles_asset_ids(self): """正式生成路径下 asset_ids 应被打乱，多次调用顺序应不同""" svc, _ = self._make_service_with_asset_repo() template = _make_template("one_take") # 6 个 clip 容纳 6 个素材 clip_configs = _make_clip_configs( template_id=template.id, specs=[ {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(6) ], ) asset_ids = ["a1", "a2", "a3", "a4", "a5", "a6"] # 收集多次调用中 distribute_assets 收到的 asset_ids 顺序 captured_orders = [] with patch( "apps.api.app.services.plan_generator_service.distribute_assets", side_effect=lambda clips, asset_ids, *a, **kw: captured_orders.append(list(asset_ids)), ): # mock _sort_assets_by_smart_score 返回固定顺序，验证 shuffle 会打乱 with patch.object( svc, "_sort_assets_by_smart_score", side_effect=lambda ids: list(ids), # 原样返回 ): with patch.object( svc, "_fetch_asset_scene_points", return_value={}, ): for _ in range(10): svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=list(asset_ids), # 每次传新列表 random_preview=False, # 正式生成 ) assert len(captured_orders) == 10 # 每次 order 应该是 asset_ids 的一个排列 expected_set = set(asset_ids) for order in captured_orders: assert set(order) == expected_set # 10 次调用中应至少出现 2 种不同顺序（概率 > 99.9%） unique_orders = set(tuple(o) for o in captured_orders) assert ( len(unique_orders) >= 2 ), f"Expected shuffled orders to vary, but got only {len(unique_orders)} unique order(s): {unique_orders}" def test_formal_generation_does_not_mutate_original_list(self): """shuffle 不应修改调用方的原始 asset_ids 列表""" svc, _ = self._make_service_with_asset_repo() template = _make_template("one_take") clip_configs = _make_clip_configs( template_id=template.id, specs=[ {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(4) ], ) original = ["a1", "a2", "a3", "a4"] original_copy = list(original) with patch("apps.api.app.services.plan_generator_service.distribute_assets"): with patch.object(svc, "_sort_assets_by_smart_score", side_effect=lambda ids: list(ids)): with patch.object(svc, "_fetch_asset_scene_points", return_value={}): svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=original, random_preview=False, ) assert original == original_copy, "Original asset_ids list should not be mutated" def test_preview_random_mode_unaffected_by_shuffle(self): """预览随机模式不走 shuffle 路径，行为不变""" svc, _ = self._make_service_with_asset_repo() template = _make_template("one_take") clip_configs = _make_clip_configs( template_id=template.id, specs=[ {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(4) ], ) asset_ids = ["a1", "a2", "a3", "a4"] captured_orders = [] with patch( "apps.api.app.services.plan_generator_service.distribute_assets", side_effect=lambda clips, asset_ids, *a, **kw: captured_orders.append(list(asset_ids)), ): for _ in range(5): svc.generate_from_template( template=template, clip_configs=clip_configs, asset_ids=list(asset_ids), random_preview=True, # 预览随机模式 ) assert len(captured_orders) == 5 # 预览模式下 random.shuffle 不应被调用（在 _distribute_assets 的 if not random_selection 块内） # 所以 asset_ids 应该保持调用方传入的顺序（可能已由上层 shuffle 过）