xiaoxia-saas/tests/unit/test_1754_visual_perturbation.py

"""#1754 渲染期画面扰动——让批量视频本身更不同。

测试覆盖：
1. generate_visual_perturbation 生成合法参数
2. _build_filter_complex 中视觉扰动滤镜正确应用
3. _render_pass_through 中视觉扰动滤镜正确应用
4. ensure_variant_plans 为每个变体生成并存储扰动参数
"""

from __future__ import annotations

import random
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# ── 1. generate_visual_perturbation 单元测试 ──────────────────────────────────


class TestGenerateVisualPerturbation:
    """测试 generate_visual_perturbation 函数。"""

    def test_returns_dict_with_expected_keys(self):
        """返回 dict 包含 hflip/zoom_ratio/speed_factor/brightness_shift。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        result = generate_visual_perturbation(random.Random(42))
        assert set(result.keys()) == {"hflip", "zoom_ratio", "speed_factor", "brightness_shift"}

    def test_hflip_is_bool(self):
        """hflip 是 bool 类型。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        result = generate_visual_perturbation(random.Random(42))
        assert isinstance(result["hflip"], bool)

    def test_zoom_ratio_range(self):
        """zoom_ratio 在 1.0~1.08 范围内。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        for seed in range(100):
            result = generate_visual_perturbation(random.Random(seed))
            assert 1.0 <= result["zoom_ratio"] <= 1.08, f"seed={seed}: {result['zoom_ratio']}"

    def test_speed_factor_range(self):
        """speed_factor 在 0.95~1.05 范围内。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        for seed in range(100):
            result = generate_visual_perturbation(random.Random(seed))
            assert 0.95 <= result["speed_factor"] <= 1.05, f"seed={seed}: {result['speed_factor']}"

    def test_brightness_shift_values(self):
        """brightness_shift 只能取特定值。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        allowed = {-10, -5, 0, 5, 10}
        for seed in range(100):
            result = generate_visual_perturbation(random.Random(seed))
            assert result["brightness_shift"] in allowed, f"seed={seed}: {result['brightness_shift']}"

    def test_different_seeds_produce_different_results(self):
        """不同种子产生不同扰动参数。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        results = [generate_visual_perturbation(random.Random(seed)) for seed in range(20)]
        # 至少有一些不同的 zoom_ratio
        zooms = {r["zoom_ratio"] for r in results}
        assert len(zooms) > 1, "所有种子产生了相同的 zoom_ratio"

    def test_hflip_probability_approximately_30_percent(self):
        """hflip 概率约 30%（大样本验证）。"""
        from packages.domain.variant_plan_selector import generate_visual_perturbation

        n = 1000
        flips = sum(1 for seed in range(n) if generate_visual_perturbation(random.Random(seed))["hflip"])
        ratio = flips / n
        assert 0.20 <= ratio <= 0.40, f"hflip 比例 {ratio:.2%} 偏离 30%"


# ── 2. 渲染侧滤镜应用测试 ────────────────────────────────────────────────────


