xiaoxia-saas/tests/unit/test_points_rules.py

"""积分消耗规则单元测试 (#1895) — v1.6.2: 仅保留 voice_clone 相关"""

from __future__ import annotations

import math

import pytest

from packages.domain.points_rules import (
    FREE_USER_MULTIPLIER,
    MEMBER_DISCOUNT,
    MEMBERSHIP_PRICES,
    POINTS_PACKAGES,
    POINTS_SCENES,
    calculate_points_cost,
)


class TestPointsScenesConfig:
    """场景配置完整性"""

    def test_registered_scenes_include_voice_clone_and_viral_video(self):
        """场景配置：包含声音克隆（训练/合成）+ 爆款视频（动态定价）。"""
        assert {"voice_clone_train", "voice_clone_synth", "viral_video"}.issubset(set(POINTS_SCENES.keys()))

    def test_viral_video_scene_is_dynamic_with_zero_base(self):
        """viral_video 必须注册但 base_points=0 且 dynamic=True，不使用 @points_gate。"""
        vv = POINTS_SCENES["viral_video"]
        assert vv["base_points"] == 0
        assert vv["dynamic"] is True
        assert vv["unit"] == "次"
        assert vv["name"] == "爆款视频"

    def test_voice_clone_scenes_defined(self):
        # 保留声音克隆两个场景
        assert "voice_clone_train" in POINTS_SCENES
        assert "voice_clone_synth" in POINTS_SCENES

    def test_required_keys_present(self):
        for key, scene in POINTS_SCENES.items():
            assert "base_points" in scene, f"{key} missing base_points"
            assert "unit" in scene, f"{key} missing unit"
            assert "name" in scene, f"{key} missing name"

    def test_voice_clone_train_is_free(self):
        assert POINTS_SCENES["voice_clone_train"]["base_points"] == 0

    def test_voice_clone_synth_is_per_minute(self):
        assert POINTS_SCENES["voice_clone_synth"]["base_points"] == 1
        assert POINTS_SCENES["voice_clone_synth"]["unit"] == "分钟"

    def test_calculate_points_cost_returns_zero_for_dynamic_viral_video(self):
        """calculate_points_cost 对动态场景 viral_video 必须返回 0（由业务侧手动计算）。"""
        assert calculate_points_cost("viral_video", is_member=False) == 0.0
        assert calculate_points_cost("viral_video", is_member=True, member_type="monthly") == 0.0


class TestPointsPackages:
    def test_three_packages(self):
        assert len(POINTS_PACKAGES) == 3
        assert POINTS_PACKAGES["starter_pack"]["points"] == 100
        assert POINTS_PACKAGES["basic_pack"]["points"] == 500
        assert POINTS_PACKAGES["pro_pack"]["points"] == 2000


class TestMembershipPrices:
    def test_three_plans(self):
        assert len(MEMBERSHIP_PRICES) == 3
        assert MEMBERSHIP_PRICES["monthly"]["price_cents"] == 1990
        assert MEMBERSHIP_PRICES["quarterly"]["duration_days"] == 90
        assert MEMBERSHIP_PRICES["yearly"]["price_cents"] == 15900


class TestCalculatePointsCost:
    """核心计费逻辑"""

    # ── 声音克隆合成（按时长计费） ──

    def test_voice_clone_synth_base(self):
        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=3)
        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)

    def test_voice_clone_synth_rounds_up(self):
        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=2.3)
        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)

    def test_voice_clone_synth_minimum_1_minute(self):
        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=0.1)
        assert cost == math.ceil(1 * FREE_USER_MULTIPLIER)

    # ── 免费场景 ──

    def test_voice_clone_train_free(self):
        cost = calculate_points_cost("voice_clone_train", is_member=False)
        assert cost == 0

    def test_voice_clone_train_free_for_member(self):
        cost = calculate_points_cost("voice_clone_train", is_member=True)
        assert cost == 0

    # ── 会员折扣 ──

    def test_monthly_member_discount(self):
        cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1, member_type="monthly")
        assert cost == max(1, math.floor(1 * MEMBER_DISCOUNT["monthly"]))

    def test_yearly_member_deep_discount(self):
        cost = calculate_points_cost(
            "voice_clone_synth",
            is_member=True,
            duration_minutes=2,
            member_type="yearly",
        )
        assert cost == max(1, math.floor(2 * MEMBER_DISCOUNT["yearly"]))

    def test_member_without_type_no_discount(self):
        cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1)
        assert cost == 1

    # ── 已下线/未知场景（向后兼容：返回 0） ──

    @pytest.mark.parametrize(
        "scene",
        [
            "ai_voice",
            "ai_video",
            "ai_digital_human",
            "ai_rewrite",
            "ai_cover",
            "ai_title",
            "douyin_extract",
            "nonexistent",
        ],
    )
    def test_retired_scenes_return_zero(self, scene):
        assert calculate_points_cost(scene, is_member=False) == 0
        assert calculate_points_cost(scene, is_member=True, duration_minutes=10) == 0


