xiaoxia-saas/tests/unit/test_pip_engine.py

"""画中画（PiP）引擎单元测试."""

from __future__ import annotations

from pathlib import Path
from unittest.mock import patch

import pytest
from video_processing.pip_engine import (
    ANIMATION_FADE,
    ANIMATION_SLIDE_BOTTOM,
    ANIMATION_SLIDE_LEFT,
    ANIMATION_SLIDE_RIGHT,
    ANIMATION_SLIDE_TOP,
    POSITION_BOTTOM_LEFT,
    POSITION_BOTTOM_RIGHT,
    POSITION_CENTER,
    POSITION_TOP_LEFT,
    POSITION_TOP_RIGHT,
    PiPConfig,
    PiPEngine,
    PiPLayerConfig,
)

# ── PiPLayerConfig.validate 测试 ──────────────────────────────────────────────


class TestPiPLayerConfigValidate:
    """PiP图层配置校验测试."""

    def test_valid_config(self):
        """正常配置应该通过校验."""
        layer = PiPLayerConfig(source="asset_001")
        ok, err = layer.validate()
        assert ok
        assert err == ""

    def test_empty_source(self):
        """空source应该失败."""
        layer = PiPLayerConfig(source="")
        ok, err = layer.validate()
        assert not ok
        assert "source" in err

    def test_invalid_position(self):
        """无效位置应该失败."""
        layer = PiPLayerConfig(source="asset_001", position="invalid_pos")
        ok, err = layer.validate()
        assert not ok
        assert "position" in err

    def test_custom_position_valid(self):
        """custom位置应该通过."""
        layer = PiPLayerConfig(source="asset_001", position="custom", x=100, y=50)
        ok, err = layer.validate()
        assert ok

    def test_opacity_out_of_range_high(self):
        """opacity超过1应该失败."""
        layer = PiPLayerConfig(source="asset_001", opacity=1.5)
        ok, err = layer.validate()
        assert not ok
        assert "opacity" in err

    def test_opacity_out_of_range_low(self):
        """opacity小于0应该失败."""
        layer = PiPLayerConfig(source="asset_001", opacity=-0.5)
        ok, err = layer.validate()
        assert not ok
        assert "opacity" in err

    def test_opacity_boundary_values(self):
        """opacity边界值应该通过."""
        for val in [0.0, 0.5, 1.0]:
            layer = PiPLayerConfig(source="asset_001", opacity=val)
            ok, _ = layer.validate()
            assert ok

    def test_negative_corner_radius(self):
        """负圆角应该失败."""
        layer = PiPLayerConfig(source="asset_001", corner_radius=-5)
        ok, err = layer.validate()
        assert not ok
        assert "corner_radius" in err

    def test_negative_start_time(self):
        """负开始时间应该失败."""
        layer = PiPLayerConfig(source="asset_001", start_time=-1.0)
        ok, err = layer.validate()
        assert not ok
        assert "start_time" in err

    def test_negative_duration(self):
        """负持续时间应该失败."""
        layer = PiPLayerConfig(source="asset_001", duration=-5.0)
        ok, err = layer.validate()
        assert not ok
        assert "duration" in err

    def test_invalid_animation_in(self):
        """无效入场动画应该失败."""
        layer = PiPLayerConfig(source="asset_001", animation_in="spin")
        ok, err = layer.validate()
        assert not ok
        assert "入场动画" in err

    def test_all_valid_animations(self):
        """所有有效动画类型应该通过."""
        for anim in [
            ANIMATION_FADE,
            ANIMATION_SLIDE_LEFT,
            ANIMATION_SLIDE_RIGHT,
            ANIMATION_SLIDE_TOP,
            ANIMATION_SLIDE_BOTTOM,
        ]:
            layer = PiPLayerConfig(source="asset_001", animation_in=anim, animation_out=anim)
            ok, _ = layer.validate()
            assert ok

    def test_zero_duration_valid(self):
        """duration=0（全程显示）应该通过."""
        layer = PiPLayerConfig(source="asset_001", duration=0.0)
        ok, _ = layer.validate()
        assert ok


# ── PiPConfig.from_dict 测试 ──────────────────────────────────────────────────


