"""Unit tests for generation_plan_builder.py — pure logic utilities. 覆盖: - VirtualPlan / VirtualClip 数据类 - extract_intro_outro_from_clip_configs - apply_template_clip_effects - build_clips_by_mode (4种模式) - build_error_info """ from __future__ import annotations from dataclasses import dataclass from typing import Any from unittest.mock import patch import pytest from worker_app.tasks.generation_plan_builder import ( VirtualClip, VirtualPlan, apply_template_clip_effects, build_clips_by_mode, build_error_info, extract_intro_outro_from_clip_configs, ) # ── 辅助：模拟 clip_config 对象 ────────────────────────────────────────────── @dataclass class MockClipType: """模拟 Enum 类型的 clip_type。""" value: str @dataclass class MockTransition: """模拟 Enum 类型的 transition_effect。""" value: str @dataclass class MockClipConfig: """模拟 TemplateClipConfig 对象。""" clip_type: Any transition_effect: Any = "cut" default_duration: float = 3.0 text_template: str = "" config: dict | None = None def _make_config( clip_type: str = "main", transition: str = "cut", duration: float = 3.0, text: str = "", config: dict | None = None, use_enum: bool = True, ) -> MockClipConfig: """创建一个模拟 clip_config。""" ct = MockClipType(clip_type) if use_enum else clip_type tr = MockTransition(transition) if use_enum and transition != "cut" else transition return MockClipConfig( clip_type=ct, transition_effect=tr, default_duration=duration, text_template=text, config=config or {}, ) # ── VirtualPlan / VirtualClip ─────────────────────────────────────────────── class TestVirtualPlan: def test_default_values(self): plan = VirtualPlan(id="plan_001") assert plan.id == "plan_001" assert plan.name == "" assert plan.config == {} def test_full_init(self): plan = VirtualPlan(id="p1", name="My Plan", config={"key": "value"}) assert plan.id == "p1" assert plan.name == "My Plan" assert plan.config == {"key": "value"} def test_mutable_config(self): plan = VirtualPlan(id="p1") plan.config["new_key"] = "new_val" assert plan.config == {"new_key": "new_val"} class TestVirtualClip: def test_default_values(self): clip = VirtualClip(id="c001") assert clip.id == "c001" assert clip.plan_id == "" assert clip.clip_type == "main" assert clip.order == 0 assert clip.asset_id == "" assert clip.duration == 0.0 assert clip.transition_effect == "cut" assert clip.transition_duration == 0.0 assert clip.playback_speed == 1.0 assert clip.status == "ready" assert clip.config == {} def test_full_init(self): clip = VirtualClip( id="c001", plan_id="p1", clip_type="overlay", order=1, asset_id="asset_001", duration=5.5, transition_effect="fade", transition_duration=0.5, playback_speed=1.5, config={"role": "b_roll"}, ) assert clip.clip_type == "overlay" assert clip.duration == 5.5 assert clip.playback_speed == 1.5 def test_mutable_config(self): clip = VirtualClip(id="c001") clip.config["effect"] = "vintage" assert clip.config == {"effect": "vintage"} # ── extract_intro_outro_from_clip_configs ─────────────────────────────────── class TestExtractIntroOutro: def test_empty_configs(self): result = extract_intro_outro_from_clip_configs([]) assert result == {} def test_no_intro_no_outro(self): configs = [_make_config("main"), _make_config("showcase")] result = extract_intro_outro_from_clip_configs(configs) assert result == {} def test_intro_only_basic(self): configs = [_make_config("intro", text="Hello", duration=2.5)] result = extract_intro_outro_from_clip_configs(configs) assert result["has_intro"] is True assert result["intro_type"] == "text" assert result["intro_duration"] == 2.5 assert result["intro_text"] == "Hello" assert "has_outro" not in result def test_outro_only_basic(self): configs = [_make_config("outro", text="Bye", duration=3.0)] result = extract_intro_outro_from_clip_configs(configs) assert result["has_outro"] is True assert result["outro_type"] == "text" assert result["outro_duration"] == 3.0 assert result["outro_text"] == "Bye" def test_both_intro_and_outro(self): configs = [ _make_config("intro", text="Start"), _make_config("main"), _make_config("outro", text="End"), ] result = extract_intro_outro_from_clip_configs(configs) assert result["has_intro"] is True assert result["intro_text"] == "Start" assert result["has_outro"] is True assert result["outro_text"] == "End" def test_intro_extra_config_pass_through(self): configs = [ _make_config( "intro", config={ "intro_text_color": "#ffffff", "intro_bg_color": "#000000", "intro_font_size": 32, "intro_video_url": "https://example.com/intro.mp4", "intro_video_path": "/tmp/intro.mp4", "random_key": "should_not_appear", }, ) ] result = extract_intro_outro_from_clip_configs(configs) assert result["intro_text_color"] == "#ffffff" assert result["intro_bg_color"] == "#000000" assert result["intro_font_size"] == 32 assert result["intro_video_url"] == "https://example.com/intro.mp4" assert "random_key" not in result def test_outro_extra_config_pass_through(self): configs = [ _make_config( "outro", config={ "outro_text_color": "#ff0000", "outro_bg_color": "#00ff00", "outro_font_size": 24, "outro_follow_text": "关注我们", }, ) ] result = extract_intro_outro_from_clip_configs(configs) assert result["outro_text_color"] == "#ff0000" assert result["outro_follow_text"] == "关注我们" def test_intro_type_from_config(self): configs = [_make_config("intro", config={"intro_type": "video"})] result = extract_intro_outro_from_clip_configs(configs) assert result["intro_type"] == "video" def test_default_duration_when_zero(self): configs = [_make_config("intro", duration=0.0)] result = extract_intro_outro_from_clip_configs(configs) assert result["intro_duration"] == 3.0 def test_empty_text_not_included(self): configs = [_make_config("intro", text="")] result = extract_intro_outro_from_clip_configs(configs) assert "intro_text" not in result def test_with_string_clip_type_no_enum(self): configs = [_make_config("intro", text="Hi", use_enum=False)] result = extract_intro_outro_from_clip_configs(configs) assert result["has_intro"] is True assert result["intro_text"] == "Hi" def test_first_intro_used_when_multiple(self): configs = [ _make_config("intro", text="First"), _make_config("intro", text="Second"), ] result = extract_intro_outro_from_clip_configs(configs) assert result["intro_text"] == "First" def test_none_config_handled(self): cfg = _make_config("intro") cfg.config = None result = extract_intro_outro_from_clip_configs([cfg]) assert result["has_intro"] is True # ── apply_template_clip_effects ───────────────────────────────────────────── class TestApplyTemplateClipEffects: def test_empty_clips(self): clips: list[VirtualClip] = [] configs = [_make_config("main", transition="fade")] apply_template_clip_effects(clips, configs, "one_take") assert clips == [] def test_empty_configs(self): clips = [VirtualClip(id="c1", clip_type="main")] apply_template_clip_effects(clips, [], "one_take") assert clips[0].transition_effect == "cut" def test_no_main_configs(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("intro"), _make_config("outro")] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_effect == "cut" def test_transition_effect_applied(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", transition="fade")] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_effect == "fade" def test_transition_duration_applied(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", transition="fade", config={"transition_duration": 0.8})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_duration == 0.8 def test_transition_duration_invalid_ignored(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", transition="fade", config={"transition_duration": "abc"})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_duration == 0.0 def test_cut_transition_not_applied(self): clips = [VirtualClip(id="c1", clip_type="main", transition_effect="dissolve")] configs = [_make_config("main", transition="cut")] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_effect == "dissolve" # 保留原值 def test_color_grade_applied(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"color_grade": "vintage"})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].config["color_grade"] == "vintage" def test_multiple_effect_keys_applied(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [ _make_config( "main", config={ "color_grade": "warm", "playback_speed": 1.5, "reverse": True, "chroma_key": {"color": "green"}, "filter": "黑白", }, ) ] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].config["color_grade"] == "warm" assert clips[0].config["playback_speed"] == 1.5 assert clips[0].config["reverse"] is True assert clips[0].config["chroma_key"] == {"color": "green"} def test_playback_speed_top_level_updated(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"playback_speed": 2.0})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].playback_speed == 2.0 assert clips[0].config["playback_speed"] == 2.0 def test_speed_fallback_sets_playback_speed(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"speed": 0.8})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].playback_speed == 0.8 def test_playback_speed_takes_priority_over_speed(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"speed": 0.5, "playback_speed": 2.0})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].playback_speed == 2.0 def test_existing_config_preserved(self): clips = [VirtualClip(id="c1", clip_type="main", config={"role": "b_roll", "original": "value"})] configs = [_make_config("main", config={"color_grade": "cool"})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].config["role"] == "b_roll" assert clips[0].config["original"] == "value" assert clips[0].config["color_grade"] == "cool" def test_corner_voice_skipped(self): clips = [ VirtualClip(id="c1", clip_type="main"), VirtualClip(id="c2", clip_type="corner_voice"), ] configs = [_make_config("main", transition="fade")] apply_template_clip_effects(clips, configs, "voice_pip") assert clips[0].transition_effect == "fade" assert clips[1].transition_effect == "cut" # 不应用效果 def test_cyclic_matching_more_clips_than_configs(self): clips = [ VirtualClip(id="c1", clip_type="main"), VirtualClip(id="c2", clip_type="main"), VirtualClip(id="c3", clip_type="main"), ] configs = [ _make_config("main", transition="fade"), _make_config("main", transition="dissolve"), ] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_effect == "fade" assert clips[1].transition_effect == "dissolve" assert clips[2].transition_effect == "dissolve" # 循环用最后一个 def test_pip_mode_main_and_overlay_both_effected(self): clips = [ VirtualClip(id="c1", clip_type="main"), VirtualClip(id="c2", clip_type="overlay"), ] configs = [_make_config("main", transition="fade")] apply_template_clip_effects(clips, configs, "pip") assert clips[0].transition_effect == "fade" assert clips[1].transition_effect == "fade" def test_showcase_and_b_roll_count_as_template_source(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [ _make_config("showcase", transition="zoom_in"), _make_config("b_roll", transition="slide_left"), ] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].transition_effect == "zoom_in" # 用第一个匹配的 def test_string_transition_no_enum(self): clips = [VirtualClip(id="c1", clip_type="main")] cfg = _make_config("main") cfg.transition_effect = "wipe" # 直接字符串 apply_template_clip_effects(clips, [cfg], "one_take") assert clips[0].transition_effect == "wipe" def test_zero_speed_ignored(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"playback_speed": 0})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].playback_speed == 1.0 # 保持默认 def test_negative_speed_ignored(self): clips = [VirtualClip(id="c1", clip_type="main")] configs = [_make_config("main", config={"playback_speed": -1.0})] apply_template_clip_effects(clips, configs, "one_take") assert clips[0].playback_speed == 1.0 # ── build_clips_by_mode ──────────────────────────────────────────────────── class TestBuildClipsByMode: def test_one_take_single_asset(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": "a1", "duration": 10.0}], mode="one_take", ) assert len(clips) == 1 assert clips[0].clip_type == "main" assert clips[0].asset_id == "a1" assert clips[0].duration == 10.0 assert clips[0].order == 0 assert clips[0].plan_id == "p1" def test_one_take_multiple_assets(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[ {"asset_id": "a1", "duration": 5.0}, {"asset_id": "a2", "duration": 10.0}, {"asset_id": "a3", "duration": 7.0}, ], mode="one_take", ) assert len(clips) == 3 assert all(c.clip_type == "main" for c in clips) assert [c.order for c in clips] == [0, 1, 2] assert [c.duration for c in clips] == [5.0, 10.0, 7.0] def test_pip_mode_main_and_overlay(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[ {"asset_id": "a1", "duration": 10.0}, {"asset_id": "a2", "duration": 5.0}, {"asset_id": "a3", "duration": 3.0}, ], mode="pip", ) assert len(clips) == 3 assert clips[0].clip_type == "main" assert clips[1].clip_type == "overlay" assert clips[2].clip_type == "overlay" def test_pip_single_asset_is_main(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": "a1", "duration": 10.0}], mode="pip", ) assert len(clips) == 1 assert clips[0].clip_type == "main" def test_voice_over_all_main_with_role(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[ {"asset_id": "a1", "duration": 5.0}, {"asset_id": "a2", "duration": 8.0}, ], mode="voice_over", ) assert len(clips) == 2 assert all(c.clip_type == "main" for c in clips) assert all(c.config.get("role") == "b_roll" for c in clips) def test_voice_pip_three_layers(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[ {"asset_id": "a1", "duration": 10.0}, {"asset_id": "a2", "duration": 5.0}, {"asset_id": "a3", "duration": 3.0}, {"asset_id": "a4", "duration": 4.0}, ], mode="voice_pip", ) assert len(clips) == 4 assert clips[0].clip_type == "background" assert clips[1].clip_type == "corner_voice" assert clips[2].clip_type == "b_roll" assert clips[3].clip_type == "b_roll" def test_voice_pip_two_assets(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[ {"asset_id": "a1", "duration": 10.0}, {"asset_id": "a2", "duration": 5.0}, ], mode="voice_pip", ) assert len(clips) == 2 assert clips[0].clip_type == "background" assert clips[1].clip_type == "corner_voice" def test_voice_pip_single_asset(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": "a1", "duration": 10.0}], mode="voice_pip", ) assert len(clips) == 1 assert clips[0].clip_type == "background" def test_default_mode_is_one_take(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": "a1", "duration": 10.0}], mode="unknown_mode", ) assert len(clips) == 1 assert clips[0].clip_type == "main" def test_empty_assets(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[], mode="one_take", ) assert clips == [] def test_default_asset_id_when_missing(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"duration": 5.0}], mode="one_take", ) assert clips[0].asset_id == "asset_000" def test_default_duration_when_missing(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": "a1"}], mode="one_take", ) assert clips[0].duration == 0.0 def test_clip_ids_sequential(self): clips = build_clips_by_mode( plan_id="p1", asset_infos=[{"asset_id": f"a{i}"} for i in range(5)], mode="one_take", ) assert [c.id for c in clips] == ["vc_000", "vc_001", "vc_002", "vc_003", "vc_004"] def test_plan_id_propagated(self): clips = build_clips_by_mode( plan_id="my_plan_123", asset_infos=[{"asset_id": "a1"}, {"asset_id": "a2"}], mode="pip", ) assert all(c.plan_id == "my_plan_123" for c in clips) # ── build_error_info ─────────────────────────────────────────────────────── class TestBuildErrorInfo: def test_basic_structure(self): try: raise ValueError("test error") except ValueError as e: info = build_error_info(e, stage="render") assert info["error_type"] == "ValueError" assert info["message"] == "test error" assert info["stage"] == "render" assert "stack_trace" in info assert "failed_at" in info assert "ValueError" in info["stack_trace"] assert "test error" in info["stack_trace"] def test_default_stage(self): try: raise RuntimeError("oops") except RuntimeError as e: info = build_error_info(e) assert info["stage"] == "render" def test_custom_stage(self): try: raise TypeError("bad type") except TypeError as e: info = build_error_info(e, stage="download") assert info["stage"] == "download" def test_failed_at_is_iso_format(self): try: raise ValueError("x") except ValueError as e: info = build_error_info(e) # ISO 格式检查：包含 T 和 + assert "T" in info["failed_at"] @patch("worker_app.tasks.generation_plan_builder.traceback.format_exc") def test_long_stack_trace_truncated(self, mock_format): # 构造30行堆栈 lines = [f' File "file{i}.py", line {i}, in func{i}' for i in range(28)] lines.append("ValueError: deep error") mock_format.return_value = "\n".join(lines) try: raise ValueError("deep") except ValueError as e: info = build_error_info(e) assert "truncated" in info["stack_trace"] assert "total 29 lines" in info["stack_trace"] # 确认只保留了前20行 assert "func19" in info["stack_trace"] assert "func20" not in info["stack_trace"] @patch("worker_app.tasks.generation_plan_builder.traceback.format_exc") def test_short_stack_trace_not_truncated(self, mock_format): # 构造5行堆栈（小于20） mock_format.return_value = ( "Traceback (most recent call last):\n" ' File "test.py", line 10, in foo\n' ' raise ValueError("simple")\n' "ValueError: simple\n" ) try: raise ValueError("simple") except ValueError as e: info = build_error_info(e) assert "truncated" not in info["stack_trace"] assert "ValueError: simple" in info["stack_trace"]