xiaoxia-saas/tests/unit/test_plan_generator_utils.py

"""plan_generator_utils 纯逻辑单测 — 第90波.

测试素材分配、clip_type映射、默认clip生成、配置转clip等纯函数。
不依赖 DB，使用领域对象直接构造。
"""

import pytest

from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.editing_mode import EditingMode
from packages.domain.plan_generator_utils import (
    DEFAULT_CLIP_DURATION,
    create_clips_from_configs,
    distribute_assets,
    generate_default_clips,
    map_clip_types_for_mode,
)
from packages.domain.template_clip_config import ClipType, TemplateClipConfig

# ── 辅助函数 ──────────────────────────────────────────────────────────


def _make_main_clip(plan_id: str = "plan1", order: int = 0) -> EditPlanClip:
    """创建一个 MAIN 类型的 clip."""
    return EditPlanClip.create(
        plan_id=plan_id,
        clip_type=ClipType.MAIN.value,
        order=order,
        duration=5.0,
    )


def _make_clips(n: int, clip_type: str = "main") -> list[EditPlanClip]:
    """创建 n 个指定类型的 clip."""
    return [
        EditPlanClip.create(
            plan_id="plan1",
            clip_type=clip_type,
            order=i,
            duration=5.0,
        )
        for i in range(n)
    ]


def _collect_asset_ids(clips: list[EditPlanClip]) -> list[str]:
    """按顺序收集 clips 的 asset_id（空的跳过）."""
    return [c.asset_id for c in clips if c.asset_id]


# ── distribute_assets: ONE_TAKE ─────────────────────────────────────


class TestDistributeOneTake:
    """ONE_TAKE 模式素材分配."""

    def test_equal_count(self):
        """素材数 == clip 数：一一对应."""
        clips = _make_clips(3)
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == "a2"
        assert clips[2].asset_id == "a3"

    def test_more_assets_than_clips(self):
        """素材多于 clip：多余的不用."""
        clips = _make_clips(2)
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == "a2"

    def test_fewer_assets_than_clips(self):
        """素材少于 clip：后面的 clip 没素材."""
        clips = _make_clips(5)
        distribute_assets(clips, ["a1", "a2"], EditingMode.ONE_TAKE.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == "a2"
        assert clips[2].asset_id == ""
        assert clips[3].asset_id == ""
        assert clips[4].asset_id == ""

    def test_empty_assets(self):
        """空素材列表：无分配."""
        clips = _make_clips(3)
        distribute_assets(clips, [], EditingMode.ONE_TAKE.value)
        for c in clips:
            assert c.asset_id == ""

    def test_empty_clips(self):
        """空 clip 列表：不报错."""
        distribute_assets([], ["a1"], EditingMode.ONE_TAKE.value)

    def test_only_main_clips_get_assigned(self):
        """只分配给 MAIN 类型 clip，其他类型不受影响."""
        clips = _make_clips(2) + _make_clips(2, "intro") + _make_clips(2, "outro")
        distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.ONE_TAKE.value)
        mains = [c for c in clips if c.clip_type == "main"]
        others = [c for c in clips if c.clip_type != "main"]
        assert mains[0].asset_id == "a1"
        assert mains[1].asset_id == "a2"
        for c in others:
            assert c.asset_id == ""


# ── distribute_assets: PIP ──────────────────────────────────────────


