xiaoxia-saas/tests/unit/test_watermark_intro_outro.py

"""水印 + 片头片尾引擎单元测试."""

import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent / "apps" / "worker"))

from video_processing.intro_outro_engine import (
    IntroOutroConfig,
    IntroOutroEngine,
)
from video_processing.watermark_engine import (
    WATERMARK_POSITIONS,
    WatermarkConfig,
    WatermarkEngine,
)


class TestWatermarkConfig(unittest.TestCase):
    """WatermarkConfig 单元测试."""

    def test_from_dict_none_disabled(self):
        """空配置或未启用 → None."""
        self.assertIsNone(WatermarkConfig.from_dict(None))
        self.assertIsNone(WatermarkConfig.from_dict({}))
        self.assertIsNone(WatermarkConfig.from_dict({"enabled": False}))

    def test_from_dict_text_mode(self):
        """文字水印模式."""
        cfg = WatermarkConfig.from_dict(
            {
                "enabled": True,
                "mode": "text",
                "text": "hello world",
                "position": "top_left",
            }
        )
        self.assertIsNotNone(cfg)
        self.assertEqual(cfg.mode, "text")
        self.assertEqual(cfg.text, "hello world")
        self.assertEqual(cfg.position, "top_left")

    def test_from_dict_image_missing_path(self):
        """图片水印缺路径 → None."""
        cfg = WatermarkConfig.from_dict(
            {
                "enabled": True,
                "mode": "image",
            }
        )
        self.assertIsNone(cfg)

    def test_from_dict_text_missing_text(self):
        """文字水印缺文字 → None."""
        cfg = WatermarkConfig.from_dict(
            {
                "enabled": True,
                "mode": "text",
            }
        )
        self.assertIsNone(cfg)

    def test_validate_text_valid(self):
        """文字水印合法配置."""
        cfg = WatermarkConfig(
            mode="text",
            text="test",
            position="bottom_right",
        )
        ok, err = cfg.validate()
        self.assertTrue(ok)
        self.assertEqual(err, "")

    def test_validate_invalid_position(self):
        """非法位置."""
        cfg = WatermarkConfig(mode="text", text="test", position="invalid")
        ok, err = cfg.validate()
        self.assertFalse(ok)
        self.assertIn("不支持的位置", err)

    def test_validate_opacity_out_of_range(self):
        """透明度超范围."""
        cfg = WatermarkConfig(mode="text", text="test", opacity=1.5)
        ok, err = cfg.validate()
        self.assertFalse(ok)

    def test_validate_image_missing_path(self):
        """图片水印缺路径."""
        cfg = WatermarkConfig(mode="image")
        ok, err = cfg.validate()
        self.assertFalse(ok)