class TestPiPConfigFromDict:
    """PiP配置字典解析测试."""

    def test_none_config(self):
        """None配置应该返回disabled."""
        config = PiPConfig.from_dict(None)
        assert not config.enabled
        assert len(config.layers) == 0

    def test_empty_config(self):
        """空字典应该返回disabled."""
        config = PiPConfig.from_dict({})
        assert not config.enabled

    def test_enabled_false(self):
        """enabled=False应该返回disabled."""
        config = PiPConfig.from_dict({"enabled": False, "layers": [{"source": "a"}]})
        assert not config.enabled

    def test_single_layer(self):
        """单图层解析."""
        data = {
            "enabled": True,
            "layers": [
                {
                    "source": "asset_001",
                    "position": POSITION_TOP_RIGHT,
                    "width": "30%",
                    "opacity": 0.9,
                    "corner_radius": 10,
                    "start_time": 2.0,
                    "duration": 5.0,
                    "z_index": 2,
                }
            ],
        }
        config = PiPConfig.from_dict(data)
        assert config.enabled
        assert len(config.layers) == 1
        layer = config.layers[0]
        assert layer.source == "asset_001"
        assert layer.position == POSITION_TOP_RIGHT
        assert layer.width == "30%"
        assert layer.opacity == 0.9
        assert layer.corner_radius == 10
        assert layer.start_time == 2.0
        assert layer.duration == 5.0
        assert layer.z_index == 2

    def test_multiple_layers_sorted_by_z_index(self):
        """多图层应该按z_index排序."""
        data = {
            "enabled": True,
            "layers": [
                {"source": "asset_high", "z_index": 5},
                {"source": "asset_low", "z_index": 1},
                {"source": "asset_mid", "z_index": 3},
            ],
        }
        config = PiPConfig.from_dict(data)
        assert len(config.layers) == 3
        assert config.layers[0].source == "asset_low"
        assert config.layers[1].source == "asset_mid"
        assert config.layers[2].source == "asset_high"

    def test_invalid_layer_skipped(self):
        """无效图层应该被跳过."""
        data = {
            "enabled": True,
            "layers": [
                {"source": "asset_good"},
                {"source": "", "position": "invalid"},  # 空source
                {"source": "asset_good2", "opacity": 2.0},  # opacity超范围
            ],
        }
        config = PiPConfig.from_dict(data)
        # 第1个有效，第2、3个无效
        assert len(config.layers) == 1
        assert config.layers[0].source == "asset_good"

    def test_all_invalid_layers_disabled(self):
        """所有图层都无效时enabled为False."""
        data = {
            "enabled": True,
            "layers": [
                {"source": ""},
                {"source": ""},
            ],
        }
        config = PiPConfig.from_dict(data)
        assert not config.enabled
        assert len(config.layers) == 0

    def test_default_values(self):
        """默认值应该正确."""
        data = {
            "enabled": True,
            "layers": [{"source": "asset_001"}],
        }
        config = PiPConfig.from_dict(data)
        layer = config.layers[0]
        assert layer.position == POSITION_BOTTOM_RIGHT
        assert layer.width == "25%"
        assert layer.opacity == 1.0
        assert layer.corner_radius == 0
        assert layer.start_time == 0.0
        assert layer.duration == 0.0
        assert layer.z_index == 1


# ── PiPEngine 位置计算测试 ────────────────────────────────────────────────────