class TestDistributePip:
    """PIP 模式素材分配."""

    def test_basic_pip_distribution(self):
        """第1个素材给 main，其余给 overlay."""
        clips = _make_clips(1) + _make_clips(3, "overlay")
        distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.PIP.value)
        mains = [c for c in clips if c.clip_type == "main"]
        overlays = [c for c in clips if c.clip_type == "overlay"]
        assert mains[0].asset_id == "a1"
        assert overlays[0].asset_id == "a2"
        assert overlays[1].asset_id == "a3"
        assert overlays[2].asset_id == "a4"

    def test_single_asset_only_main(self):
        """只有1个素材：只分配给 main，overlay 没素材."""
        clips = _make_clips(1) + _make_clips(2, "overlay")
        distribute_assets(clips, ["a1"], EditingMode.PIP.value)
        mains = [c for c in clips if c.clip_type == "main"]
        overlays = [c for c in clips if c.clip_type == "overlay"]
        assert mains[0].asset_id == "a1"
        assert overlays[0].asset_id == ""
        assert overlays[1].asset_id == ""

    def test_more_overlays_than_assets(self):
        """overlay 多于剩余素材：后面的 overlay 没素材."""
        clips = _make_clips(1) + _make_clips(5, "overlay")
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value)
        overlays = [c for c in clips if c.clip_type == "overlay"]
        assert overlays[0].asset_id == "a2"
        assert overlays[1].asset_id == "a3"
        assert overlays[2].asset_id == ""
        assert overlays[3].asset_id == ""
        assert overlays[4].asset_id == ""

    def test_no_main_clip(self):
        """没有 main clip：第1个素材没人拿，overlay 从第2个素材开始."""
        clips = _make_clips(3, "overlay")
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value)
        overlays = [c for c in clips if c.clip_type == "overlay"]
        # PIP 逻辑：先给 main 分配第1个素材（没有 main 则跳过），
        # 剩余从第2个开始分配给 overlay
        assert overlays[0].asset_id == "a2"
        assert overlays[1].asset_id == "a3"
        assert overlays[2].asset_id == ""


# ── distribute_assets: VOICE_OVER ───────────────────────────────────


class TestDistributeVoiceOver:
    """VOICE_OVER 模式素材分配."""

    def test_voice_over_same_as_one_take(self):
        """VOICE_OVER 和 ONE_TAKE 分配策略相同：按顺序给 main."""
        clips = _make_clips(3)
        assets = ["a1", "a2", "a3"]
        distribute_assets(clips, assets, EditingMode.VOICE_OVER.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == "a2"
        assert clips[2].asset_id == "a3"

    def test_voice_over_fewer_assets(self):
        """素材不足时，后面的 main clip 没素材."""
        clips = _make_clips(5)
        distribute_assets(clips, ["a1"], EditingMode.VOICE_OVER.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == ""


# ── distribute_assets: VOICE_PIP ────────────────────────────────────


class TestDistributeVoicePip:
    """VOICE_PIP 模式素材分配."""

    def test_three_assets_full_distribution(self):
        """3个素材：background + corner_voice + b_roll 各一个."""
        clips = _make_clips(1, "background") + _make_clips(1, "corner_voice") + _make_clips(1, "b_roll")
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value)
        bgs = [c for c in clips if c.clip_type == "background"]
        voices = [c for c in clips if c.clip_type == "corner_voice"]
        brolls = [c for c in clips if c.clip_type == "b_roll"]
        assert bgs[0].asset_id == "a1"
        assert voices[0].asset_id == "a2"
        assert brolls[0].asset_id == "a3"

    def test_single_asset_only_background(self):
        """1个素材：只分配给 background."""
        clips = _make_clips(1, "background") + _make_clips(1, "corner_voice") + _make_clips(2, "b_roll")
        distribute_assets(clips, ["a1"], EditingMode.VOICE_PIP.value)
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == ""
        assert clips[2].asset_id == ""
        assert clips[3].asset_id == ""

    def test_two_assets_bg_and_voice(self):
        """2个素材：background + corner_voice."""
        clips = _make_clips(1, "background") + _make_clips(1, "corner_voice") + _make_clips(2, "b_roll")
        distribute_assets(clips, ["a1", "a2"], EditingMode.VOICE_PIP.value)
        bgs = [c for c in clips if c.clip_type == "background"]
        voices = [c for c in clips if c.clip_type == "corner_voice"]
        brolls = [c for c in clips if c.clip_type == "b_roll"]
        assert bgs[0].asset_id == "a1"
        assert voices[0].asset_id == "a2"
        assert brolls[0].asset_id == ""

    def test_many_broll_clips(self):
        """多个 b_roll clip：按顺序分配剩余素材."""
        clips = _make_clips(1, "background") + _make_clips(1, "corner_voice") + _make_clips(5, "b_roll")
        distribute_assets(
            clips,
            ["a1", "a2", "a3", "a4", "a5"],
            EditingMode.VOICE_PIP.value,
        )
        brolls = [c for c in clips if c.clip_type == "b_roll"]
        assert brolls[0].asset_id == "a3"
        assert brolls[1].asset_id == "a4"
        assert brolls[2].asset_id == "a5"
        assert brolls[3].asset_id == ""
        assert brolls[4].asset_id == ""

    def test_missing_some_layer_clips(self):
        """缺少某些层的 clip 不影响其他层."""
        # 没有 corner_voice，素材应该按顺序：bg 拿 a1，b_roll 从 a2 开始
        clips = _make_clips(1, "background") + _make_clips(3, "b_roll")
        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value)
        bgs = [c for c in clips if c.clip_type == "background"]
        brolls = [c for c in clips if c.clip_type == "b_roll"]
        assert bgs[0].asset_id == "a1"
        # 没有 corner_voice，b_roll 从第2个素材开始
        assert brolls[0].asset_id == "a2"
        assert brolls[1].asset_id == "a3"


