xiaoxia-saas/tests/unit/test_generation_plan_builder.py

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