xiaoxia-saas/tests/unit/test_preview_generation_fixes.py

"""Tests for preview generation fixes: URL signing, duration capping, resolution."""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# ── Fix 1: URL 签名 ──────────────────────────────────────────────────────────


class TestSignVideoUrl:
    """_sign_video_url 单元测试。"""

    def test_empty_url_returns_empty(self):
        """空 URL 直接返回空字符串。"""
        from app.api.routes.generation_preview import _sign_video_url

        assert _sign_video_url("") == ""

    def test_signs_oss_url(self):
        """OSS URL 应被签名。"""
        from app.api.routes.generation_preview import _sign_video_url

        mock_storage = MagicMock()
        mock_storage.get_download_url.return_value = "https://signed-url.example.com/video.mp4?sig=abc"

        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
            result = _sign_video_url("https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/video.mp4")

        assert result == "https://signed-url.example.com/video.mp4?sig=abc"
        mock_storage.get_download_url.assert_called_once_with(
            "https://bucket.oss-cn-hangzhou.aliyuncs.com/generated/video.mp4",
            expires_seconds=7200,
        )

    def test_fallback_on_sign_failure(self):
        """签名失败时降级返回原始 URL。"""
        from app.api.routes.generation_preview import _sign_video_url

        mock_storage = MagicMock()
        mock_storage.get_download_url.side_effect = Exception("OSS not configured")

        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")

        assert result == "https://bucket.oss.example.com/video.mp4"

    def test_fallback_on_storage_error(self):
        """get_storage_service 抛异常时降级返回原始 URL。"""
        from app.api.routes.generation_preview import _sign_video_url

        with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")):
            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")

        assert result == "https://bucket.oss.example.com/video.mp4"

    def test_sign_returns_none_fallback(self):
        """get_download_url 返回 None 时降级返回原始 URL。"""
        from app.api.routes.generation_preview import _sign_video_url

        mock_storage = MagicMock()
        mock_storage.get_download_url.return_value = None

        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")

        assert result == "https://bucket.oss.example.com/video.mp4"


# ── Fix 2: 模板 segment 时长约束 ─────────────────────────────────────────────