# ── distribute_assets: 边缘情况 ─────────────────────────────────────


class TestDistributeEdgeCases:
    """素材分配边缘情况."""

    def test_unknown_mode_falls_back_to_one_take(self):
        """未知模式退化为 ONE_TAKE."""
        clips = _make_clips(3)
        distribute_assets(clips, ["a1", "a2", "a3"], "unknown_mode")
        assert clips[0].asset_id == "a1"
        assert clips[1].asset_id == "a2"
        assert clips[2].asset_id == "a3"

    def test_both_empty(self):
        """两边都空：不报错."""
        distribute_assets([], [], EditingMode.ONE_TAKE.value)


# ── map_clip_types_for_mode ─────────────────────────────────────────


class TestMapClipTypesForMode:
    """clip_type 按模式映射."""

    def test_one_take_unchanged(self):
        """ONE_TAKE 模式：main 保持 main."""
        clips = _make_clips(5)
        map_clip_types_for_mode(clips, EditingMode.ONE_TAKE.value)
        for c in clips:
            assert c.clip_type == "main"

    def test_voice_over_unchanged(self):
        """VOICE_OVER 模式：main 保持 main."""
        clips = _make_clips(5)
        map_clip_types_for_mode(clips, EditingMode.VOICE_OVER.value)
        for c in clips:
            assert c.clip_type == "main"

    def test_pip_first_main_stays_rest_become_overlay(self):
        """PIP 模式：第1个 main 保持，其余变 overlay."""
        clips = _make_clips(5)
        map_clip_types_for_mode(clips, EditingMode.PIP.value)
        assert clips[0].clip_type == "main"
        assert clips[1].clip_type == "overlay"
        assert clips[2].clip_type == "overlay"
        assert clips[3].clip_type == "overlay"
        assert clips[4].clip_type == "overlay"

    def test_pip_single_main_unchanged(self):
        """PIP 模式只有1个 main：保持 main."""
        clips = _make_clips(1)
        map_clip_types_for_mode(clips, EditingMode.PIP.value)
        assert clips[0].clip_type == "main"

    def test_voice_pip_three_types(self):
        """VOICE_PIP 模式：background + corner_voice + b_roll."""
        clips = _make_clips(5)
        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
        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"
        assert clips[4].clip_type == "b_roll"

    def test_voice_pip_one_main(self):
        """VOICE_PIP 只有1个 main：变成 background."""
        clips = _make_clips(1)
        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
        assert clips[0].clip_type == "background"

    def test_voice_pip_two_mains(self):
        """VOICE_PIP 2个 main：background + corner_voice."""
        clips = _make_clips(2)
        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
        assert clips[0].clip_type == "background"
        assert clips[1].clip_type == "corner_voice"

    def test_non_main_clips_unchanged(self):
        """非 MAIN 类型 clip 不受影响."""
        clips = _make_clips(1, "intro") + _make_clips(3) + _make_clips(1, "outro")  # main
        map_clip_types_for_mode(clips, EditingMode.PIP.value)
        assert clips[0].clip_type == "intro"
        assert clips[1].clip_type == "main"  # 第1个 main
        assert clips[2].clip_type == "overlay"  # 第2个 main → overlay
        assert clips[3].clip_type == "overlay"  # 第3个 main → overlay
        assert clips[4].clip_type == "outro"

    def test_no_main_clips_noop(self):
        """没有 main clip：什么都不做."""
        clips = _make_clips(3, "intro")
        original_types = [c.clip_type for c in clips]
        map_clip_types_for_mode(clips, EditingMode.PIP.value)
        assert [c.clip_type for c in clips] == original_types

    def test_empty_clips_noop(self):
        """空列表：不报错."""
        map_clip_types_for_mode([], EditingMode.PIP.value)