# ============ 爆款视频动态定价 (#2151) ============


class TestResolveVideoDimensions:
    """resolve_video_dimensions(): 分辨率别名、比例、默认兜底。"""

    def test_1080p_16_9(self):
        """1080p + 16:9 → w=1920, h=1080。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("1080p", "16:9")
        assert (w, h) == (1920, 1080)

    def test_480p_16_9(self):
        """480p + 16:9 → h=480, w 按 16//9 计算。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("480p", "16:9")
        assert h == 480
        assert w == 480 * 16 // 9

    def test_720p_1_1(self):
        """1:1 正方形 → w == h。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("720p", "1:1")
        assert (w, h) == (720, 720)

    def test_1080p_1_1(self):
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("1080p", "1:1")
        assert (w, h) == (1080, 1080)

    def test_resolution_aliases(self):
        """中文/英文别名应正确映射到对应高度。"""
        from packages.domain.points_rules import resolve_video_dimensions

        cases = [
            ("普清", 480),
            ("sd", 480),
            ("low", 480),
            ("default", 480),
            ("高清", 720),
            ("medium", 720),
            ("hd", 720),
            ("超清", 1080),
            ("fhd", 1080),
            ("ultra", 1080),
            ("全能", 1080),
            ("high", 1080),
        ]
        for alias, expected_h in cases:
            _, h = resolve_video_dimensions(alias, "1:1")
            assert h == expected_h, f"{alias} -> h={h}, expected {expected_h}"

    def test_unknown_resolution_falls_back_to_720p(self):
        """未知分辨率字符串兜底到 720p。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("2160p", "1:1")
        assert h == 720
        assert w == 720

    def test_empty_resolution_defaults_to_720p_9_16(self):
        """空 resolution + 空 ratio → 默认 720p + 9:16。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("", "")
        assert h == 720
        assert w == 720 * 9 // 16

    def test_none_resolution_default_ratio(self):
        """None resolution + None ratio → 720p + 9:16 默认。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions(None, None)
        assert h == 720
        assert w == 720 * 9 // 16

    def test_whitespace_resolution_case_insensitive(self):
        """前后空格 + 大写应被规范化处理。"""
        from packages.domain.points_rules import resolve_video_dimensions

        w, h = resolve_video_dimensions("  1080P  ", " 16:9 ")
        assert (w, h) == (1920, 1080)


class TestMatchModelPrefix:
    """_match_model_prefix() 前缀匹配 + 兜底。"""

    def test_seedance_2_5_exact(self):
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("seedance-2.5") == "seedance-2.5"

    def test_seedance_2_5_with_variant(self):
        """带后缀版本号（如 seedance-2.5-pro）仍匹配 seedance-2.5。"""
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("seedance-2.5-pro") == "seedance-2.5"

    def test_seedance_2_0_exact(self):
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("seedance-2.0") == "seedance-2.0"

    def test_seedance_2_0_with_variant(self):
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("seedance-2.0-lite") == "seedance-2.0"

    def test_unknown_model_falls_back_to_2_5(self):
        """未知模型前缀兜底 seedance-2.5。"""
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("kling-v1") == "seedance-2.5"
        assert _match_model_prefix("") == "seedance-2.5"
        assert _match_model_prefix(None) == "seedance-2.5"

    def test_case_insensitive(self):
        from packages.domain.points_rules import _match_model_prefix

        assert _match_model_prefix("SEEDANCE-2.0") == "seedance-2.0"