class TestVisualPerturbationFilters:
    """测试 _get_visual_perturbation 和滤镜应用方法。"""

    def _make_render_service(self, perturbation: dict | None = None):
        """构造一个带 mock plan 的 UnifiedRenderService 实例。"""
        # 避免实际初始化
        from video_processing.unified_render_service import UnifiedRenderService

        svc = object.__new__(UnifiedRenderService)
        svc.output_width = 1920
        svc.output_height = 1080
        svc.plan = MagicMock()
        svc.plan.config = {"visual_perturbation": perturbation} if perturbation else {}
        return svc

    def test_get_visual_perturbation_empty_when_no_config(self):
        """plan.config 无 visual_perturbation 时返回空 dict。"""
        svc = self._make_render_service(None)
        assert svc._get_visual_perturbation() == {}

    def test_get_visual_perturbation_parses_values(self):
        """正确解析 perturbation dict 中的各字段。"""
        perturbation = {
            "hflip": True,
            "zoom_ratio": 1.05,
            "speed_factor": 0.97,
            "brightness_shift": 5,
        }
        svc = self._make_render_service(perturbation)
        result = svc._get_visual_perturbation()
        assert result["hflip"] is True
        assert result["zoom_ratio"] == 1.05
        assert result["speed_factor"] == 0.97
        assert result["brightness_shift"] == 5

    def test_get_visual_perturbation_clamps_zoom_ratio(self):
        """zoom_ratio 被限制在 1.0~1.2 范围内。"""
        svc = self._make_render_service({"zoom_ratio": 2.0})
        assert svc._get_visual_perturbation()["zoom_ratio"] == 1.2

        svc2 = self._make_render_service({"zoom_ratio": 0.5})
        assert svc2._get_visual_perturbation()["zoom_ratio"] == 1.0

    def test_get_visual_perturbation_clamps_brightness(self):
        """brightness_shift 被限制在 -30~30 范围内。"""
        svc = self._make_render_service({"brightness_shift": 50})
        assert svc._get_visual_perturbation()["brightness_shift"] == 30

        svc2 = self._make_render_service({"brightness_shift": -50})
        assert svc2._get_visual_perturbation()["brightness_shift"] == -30

    def test_pre_scale_hflip(self):
        """hflip=True 时 filters 列表追加 hflip。"""
        svc = self._make_render_service({"hflip": True})
        filters: list[str] = []
        svc._apply_visual_perturbation_pre_scale(filters, svc._get_visual_perturbation())
        assert "hflip" in filters

    def test_pre_scale_no_hflip(self):
        """hflip=False 时 filters 列表不变。"""
        svc = self._make_render_service({"hflip": False})
        filters: list[str] = ["existing_filter"]
        svc._apply_visual_perturbation_pre_scale(filters, svc._get_visual_perturbation())
        assert filters == ["existing_filter"]

    def test_post_scale_zoom_and_brightness(self):
        """zoom_ratio>1 时追加 scale+crop，brightness!=0 时追加 eq。"""
        svc = self._make_render_service({"zoom_ratio": 1.05, "brightness_shift": 10})
        perturbation = svc._get_visual_perturbation()
        filters: list[str] = []
        svc._apply_visual_perturbation_post_scale(filters, perturbation)
        # zoom: scale=2016:1134 (1920*1.05=2016, 1080*1.05=1134)
        assert "scale=2016:1134" in filters
        assert "crop=1920:1080" in filters
        # brightness: 10/100 = 0.1
        assert "eq=brightness=0.100" in filters

    def test_post_scale_no_perturbation_when_defaults(self):
        """zoom_ratio=1.0 且 brightness_shift=0 时不追加滤镜。"""
        svc = self._make_render_service({"zoom_ratio": 1.0, "brightness_shift": 0})
        perturbation = svc._get_visual_perturbation()
        filters: list[str] = []
        svc._apply_visual_perturbation_post_scale(filters, perturbation)
        assert filters == []


# ── 3. ensure_variant_plans 扰动注入测试 ──────────────────────────────────────


class TestEnsureVariantPlansPerturbation:
    """测试 ensure_variant_plans 为每个变体注入视觉扰动。"""

    def test_variant_plans_get_perturbation_in_config(self):
        """每个变体 plan 的 config 中包含 visual_perturbation。"""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        svc = EditPlanService(db)

        # Mock 所有依赖方法
        mock_plan0 = MagicMock()
        mock_plan0.id = "plan-0"
        mock_variant = MagicMock()
        mock_variant.id = "plan-1"

        svc.clone_plan_for_variant = MagicMock(return_value=mock_plan0)
        svc.reselect_plan_for_variant = MagicMock(return_value=mock_variant)
        svc.apply_voice_duration_to_plan = MagicMock()
        svc.mark_clips_ready = MagicMock()
        svc.update_plan_config = MagicMock()

        rng = random.Random(42)
        plan_ids = svc.ensure_variant_plans(
            "source-plan-id",
            count=3,
            candidate_asset_ids=["asset-1", "asset-2", "asset-3"],
            rng=rng,
        )

        assert len(plan_ids) == 3

        # update_plan_config 应该被调用 3 次（每个变体一次）
        assert svc.update_plan_config.call_count == 3

        # 检查每次调用都传入了 visual_perturbation
        for call in svc.update_plan_config.call_args_list:
            args, kwargs = call
            plan_id = args[0]
            config_updates = args[1]
            assert "visual_perturbation" in config_updates
            perturbation = config_updates["visual_perturbation"]
            assert "hflip" in perturbation
            assert "zoom_ratio" in perturbation
            assert "speed_factor" in perturbation
            assert "brightness_shift" in perturbation

    def test_variant_0_no_hflip(self):
        """变体 0（预览 plan）不做 hflip。"""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        svc = EditPlanService(db)

        mock_plan0 = MagicMock()
        mock_plan0.id = "plan-0"
        mock_variant = MagicMock()
        mock_variant.id = "plan-1"

        svc.clone_plan_for_variant = MagicMock(return_value=mock_plan0)
        svc.reselect_plan_for_variant = MagicMock(return_value=mock_variant)
        svc.apply_voice_duration_to_plan = MagicMock()
        svc.mark_clips_ready = MagicMock()
        svc.update_plan_config = MagicMock()

        rng = random.Random(42)
        svc.ensure_variant_plans(
            "source-plan-id",
            count=2,
            candidate_asset_ids=["asset-1", "asset-2"],
            rng=rng,
        )

        # 第一次调用（变体 0）的 hflip 应该是 False
        first_call_args = svc.update_plan_config.call_args_list[0]
        perturbation_0 = first_call_args[0][1]["visual_perturbation"]
        assert perturbation_0["hflip"] is False