# ── generate_default_clips ──────────────────────────────────────────


class TestGenerateDefaultClips:
    """默认 clip 生成."""

    def test_one_take_normal(self):
        """ONE_TAKE：N 个 main clip."""
        clips = generate_default_clips("plan1", EditingMode.ONE_TAKE.value, 5)
        assert len(clips) == 5
        for c in clips:
            assert c.clip_type == "main"
            assert c.plan_id == "plan1"
            assert c.duration == DEFAULT_CLIP_DURATION
        # order 递增
        for i in range(5):
            assert clips[i].order == i

    def test_pip_structure(self):
        """PIP：1个 main + (N-1)个 overlay."""
        clips = generate_default_clips("plan1", EditingMode.PIP.value, 4)
        assert len(clips) == 4
        assert clips[0].clip_type == "main"
        assert clips[1].clip_type == "overlay"
        assert clips[2].clip_type == "overlay"
        assert clips[3].clip_type == "overlay"
        assert clips[0].order == 0
        assert clips[3].order == 3

    def test_pip_single_asset(self):
        """PIP 只有1个素材：1个 main，没有 overlay."""
        clips = generate_default_clips("plan1", EditingMode.PIP.value, 1)
        assert len(clips) == 1
        assert clips[0].clip_type == "main"

    def test_voice_over_structure(self):
        """VOICE_OVER：N 个 main clip，带 b_roll 标记."""
        clips = generate_default_clips("plan1", EditingMode.VOICE_OVER.value, 3)
        assert len(clips) == 3
        for c in clips:
            assert c.clip_type == "main"
            assert c.config.get("role") == "b_roll"

    def test_voice_pip_three_layers(self):
        """VOICE_PIP：background + corner_voice + b_roll."""
        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 5)
        assert len(clips) == 5
        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"
        assert clips[4].clip_type == "b_roll"

    def test_voice_pip_single_asset(self):
        """VOICE_PIP 1个素材：只有 background."""
        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 1)
        assert len(clips) == 1
        assert clips[0].clip_type == "background"

    def test_voice_pip_two_assets(self):
        """VOICE_PIP 2个素材：background + corner_voice."""
        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 2)
        assert len(clips) == 2
        assert clips[0].clip_type == "background"
        assert clips[1].clip_type == "corner_voice"

    def test_zero_assets_at_least_one(self):
        """0 个素材：至少生成 1 个 clip."""
        for mode in [
            EditingMode.ONE_TAKE.value,
            EditingMode.PIP.value,
            EditingMode.VOICE_OVER.value,
            EditingMode.VOICE_PIP.value,
        ]:
            clips = generate_default_clips("plan1", mode, 0)
            assert len(clips) >= 1

    def test_unknown_mode_falls_back(self):
        """未知模式退化为 ONE_TAKE 风格."""
        clips = generate_default_clips("plan1", "unknown", 3)
        assert len(clips) == 3
        for c in clips:
            assert c.clip_type == "main"

    def test_order_is_sequential(self):
        """所有模式下 order 都是从 0 开始连续递增."""
        for mode in [
            EditingMode.ONE_TAKE.value,
            EditingMode.PIP.value,
            EditingMode.VOICE_OVER.value,
            EditingMode.VOICE_PIP.value,
        ]:
            clips = generate_default_clips("plan1", mode, 5)
            for i, c in enumerate(clips):
                assert c.order == i