class TestInferResolutionKey:
    """_infer_resolution_key(): 1000+/650-999/<650 三个分支。"""

    def test_height_ge_1000_is_1080p(self):
        from packages.domain.points_rules import _infer_resolution_key

        assert _infer_resolution_key(1000) == "1080p"
        assert _infer_resolution_key(1080) == "1080p"
        assert _infer_resolution_key(2160) == "1080p"

    def test_height_650_to_999_is_720p(self):
        from packages.domain.points_rules import _infer_resolution_key

        assert _infer_resolution_key(650) == "720p"
        assert _infer_resolution_key(720) == "720p"
        assert _infer_resolution_key(999) == "720p"

    def test_height_lt_650_is_480p(self):
        from packages.domain.points_rules import _infer_resolution_key

        assert _infer_resolution_key(480) == "480p"
        assert _infer_resolution_key(649) == "480p"
        assert _infer_resolution_key(0) == "480p"


class TestCalculateViralVideoCredits:
    """calculate_viral_video_credits()：爆款视频动态定价核心函数。"""

    def test_default_args_returns_float(self):
        """默认参数返回 float。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        credits = calculate_viral_video_credits(15, 1280, 720)
        assert isinstance(credits, float)

    def test_return_is_rounded_to_two_decimals(self):
        """round(..., 2) 后值本身就是两位小数（再 round 不变化）。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        for dur, w, h in [(15, 1280, 720), (5, 854, 480), (30, 1920, 1080), (10, 720, 720)]:
            credits = calculate_viral_video_credits(dur, w, h)
            assert round(credits, 2) == credits

    def test_has_video_input_uses_lower_price(self):
        """has_video_input=True 时使用参考视频价格（有视频输入便宜）。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        no_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=False)
        with_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=True)
        assert with_input < no_input

    def test_unknown_model_falls_back_to_seedance_2_5(self):
        """未知 model 前缀兜底到 seedance-2.5 价格，与默认等价。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        unknown = calculate_viral_video_credits(15, 1280, 720, model="unknown-model")
        default = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5")
        assert unknown == default

    def test_actual_tokens_overrides_calculation(self):
        """传入 actual_tokens>0 时用它替代公式计算的 tokens。"""
        from packages.domain.points_rules import (
            VIRAL_VIDEO_FIXED_COST,
            VIRAL_VIDEO_MODEL_PRICES,
            VIRAL_VIDEO_PROFIT_MULTIPLIER,
            calculate_viral_video_credits,
        )

        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
        actual_tokens = 2_000_000
        expected = round(
            (actual_tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2
        )
        credits = calculate_viral_video_credits(15, 1280, 720, actual_tokens=actual_tokens)
        assert credits == expected

    def test_zero_duration_width_height_defensive_max1(self):
        """duration/width/height 为 0/None 时 max(1,...) 防御，结果>0。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        c_zero = calculate_viral_video_credits(0, 0, 0)
        assert c_zero > 0
        c_none = calculate_viral_video_credits(None, None, None)
        assert c_none > 0
        c_one = calculate_viral_video_credits(1, 1, 1)
        assert c_none == c_one

    def test_non_default_fps_affects_tokens(self):
        """fps 非默认值(30) 应比默认(24) 积分高。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        c24 = calculate_viral_video_credits(15, 1280, 720, fps=24)
        c30 = calculate_viral_video_credits(15, 1280, 720, fps=30)
        assert c30 > c24

    def test_seedance_2_0_priced_lower_than_2_5_at_1080p(self):
        """seedance-2.0 在 1080p 无视频输入时定价低于 seedance-2.5。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        c20 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.0", has_video_input=False)
        c25 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.5", has_video_input=False)
        assert c20 < c25

    def test_formula_includes_fixed_cost_and_multiplier(self):
        """手算公式结果应与函数返回一致（固定成本 + 利润系数）。"""
        from packages.domain.points_rules import (
            VIRAL_VIDEO_FIXED_COST,
            VIRAL_VIDEO_FPS,
            VIRAL_VIDEO_MODEL_PRICES,
            VIRAL_VIDEO_PROFIT_MULTIPLIER,
            calculate_viral_video_credits,
        )

        dur, w, h = 10, 1280, 720
        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
        tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
        expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
        assert calculate_viral_video_credits(dur, w, h) == expected

    def test_seedance_2_0_with_video_input_falls_back_to_seedance_2_5_price(self):
        """seedance-2.0 + has_video_input=True 组合不在价格表，走 line 111 fallback 到 seedance-2.5 的 720p False 价格。"""
        from packages.domain.points_rules import (
            VIRAL_VIDEO_FIXED_COST,
            VIRAL_VIDEO_FPS,
            VIRAL_VIDEO_MODEL_PRICES,
            VIRAL_VIDEO_PROFIT_MULTIPLIER,
            calculate_viral_video_credits,
        )

        dur, w, h = 10, 1280, 720
        # 兜底价格 = seedance-2.5/720p/False = 70.0
        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
        assert price == 70.0
        tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
        expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
        credits = calculate_viral_video_credits(dur, w, h, model="seedance-2.0", has_video_input=True)
        assert credits == expected

    def test_fps_zero_or_none_falls_back_to_default(self):
        """fps=0/None 时 int(fps or 24) 兜底到默认 24，结果与 fps=24 一致。"""
        from packages.domain.points_rules import calculate_viral_video_credits

        c_default = calculate_viral_video_credits(10, 1280, 720, fps=24)
        c_zero = calculate_viral_video_credits(10, 1280, 720, fps=0)
        c_none = calculate_viral_video_credits(10, 1280, 720, fps=None)
        assert c_zero == c_default
        assert c_none == c_default


class TestViralVideoCreditsWithBreakdown:
    """calculate_viral_video_credits_with_breakdown：返回 (credits, breakdown_dict)。"""

    def test_returns_credits_matching_plain_version(self):
        """新函数返回的 credits 必须与 calculate_viral_video_credits 完全一致，且 breakdown 字段齐全。"""
        from packages.domain.points_rules import (
            calculate_viral_video_credits,
            calculate_viral_video_credits_with_breakdown,
        )

        for dur, w, h, model, hvi in [
            (15, 1280, 720, "seedance-2.5", False),
            (10, 720, 1280, "seedance-2.0", False),
            (30, 1920, 1080, "seedance-2.5", False),
            (5, 480, 480, "", False),
        ]:
            c1 = calculate_viral_video_credits(dur, w, h, model=model, has_video_input=hvi)
            c2, bd = calculate_viral_video_credits_with_breakdown(dur, w, h, model=model, has_video_input=hvi)
            assert c1 == c2
            assert isinstance(bd, dict)
            for key in (
                "tokens",
                "video_cost",
                "fixed_cost",
                "profit_multiplier",
                "model_price",
                "width",
                "height",
                "fps",
            ):
                assert key in bd, f"breakdown missing key: {key}"
            assert bd["fixed_cost"] == 0.15
            assert bd["profit_multiplier"] == 1.3
            assert bd["width"] == w
            assert bd["height"] == h
            assert bd["fps"] == 24
            assert bd["tokens"] > 0
            assert bd["model_price"] > 0
            expected = round((bd["video_cost"] + bd["fixed_cost"]) * bd["profit_multiplier"], 2)
            assert expected == c2

    def test_actual_tokens_overrides_computed(self):
        """actual_tokens 传入时应覆盖按公式计算的 tokens。"""
        from packages.domain.points_rules import calculate_viral_video_credits_with_breakdown

        c, bd = calculate_viral_video_credits_with_breakdown(
            15,
            1280,
            720,
            actual_tokens=1_000_000,
        )
        assert bd["tokens"] == 1_000_000.0
        # video_cost = 1M/1M * 70 = 70; total = (70+0.15)*1.3 = 91.195 → 91.20
        assert c == 91.20