class TestWatermarkEnginePosition(unittest.TestCase):
    """水印位置计算单元测试."""

    def setUp(self):
        self.out_w = 1920
        self.out_h = 1080
        self.wm_w = 200
        self.wm_h = 100
        self.mx = 20
        self.my = 20

    def test_top_left(self):
        x, y = WatermarkEngine.calc_position("top_left", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
        self.assertEqual(x, 20)
        self.assertEqual(y, 20)

    def test_top_center(self):
        x, y = WatermarkEngine.calc_position(
            "top_center", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, (1920 - 200) // 2)
        self.assertEqual(y, 20)

    def test_top_right(self):
        x, y = WatermarkEngine.calc_position(
            "top_right", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, 1920 - 200 - 20)
        self.assertEqual(y, 20)

    def test_center_left(self):
        x, y = WatermarkEngine.calc_position(
            "center_left", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, 20)
        self.assertEqual(y, (1080 - 100) // 2)

    def test_center(self):
        x, y = WatermarkEngine.calc_position("center", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
        self.assertEqual(x, (1920 - 200) // 2)
        self.assertEqual(y, (1080 - 100) // 2)

    def test_center_right(self):
        x, y = WatermarkEngine.calc_position(
            "center_right", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, 1920 - 200 - 20)
        self.assertEqual(y, (1080 - 100) // 2)

    def test_bottom_left(self):
        x, y = WatermarkEngine.calc_position(
            "bottom_left", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, 20)
        self.assertEqual(y, 1080 - 100 - 20)

    def test_bottom_center(self):
        x, y = WatermarkEngine.calc_position(
            "bottom_center", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, (1920 - 200) // 2)
        self.assertEqual(y, 1080 - 100 - 20)

    def test_bottom_right(self):
        x, y = WatermarkEngine.calc_position(
            "bottom_right", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my
        )
        self.assertEqual(x, 1920 - 200 - 20)
        self.assertEqual(y, 1080 - 100 - 20)

    def test_default_fallback(self):
        """非法位置默认右下角."""
        x, y = WatermarkEngine.calc_position("unknown", self.out_w, self.out_h, self.wm_w, self.wm_h, self.mx, self.my)
        self.assertEqual(x, 1920 - 200 - 20)
        self.assertEqual(y, 1080 - 100 - 20)

    def test_nine_positions_all_present(self):
        """9宫格位置都有定义."""
        self.assertEqual(len(WATERMARK_POSITIONS), 9)


class TestWatermarkEngineFilters(unittest.TestCase):
    """水印滤镜构建单元测试."""

    def test_text_watermark_filter(self):
        """文字水印滤镜构建."""
        cfg = WatermarkConfig(
            mode="text",
            text="hello",
            position="top_left",
            font_size=24,
            font_color="white",
            opacity=0.8,
            margin_x=10,
            margin_y=10,
        )
        result = WatermarkEngine.build_text_watermark_filter("[in]", "[out]", cfg, 1920, 1080)
        self.assertTrue(result.startswith("[in]drawtext="))
        self.assertIn("text='hello'", result)
        self.assertIn("fontsize=24", result)
        self.assertIn("fontcolor=white@0.8", result)
        self.assertTrue(result.endswith("[out]"))

    def test_text_watermark_scroll(self):
        """滚动文字水印."""
        cfg = WatermarkConfig(
            mode="text",
            text="scroll",
            position="bottom_left",
            scroll=True,
            scroll_speed=60,
        )
        result = WatermarkEngine.build_text_watermark_filter("[in]", "[out]", cfg, 1920, 1080)
        self.assertIn("mod(60*t", result)


class TestIntroOutroConfig(unittest.TestCase):
    """IntroOutroConfig 单元测试."""

    def test_from_dict_disabled(self):
        """未启用 → 空配置."""
        cfg = IntroOutroConfig.from_dict(None)
        self.assertFalse(cfg.enabled)
        self.assertFalse(cfg.has_intro)
        self.assertFalse(cfg.has_outro)

    def test_from_dict_intro_text(self):
        """文字片头配置."""
        cfg = IntroOutroConfig.from_dict(
            {
                "enabled": True,
                "intro": {
                    "type": "text",
                    "title": "欢迎观看",
                    "subtitle": "精彩内容马上开始",
                    "duration": 3.0,
                    "background": "#1a1a2e",
                },
            }
        )
        self.assertTrue(cfg.enabled)
        self.assertTrue(cfg.has_intro)
        self.assertFalse(cfg.has_outro)
        self.assertEqual(cfg.intro_type, "text")
        self.assertEqual(cfg.intro_title, "欢迎观看")
        self.assertEqual(cfg.intro_duration, 3.0)

    def test_from_dict_outro_video(self):
        """视频片尾配置."""
        cfg = IntroOutroConfig.from_dict(
            {
                "enabled": True,
                "outro": {
                    "type": "video",
                    "video_path": "/tmp/outro.mp4",
                    "duration": 5.0,
                },
            }
        )
        self.assertTrue(cfg.has_outro)
        self.assertEqual(cfg.outro_type, "video")
        self.assertEqual(cfg.outro_video_path, "/tmp/outro.mp4")

    def test_validate_valid(self):
        """合法配置."""
        cfg = IntroOutroConfig(
            enabled=True,
            intro_type="text",
            intro_title="标题",
            intro_duration=3.0,
            outro_type="text",
            outro_title="片尾",
            outro_duration=3.0,
        )
        ok, err = cfg.validate()
        self.assertTrue(ok)

    def test_validate_video_intro_missing_path(self):
        """视频片头缺路径."""
        cfg = IntroOutroConfig(
            enabled=True,
            intro_type="video",
            intro_duration=3.0,
        )
        ok, err = cfg.validate()
        self.assertFalse(ok)
        self.assertIn("video_path", err)

    def test_validate_text_intro_missing_title(self):
        """文字片头缺标题."""
        cfg = IntroOutroConfig(
            enabled=True,
            intro_type="text",
            intro_duration=3.0,
        )
        ok, err = cfg.validate()
        self.assertFalse(ok)

    def test_has_intro_false_when_none(self):
        """type=none 时 has_intro 为 False."""
        cfg = IntroOutroConfig(enabled=True, intro_type="none")
        self.assertFalse(cfg.has_intro)

    def test_has_outro_follow_type(self):
        """follow 类型也算有片尾."""
        cfg = IntroOutroConfig(enabled=True, outro_type="follow", outro_title="关注")
        self.assertTrue(cfg.has_outro)


class TestIntroOutroEngineConcat(unittest.TestCase):
    """片头片尾拼接单元测试."""

    def test_concat_no_intro_outro(self):
        """没有片头片尾 → 直接复制."""
        import tempfile

        with tempfile.TemporaryDirectory() as tmpdir:
            main_video = Path(tmpdir) / "main.mp4"
            output = Path(tmpdir) / "output.mp4"
            # 创建空文件模拟
            main_video.write_bytes(b"fake video data")

            result = IntroOutroEngine.concat_with_intro_outro(main_video, None, None, output)
            self.assertTrue(result)
            self.assertTrue(output.exists())
            self.assertEqual(main_video.read_bytes(), output.read_bytes())

    def test_concat_intro_only_no_file(self):
        """只有片头但文件不存在 → 直接复制主视频."""
        import tempfile

        with tempfile.TemporaryDirectory() as tmpdir:
            main_video = Path(tmpdir) / "main.mp4"
            output = Path(tmpdir) / "output.mp4"
            main_video.write_bytes(b"fake data")

            # intro 路径不存在
            intro = Path(tmpdir) / "nonexistent.mp4"

            result = IntroOutroEngine.concat_with_intro_outro(main_video, intro, None, output)
            self.assertTrue(result)
            self.assertTrue(output.exists())


class TestResolveWatermarkConfig(unittest.TestCase):
    """UnifiedRenderService._resolve_watermark_config 兼容性测试."""

    def _resolve(self, plan_config):
        from video_processing.unified_render_service import UnifiedRenderService

        return UnifiedRenderService._resolve_watermark_config(plan_config)

    def test_none_or_empty_config(self):
        """空配置 → None."""
        self.assertIsNone(self._resolve(None))
        self.assertIsNone(self._resolve({}))
        self.assertIsNone(self._resolve([]))  # 非dict安全处理

    def test_nested_format_enabled(self):
        """嵌套格式 config.watermark 正常解析."""
        cfg = {"watermark": {"enabled": True, "mode": "text", "text": "测试水印"}}
        result = self._resolve(cfg)
        self.assertIsNotNone(result)
        self.assertEqual(result.mode, "text")
        self.assertEqual(result.text, "测试水印")

    def test_nested_format_disabled(self):
        """嵌套格式未启用 → None."""
        cfg = {"watermark": {"enabled": False, "mode": "text", "text": "测试"}}
        self.assertIsNone(self._resolve(cfg))

    def test_flat_export_format_enabled(self):
        """扁平格式 config.export.watermark_enabled + text 正常解析."""
        cfg = {"export": {"watermark_enabled": True, "watermark_text": "我的水印"}}
        result = self._resolve(cfg)
        self.assertIsNotNone(result)
        self.assertEqual(result.mode, "text")
        self.assertEqual(result.text, "我的水印")
        self.assertEqual(result.position, "bottom_right")

    def test_flat_export_format_disabled(self):
        """扁平格式未启用 → None."""
        cfg = {"export": {"watermark_enabled": False, "watermark_text": "测试"}}
        self.assertIsNone(self._resolve(cfg))

    def test_flat_export_format_no_text(self):
        """扁平格式启用但无文字 → None."""
        cfg = {"export": {"watermark_enabled": True, "watermark_text": ""}}
        self.assertIsNone(self._resolve(cfg))

    def test_nested_takes_priority(self):
        """嵌套格式存在时优先使用嵌套格式（忽略扁平格式）."""
        cfg = {
            "watermark": {"enabled": True, "mode": "text", "text": "嵌套水印"},
            "export": {"watermark_enabled": True, "watermark_text": "扁平水印"},
        }
        result = self._resolve(cfg)
        self.assertIsNotNone(result)
        self.assertEqual(result.text, "嵌套水印")

    def test_flat_with_custom_position(self):
        """扁平格式支持自定义位置、透明度等参数."""
        cfg = {
            "export": {
                "watermark_enabled": True,
                "watermark_text": "自定义水印",
                "watermark_position": "top_left",
                "watermark_opacity": 0.5,
                "watermark_font_size": 32,
                "watermark_font_color": "red",
            }
        }
        result = self._resolve(cfg)
        self.assertIsNotNone(result)
        self.assertEqual(result.position, "top_left")
        self.assertAlmostEqual(result.opacity, 0.5)
        self.assertEqual(result.font_size, 32)
        self.assertEqual(result.font_color, "red")

    def test_no_export_key(self):
        """没有 export 字段时不报错."""
        cfg = {"other": "value"}
        self.assertIsNone(self._resolve(cfg))


if __name__ == "__main__":
    unittest.main()