# ── create_clips_from_configs ───────────────────────────────────────


class TestCreateClipsFromConfigs:
    """从模板配置创建 clips."""

    def test_basic_creation(self):
        """基本创建：按 order 排序，属性正确传递."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=1,
                min_duration=3.0,
                max_duration=7.0,
                transition_effect="fade",
            ),
            TemplateClipConfig(
                id="cfg2",
                template_id="tpl1",
                clip_type=ClipType.INTRO,
                order=0,
                min_duration=2.0,
                max_duration=4.0,
                transition_effect="cut",
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert len(clips) == 2
        # 按 order 排序：intro(order=0) 在前，main(order=1) 在后
        assert clips[0].clip_type == "intro"
        assert clips[1].clip_type == "main"
        assert clips[0].order == 0
        assert clips[1].order == 1
        assert clips[0].template_clip_config_id == "cfg2"
        assert clips[1].template_clip_config_id == "cfg1"

    def test_duration_average_of_min_max(self):
        """min_duration 和 max_duration 都有时，取平均值."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=4.0,
                max_duration=6.0,
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].duration == 5.0  # (4+6)/2

    def test_duration_only_min(self):
        """只有 min_duration 时，用 min_duration."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3.5,
                max_duration=0,
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].duration == 3.5

    def test_duration_only_max(self):
        """只有 max_duration 时，用 max_duration."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=0,
                max_duration=8.0,
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].duration == 8.0

    def test_duration_default_when_both_zero(self):
        """都为 0 时用默认时长."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=0,
                max_duration=0,
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].duration == DEFAULT_CLIP_DURATION

    def test_empty_configs_returns_empty(self):
        """空配置列表返回空列表."""
        clips = create_clips_from_configs("plan1", [])
        assert clips == []

    def test_plan_id_passed_through(self):
        """plan_id 正确传递给所有 clip."""
        configs = [
            TemplateClipConfig(
                id=f"cfg{i}",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=i,
                min_duration=3,
                max_duration=5,
            )
            for i in range(3)
        ]
        clips = create_clips_from_configs("my_plan", configs)
        for c in clips:
            assert c.plan_id == "my_plan"

    def test_playback_speed_from_config(self):
        """playback_speed 从 config.playback_speed 读取."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3,
                max_duration=5,
                config={"playback_speed": 1.5},
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].playback_speed == 1.5

    def test_playback_speed_fallback_to_speed_ratio(self):
        """playback_speed 不存在时回退到 speed_ratio."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3,
                max_duration=5,
                config={"speed_ratio": 0.8},
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].playback_speed == 0.8

    def test_playback_speed_default_1(self):
        """没有 speed 配置时默认为 1.0."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3,
                max_duration=5,
                config={},
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].playback_speed == 1.0

    def test_playback_speed_none_falls_back(self):
        """playback_speed 为 None 时回退到 1.0."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3,
                max_duration=5,
                config={"playback_speed": None},
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].playback_speed == 1.0

    def test_transition_default_cut(self):
        """transition_effect 为空时默认为 cut."""
        configs = [
            TemplateClipConfig(
                id="cfg1",
                template_id="tpl1",
                clip_type=ClipType.MAIN,
                order=0,
                min_duration=3,
                max_duration=5,
                transition_effect=None,
            ),
        ]
        clips = create_clips_from_configs("plan1", configs)
        assert clips[0].transition_effect == "cut"


# ── 常量导出 ────────────────────────────────────────────────────────


class TestConstants:
    """常量导出验证."""

    def test_default_duration_value(self):
        """默认片段时长应为 5 秒."""
        assert DEFAULT_CLIP_DURATION == 5.0