class TestLoadTemplateSegmentDurations:
    """_load_template_segment_durations 单元测试。"""

    def test_empty_template_id_returns_empty(self):
        """空 template_id 返回空列表。"""
        import os
        import sys

        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "apps", "worker"))
        from worker_app.tasks.generation import _load_template_segment_durations

        assert _load_template_segment_durations("") == []

    def test_loads_durations_ordered(self):
        """按 segment_order 排序返回 duration_max 列表。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        mock_segment1 = MagicMock()
        mock_segment1.duration_max = 5.0
        mock_segment2 = MagicMock()
        mock_segment2.duration_max = 8.0
        mock_segment3 = MagicMock()
        mock_segment3.duration_max = 3.0

        mock_query = MagicMock()
        mock_query.filter.return_value.order_by.return_value.all.return_value = [
            mock_segment1,
            mock_segment2,
            mock_segment3,
        ]

        mock_session = MagicMock()
        mock_session.query.return_value = mock_query

        mock_session_local = MagicMock(return_value=mock_session)

        with patch("worker_app.tasks.generation.SessionLocal", mock_session_local):
            result = _load_template_segment_durations("tpl_123")

        assert result == [5.0, 8.0, 3.0]

    def test_filters_zero_durations(self):
        """duration_max <= 0 的 segment 被过滤。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        mock_seg_valid = MagicMock()
        mock_seg_valid.duration_max = 5.0
        mock_seg_zero = MagicMock()
        mock_seg_zero.duration_max = 0.0
        mock_seg_neg = MagicMock()
        mock_seg_neg.duration_max = -1.0

        mock_query = MagicMock()
        mock_query.filter.return_value.order_by.return_value.all.return_value = [
            mock_seg_valid,
            mock_seg_zero,
            mock_seg_neg,
        ]

        mock_session = MagicMock()
        mock_session.query.return_value = mock_query

        with patch("worker_app.tasks.generation.SessionLocal", MagicMock(return_value=mock_session)):
            result = _load_template_segment_durations("tpl_123")

        assert result == [5.0]

    def test_db_error_returns_empty(self):
        """数据库异常返回空列表，不抛出。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        with patch("worker_app.tasks.generation.SessionLocal", side_effect=Exception("DB down")):
            result = _load_template_segment_durations("tpl_123")

        assert result == []


class TestDurationCappingInBuildPlan:
    """_build_plan_and_clips_from_task 中时长约束的集成测试。"""

    def test_clips_capped_by_segment_max(self):
        """clip 时长超过 segment duration_max 时应被截断。"""
        # 创建临时假视频文件
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            paths = []
            for i in range(3):
                p = Path(tmpdir) / f"video_{i}.mp4"
                p.write_bytes(b"\x00" * 100)  # 假文件
                paths.append(p)

            # Mock probe_duration 返回很长的时长
            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
                # Mock segment durations: 5s, 4s, 3s
                with patch(
                    "worker_app.tasks.generation._load_template_segment_durations",
                    return_value=[5.0, 4.0, 3.0],
                ):
                    # Mock _load_template_clip_configs 返回空（跳过效果层）
                    with patch(
                        "worker_app.tasks.generation._load_template_clip_configs",
                        return_value=[],
                    ):
                        plan, clips, asset_map = _build_plan_and_clips_from_task(
                            task_id="test_task_123",
                            downloaded_paths=paths,
                            mode="one_take",
                            template_id="tpl_test",
                        )

            # 每个 clip 的时长应被截断到对应 segment 的 duration_max
            assert clips[0].duration == 5.0
            assert clips[1].duration == 4.0
            assert clips[2].duration == 3.0

    def test_clips_not_capped_when_under_max(self):
        """clip 时长小于 segment duration_max 时不截断。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            paths = [Path(tmpdir) / "video_0.mp4"]
            paths[0].write_bytes(b"\x00" * 100)

            with patch("worker_app.tasks.generation.probe_duration", return_value=3.0):
                with patch(
                    "worker_app.tasks.generation._load_template_segment_durations",
                    return_value=[5.0],
                ):
                    with patch(
                        "worker_app.tasks.generation._load_template_clip_configs",
                        return_value=[],
                    ):
                        plan, clips, asset_map = _build_plan_and_clips_from_task(
                            task_id="test_task_456",
                            downloaded_paths=paths,
                            mode="one_take",
                            template_id="tpl_test",
                        )

            # 3.0 < 5.0, 不应截断
            assert clips[0].duration == 3.0

    def test_no_capping_without_template(self):
        """无 template_id 时不截断。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            paths = [Path(tmpdir) / "video_0.mp4"]
            paths[0].write_bytes(b"\x00" * 100)

            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
                plan, clips, asset_map = _build_plan_and_clips_from_task(
                    task_id="test_task_789",
                    downloaded_paths=paths,
                    mode="one_take",
                    template_id="",
                )

            # 无模板，使用素材完整时长
            assert clips[0].duration == 30.0


# ── Fix 3: 预览分辨率 ────────────────────────────────────────────────────────


class TestPreviewResolution:
    """_render_video 预览分辨率逻辑测试。"""

    def test_preview_uses_passed_resolution(self):
        """is_preview=True 且有 resolution 参数时，使用传入的分辨率。"""
        import tempfile

        from worker_app.tasks.generation import _render_video

        # 验证逻辑：检查 _render_video 在 is_preview + resolution 时的行为
        # 由于 _render_video 内部会调用 RenderAdapter，这里只验证分辨率配置逻辑
        # 通过 mock 掉渲染部分，检查 export_cfg

        with tempfile.TemporaryDirectory() as tmpdir:
            temp_path = Path(tmpdir)
            video_path = temp_path / "input.mp4"
            video_path.write_bytes(b"\x00" * 100)

            # Mock probe_duration
            with patch("worker_app.tasks.generation.probe_duration", return_value=5.0):
                with patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build:
                    mock_clip = MagicMock()
                    mock_clip.duration = 5.0
                    mock_clip.id = "vc_000"
                    mock_clip.plan_id = "test"
                    mock_clip.clip_type = "main"
                    mock_clip.order = 0
                    mock_clip.asset_id = "asset_0"
                    mock_clip.config = {}

                    mock_plan = MagicMock()
                    mock_plan.config = {}
                    mock_plan.id = "test"
                    mock_plan.name = "test"

                    mock_build.return_value = (mock_plan, [mock_clip], {"asset_0": video_path})

                    # Mock RenderAdapter
                    with patch("worker_app.tasks.generation.RenderAdapter") as MockAdapter:
                        mock_result = MagicMock()
                        mock_result.success = True
                        mock_result.output_path = temp_path / "output.mp4"
                        mock_result.output_path.write_bytes(b"\x00" * 100)
                        mock_result.duration = 5.0

                        mock_adapter_instance = MagicMock()
                        mock_adapter_instance.render_from_memory.return_value = mock_result
                        MockAdapter.return_value = mock_adapter_instance

                        with patch("worker_app.tasks.generation.SessionLocal"):
                            try:
                                _render_video(
                                    task_id="test_resolution",
                                    downloaded_videos=[video_path],
                                    voice_path=None,
                                    editing_mode=MagicMock(value="one_take"),
                                    project_id="",
                                    template_id="",
                                    user_id="",
                                    temp_path=temp_path,
                                    output_name="output.mp4",
                                    resolution="480x854",
                                    is_preview=True,
                                )
                            except Exception:
                                pass  # 可能会在其他地方失败，但我们只关心分辨率配置

                        # 检查传给 RenderAdapter 的 plan.config 中的分辨率
                        if mock_adapter_instance.render_from_memory.called:
                            call_args = mock_adapter_instance.render_from_memory.call_args
                            plan_arg = call_args.kwargs.get("plan") or call_args[1].get("plan")
                            if plan_arg and hasattr(plan_arg, "config"):
                                export = (plan_arg.config or {}).get("export", {})
                                assert export.get("resolution") == "480x854"

    def test_preview_defaults_to_landscape_when_no_resolution(self):
        """is_preview=True 且无 resolution 参数时，默认 854x480。"""
        import tempfile

        from worker_app.tasks.generation import _render_video

        with tempfile.TemporaryDirectory() as tmpdir:
            temp_path = Path(tmpdir)
            video_path = temp_path / "input.mp4"
            video_path.write_bytes(b"\x00" * 100)

            with patch("worker_app.tasks.generation.probe_duration", return_value=5.0):
                with patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build:
                    mock_clip = MagicMock()
                    mock_clip.duration = 5.0
                    mock_clip.config = {}

                    mock_plan = MagicMock()
                    mock_plan.config = {}
                    mock_plan.id = "test"
                    mock_plan.name = "test"

                    mock_build.return_value = (mock_plan, [mock_clip], {"": video_path})

                    with patch("worker_app.tasks.generation.RenderAdapter") as MockAdapter:
                        mock_result = MagicMock()
                        mock_result.success = True
                        mock_result.output_path = temp_path / "output.mp4"
                        mock_result.output_path.write_bytes(b"\x00" * 100)
                        mock_result.duration = 5.0

                        mock_adapter_instance = MagicMock()
                        mock_adapter_instance.render_from_memory.return_value = mock_result
                        MockAdapter.return_value = mock_adapter_instance

                        with patch("worker_app.tasks.generation.SessionLocal"):
                            try:
                                _render_video(
                                    task_id="test_resolution_default",
                                    downloaded_videos=[video_path],
                                    voice_path=None,
                                    editing_mode=MagicMock(value="one_take"),
                                    project_id="",
                                    template_id="",
                                    user_id="",
                                    temp_path=temp_path,
                                    output_name="output.mp4",
                                    resolution="",
                                    is_preview=True,
                                )
                            except Exception:
                                pass

                        if mock_adapter_instance.render_from_memory.called:
                            call_args = mock_adapter_instance.render_from_memory.call_args
                            plan_arg = call_args.kwargs.get("plan") or call_args[1].get("plan")
                            if plan_arg and hasattr(plan_arg, "config"):
                                export = (plan_arg.config or {}).get("export", {})
                                assert export.get("resolution") == "854x480"


class TestCalcPreviewResolution:
    """_calc_preview_resolution 单元测试。"""

    def test_portrait_9_16(self):
        from app.api.routes.generation_preview import _calc_preview_resolution

        assert _calc_preview_resolution("9:16") == "480x854"

    def test_landscape_16_9(self):
        from app.api.routes.generation_preview import _calc_preview_resolution

        assert _calc_preview_resolution("16:9") == "854x480"

    def test_square_1_1(self):
        from app.api.routes.generation_preview import _calc_preview_resolution

        assert _calc_preview_resolution("1:1") == "480x480"

    def test_unknown_defaults_to_landscape(self):
        from app.api.routes.generation_preview import _calc_preview_resolution

        assert _calc_preview_resolution("unknown") == "854x480"

    def test_empty_defaults_to_landscape(self):
        from app.api.routes.generation_preview import _calc_preview_resolution

        assert _calc_preview_resolution("") == "854x480"