class TestPiPEnginePosition:
    """PiP引擎位置计算测试."""

    @pytest.fixture
    def engine(self):
        return PiPEngine(output_width=1920, output_height=1080, output_fps=30)

    def test_top_left_position(self, engine):
        """左上角位置."""
        layer = PiPLayerConfig(source="a", position=POSITION_TOP_LEFT, margin=20)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 20
        assert y == 20

    def test_top_right_position(self, engine):
        """右上角位置."""
        layer = PiPLayerConfig(source="a", position=POSITION_TOP_RIGHT, margin=20)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 1920 - 480 - 20
        assert y == 20

    def test_bottom_right_position(self, engine):
        """右下角位置（默认）."""
        layer = PiPLayerConfig(source="a", position=POSITION_BOTTOM_RIGHT, margin=30)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 1920 - 480 - 30
        assert y == 1080 - 270 - 30

    def test_bottom_left_position(self, engine):
        """左下角位置."""
        layer = PiPLayerConfig(source="a", position=POSITION_BOTTOM_LEFT, margin=15)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 15
        assert y == 1080 - 270 - 15

    def test_center_position(self, engine):
        """中心位置."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, margin=0)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == (1920 - 480) // 2
        assert y == (1080 - 270) // 2

    def test_custom_position_pixel(self, engine):
        """自定义像素位置."""
        layer = PiPLayerConfig(source="a", position="custom", x=100, y=200)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 100
        assert y == 200

    def test_custom_position_percentage(self, engine):
        """自定义百分比位置."""
        layer = PiPLayerConfig(source="a", position="custom", x="50%", y="25%")
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 1920 // 2
        assert y == 1080 // 4

    def test_top_center_position(self, engine):
        """顶部居中位置."""
        layer = PiPLayerConfig(source="a", position="top_center", margin=10)
        x, y = engine._parse_position(layer, 480, 270)
        assert x == (1920 - 480) // 2
        assert y == 10

    def test_invalid_position_fallback(self, engine):
        """无效位置应该fallback到右下角."""
        layer = PiPLayerConfig(source="a", position="unknown_position", margin=20)
        # 直接测试_parse_position（注意：validate会拦截，但_parse_position自己也有fallback）
        x, y = engine._parse_position(layer, 480, 270)
        assert x == 1920 - 480 - 20
        assert y == 1080 - 270 - 20


# ── PiPEngine 尺寸解析测试 ────────────────────────────────────────────────────


class TestPiPEngineSize:
    """PiP引擎尺寸解析测试."""

    @pytest.fixture
    def engine(self):
        return PiPEngine(output_width=1920, output_height=1080, output_fps=30)

    def test_pixel_size_int(self, engine):
        """像素尺寸（整数）."""
        assert engine._parse_size(500, 1920) == 500

    def test_pixel_size_str(self, engine):
        """像素尺寸（字符串数字）."""
        assert engine._parse_size("500", 1920) == 500

    def test_percentage_size(self, engine):
        """百分比尺寸."""
        assert engine._parse_size("50%", 1920) == 960
        assert engine._parse_size("25%", 1920) == 480

    def test_zero_size_default(self, engine):
        """0或无效值应该有最小值保护."""
        assert engine._parse_size(0, 1920) == 1
        assert engine._parse_size("", 1920) == 480  # 默认25%

    def test_negative_size_default(self, engine):
        """负值应该取绝对值后至少为1."""
        # _parse_size 用 max(1, value)，负值会走 except 分支
        result = engine._parse_size("-100", 1920)
        # 会走ValueError分支，返回默认值
        assert result > 0


# ── PiPEngine 滤镜构建测试 ────────────────────────────────────────────────────


class TestPiPEngineBuildFilters:
    """PiP引擎滤镜构建测试."""

    @pytest.fixture
    def engine(self):
        return PiPEngine(output_width=1920, output_height=1080, output_fps=30)

    @pytest.fixture
    def fake_video(self, tmp_path):
        """创建一个假的视频文件路径."""
        path = tmp_path / "test_video.mp4"
        path.write_bytes(b"fake video data")
        return path

    def test_empty_sources(self, engine):
        """空素材列表应该返回空."""
        filters, inputs, label = engine.build_pip_filters("base_label", [])
        assert filters == []
        assert inputs == []
        assert label == "base_label"

    def test_single_layer_basic(self, engine, fake_video):
        """单图层基础滤镜构建."""
        layer = PiPLayerConfig(
            source="asset_001",
            position=POSITION_TOP_RIGHT,
            width="25%",
        )
        sources = [("pip_src_0", layer, fake_video)]

        filters, inputs, final_label = engine.build_pip_filters("base_video", sources, base_input_idx=3)

        # 应该有2个滤镜: 预处理 + overlay
        assert len(filters) == 2
        # 输入参数应该有2个（-i + path）
        assert len(inputs) == 2
        assert inputs[0] == "-i"
        assert inputs[1] == str(fake_video)

        # 预处理滤镜应该使用正确的输入索引
        assert "3:v" in filters[0]
        # 应该包含scale
        assert "scale=" in filters[0]
        # 应该有pip_pre_0标签
        assert "[pip_pre_0]" in filters[0]

        # overlay滤镜
        assert "overlay=" in filters[1]
        assert "[base_video][pip_pre_0]" in filters[1]

    def test_single_layer_final_label(self, engine, fake_video):
        """最终输出标签应该正确."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER)
        sources = [("s0", layer, fake_video)]

        _, _, final_label = engine.build_pip_filters("main_v", sources)
        assert final_label == "pip_combined_0"

    def test_multiple_layers(self, engine, fake_video):
        """多图层叠加."""
        layer1 = PiPLayerConfig(source="a", position=POSITION_TOP_LEFT, z_index=1)
        layer2 = PiPLayerConfig(source="b", position=POSITION_BOTTOM_RIGHT, z_index=2)
        sources = [
            ("s0", layer1, fake_video),
            ("s1", layer2, fake_video),
        ]

        filters, inputs, final_label = engine.build_pip_filters("base", sources, base_input_idx=0)

        # 2层 × 2个滤镜（预处理+overlay）= 4个滤镜
        assert len(filters) == 4
        # 2个输入文件
        assert len(inputs) == 4  # 2 × (-i + path)

        # 输入索引应该连续
        assert "0:v" in filters[0]
        assert "1:v" in filters[2]

        # 最终标签应该是第二个overlay的输出
        assert final_label == "pip_combined_1"

    def test_with_opacity(self, engine, fake_video):
        """透明度应该在滤镜中体现."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, opacity=0.5)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "colorchannelmixer=aa=0.5" in pre_filter
        assert "yuva420p" in pre_filter

    def test_with_corner_radius(self, engine, fake_video):
        """圆角裁剪应该在滤镜中体现."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, corner_radius=20)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "geq=" in pre_filter

    def test_with_border(self, engine, fake_video):
        """边框应该在滤镜中体现."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, border_width=3, border_color="red")
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "pad=" in pre_filter
        assert "red" in pre_filter

    def test_timing_start_time_and_duration(self, engine, fake_video):
        """时间控制应该生成enable表达式."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=5.0, duration=10.0)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        overlay_filter = filters[1]
        assert "enable=" in overlay_filter
        assert "between(t,5.0,15.0)" in overlay_filter

    def test_timing_start_time_only(self, engine, fake_video):
        """只有开始时间（全程显示到结束）."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=3.0, duration=0.0)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        overlay_filter = filters[1]
        assert "enable=" in overlay_filter
        assert "gte(t,3.0)" in overlay_filter

    def test_no_timing_no_enable(self, engine, fake_video):
        """无时间限制时不应该有enable表达式."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, start_time=0.0, duration=0.0)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        overlay_filter = filters[1]
        assert "enable=" not in overlay_filter

    def test_fade_animation(self, engine, fake_video):
        """淡入淡出动画."""
        layer = PiPLayerConfig(
            source="a",
            position=POSITION_CENTER,
            animation_in=ANIMATION_FADE,
            animation_out=ANIMATION_FADE,
            duration=10.0,
            animation_duration=0.8,
        )
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "fade=t=in" in pre_filter
        assert "fade=t=out" in pre_filter
        assert "alpha=1" in pre_filter

    def test_slide_animation_in(self, engine, fake_video):
        """滑入动画应该在overlay表达式中."""
        layer = PiPLayerConfig(
            source="a",
            position=POSITION_CENTER,
            animation_in=ANIMATION_SLIDE_LEFT,
            animation_duration=0.5,
        )
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        overlay_filter = filters[1]
        # x表达式应该包含动态变化
        assert "overlay=" in overlay_filter

    def test_full_opacity_no_alpha(self, engine, fake_video):
        """opacity=1时不应该有colorchannelmixer."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, opacity=1.0)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "colorchannelmixer" not in pre_filter

    def test_zero_corner_radius_no_geq(self, engine, fake_video):
        """corner_radius=0时不应该有geq滤镜."""
        layer = PiPLayerConfig(source="a", position=POSITION_CENTER, corner_radius=0)
        sources = [("s0", layer, fake_video)]

        filters, _, _ = engine.build_pip_filters("base", sources)
        pre_filter = filters[0]
        assert "geq=" not in pre_filter


# ── PiPEngine 素材验证（降级策略）测试 ────────────────────────────────────────


class TestPiPEngineValidateSource:
    """PiP引擎素材验证与降级测试."""

    @pytest.fixture
    def engine(self):
        return PiPEngine(output_width=1920, output_height=1080, output_fps=30)

    def test_asset_id_in_map(self, engine, tmp_path):
        """asset_id在map中应该返回路径."""
        asset_path = tmp_path / "test.mp4"
        asset_path.write_bytes(b"data")
        asset_map = {"asset_001": asset_path}

        layer = PiPLayerConfig(source="asset_001", source_type="asset_id")
        result = engine.validate_layer_source(layer, asset_map)
        assert result == asset_path

    def test_asset_id_not_in_map(self, engine):
        """asset_id不在map中应该返回None（降级）."""
        layer = PiPLayerConfig(source="nonexistent", source_type="asset_id")
        result = engine.validate_layer_source(layer, {})
        assert result is None

    def test_local_path_exists(self, engine, tmp_path):
        """本地路径存在应该返回."""
        path = tmp_path / "video.mp4"
        path.write_bytes(b"data")

        layer = PiPLayerConfig(source=str(path), source_type="local_path")
        result = engine.validate_layer_source(layer, {})
        assert result == path

    def test_local_path_not_exists(self, engine):
        """本地路径不存在应该返回None（降级）."""
        layer = PiPLayerConfig(source="/nonexistent/path.mp4", source_type="local_path")
        result = engine.validate_layer_source(layer, {})
        assert result is None

    def test_url_type_not_supported(self, engine):
        """URL类型暂时不支持，返回None."""
        layer = PiPLayerConfig(source="http://example.com/video.mp4", source_type="url")
        result = engine.validate_layer_source(layer, {})
        assert result is None

    def test_exception_handling(self, engine):
        """异常情况应该返回None（不阻断）."""
        layer = PiPLayerConfig(source=None, source_type="local_path")  # type: ignore
        # 模拟异常情况
        result = engine.validate_layer_source(layer, {})
        assert result is None