xiaoxia-saas/tests/unit/test_1970_micro_transform_render.py

"""#1970 PR2 渲染服务微变换注入测试。

不做真实渲染，只验证 UnifiedRenderService 上微变换计划的开关、缓存、
滤镜注入与速度因子；纯参数生成在 test_1970_micro_transform_pure 覆盖。
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock

import pytest


def _make_service(plan_config: dict | None = None, clips=None):
    from video_processing.unified_render_service import UnifiedRenderService

    svc = object.__new__(UnifiedRenderService)
    svc.plan = MagicMock()
    svc.plan.config = plan_config or {}
    svc.plan.id = "plan-1"
    svc.plan.clips = clips or []
    svc._micro_plan_cache = None
    svc._micro_plan_loaded = False
    svc._clip_has_text = None
    return svc


class TestDedupGate:
    def test_default_enabled_when_config_missing(self):
        svc = _make_service({})
        assert svc._dedup_enabled() is True

    def test_explicit_true(self):
        svc = _make_service({"dedup_enabled": True})
        assert svc._dedup_enabled() is True

    def test_explicit_false(self):
        svc = _make_service({"dedup_enabled": False})
        assert svc._dedup_enabled() is False

    def test_plan_none_config_treated_enabled(self):
        svc = _make_service(None)
        svc.plan.config = None
        assert svc._dedup_enabled() is True


class TestPlanBuild:
    def test_disabled_returns_none_and_cached(self):
        svc = _make_service({"dedup_enabled": False, "generation_task_id": "t1"})
        assert svc._get_micro_transform_plan(5) is None
        # 第二次走缓存
        svc._dedup_enabled = MagicMock(side_effect=AssertionError("不应再次计算"))
        assert svc._get_micro_transform_plan(5) is None

    def test_zero_clips_returns_none(self):
        svc = _make_service({"generation_task_id": "t1"})
        assert svc._get_micro_transform_plan(0) is None

    def test_enabled_builds_reproducible_plan(self):
        cfg = {"generation_task_id": "task-abc", "video_index": 2, "bgm": {"enabled": True}}
        svc1 = _make_service(cfg)
        svc2 = _make_service(dict(cfg))
        p1 = svc1._get_micro_transform_plan(6)
        p2 = svc2._get_micro_transform_plan(6)
        assert p1 is not None and p2 is not None
        assert [c.speed for c in p1.clips] == [c.speed for c in p2.clips]
        assert p1.seed == p2.seed
        assert len(p1.clips) == 6

    def test_p1_conservative_no_hflip(self):
        svc = _make_service({"generation_task_id": "t1"})
        plan = svc._get_micro_transform_plan(30)
        assert all(not c.hflip for c in plan.clips)

    def test_no_bgm_config_zero_offset(self):
        svc = _make_service({"generation_task_id": "t1"})
        plan = svc._get_micro_transform_plan(3)
        assert plan.bgm_start_offset == 0.0

    def test_bgm_enabled_offset_in_range(self):
        svc = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}})
        plan = svc._get_micro_transform_plan(3)
        assert 2.0 <= plan.bgm_start_offset <= 8.0


class TestFilterInjection:
    def test_none_mt_noop(self):
        svc = _make_service({})
        from video_processing.unified_render_service import UnifiedRenderService

        filters = ["scale=100:100"]
        UnifiedRenderService._apply_micro_transform_video(filters, None)
        UnifiedRenderService._apply_micro_hflip(filters, None)
        assert filters == ["scale=100:100"]

    def test_eq_injection(self):
        svc = _make_service({"generation_task_id": "t1"})
        plan = svc._get_micro_transform_plan(1)
        mt = plan.clips[0]
        from video_processing.unified_render_service import UnifiedRenderService

        filters: list[str] = []
        UnifiedRenderService._apply_micro_transform_video(filters, mt)
        assert filters and filters[0].startswith("eq=brightness=")
        assert "contrast=" in filters[0] and "saturation=" in filters[0]

    def test_hflip_skipped_p1(self):
        svc = _make_service({"generation_task_id": "t1"})
        plan = svc._get_micro_transform_plan(10)
        from video_processing.unified_render_service import UnifiedRenderService

        for mt in plan.clips:
            filters: list[str] = []
            UnifiedRenderService._apply_micro_hflip(filters, mt)
            assert filters == []

    def test_speed_factor(self):
        from video_processing.micro_transform_pure import ClipMicroTransform
        from video_processing.unified_render_service import UnifiedRenderService

        assert UnifiedRenderService._micro_speed_factor(None) == 1.0
        assert UnifiedRenderService._micro_speed_factor(ClipMicroTransform(0, speed=1.025)) == pytest.approx(1.025)
        assert UnifiedRenderService._micro_speed_factor(MagicMock(speed=0.97)) == pytest.approx(0.97)

    def test_bgm_offset_reader_respects_flag(self):
        svc_off = _make_service({"dedup_enabled": False})
        assert svc_off._get_micro_bgm_offset() == 0.0

        svc_on = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}}, clips=[MagicMock()])
        off = svc_on._get_micro_bgm_offset()
        assert 2.0 <= off <= 8.0

    def test_bgm_offset_zero_without_bgm(self):
        svc = _make_service({"generation_task_id": "t1"}, clips=[MagicMock()])
        assert svc._get_micro_bgm_offset() == 0.0


class TestStreamCopyGate:
    """dedup 开启时微变换需要重编码，stream copy 必须被拒绝。"""

    def _build(self, config):
        from types import SimpleNamespace
        from unittest.mock import patch

        from video_processing.unified_render_service import ResolvedClip, UnifiedRenderService

        source = SimpleNamespace(
            id="c1",
            clip_type="main",
        )
        svc = object.__new__(UnifiedRenderService)
        svc.output_width = 1280
        svc.output_height = 720
        svc.output_fps = 25
        svc.plan = MagicMock()
        svc.plan.id = "plan-1"
        svc.plan.config = config
        svc.plan.clips = [source]
        svc.clips = [source]
        svc._micro_plan_cache = None
        svc._micro_plan_loaded = False
        svc._clip_has_text = None
        resolved = ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/a1.mp4"),
            clip_type="main",
            order=0,
        )
        info = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        return svc, resolved, info

    def test_dedup_enabled_blocks_stream_copy(self):
        from unittest.mock import patch

        svc, resolved, info = self._build({"dedup_enabled": True, "generation_task_id": "t1"})
        with patch("video_processing.unified_render_service.probe_video_info", return_value=info):
            can_copy, reason = svc._can_use_stream_copy(resolved)
        assert can_copy is False
        assert "微变换" in reason

    def test_dedup_disabled_allows_stream_copy(self):
        from unittest.mock import patch

        svc, resolved, info = self._build({"dedup_enabled": False})
        with patch("video_processing.unified_render_service.probe_video_info", return_value=info):
            can_copy, _ = svc._can_use_stream_copy(resolved)
        assert can_copy is True