xiaoxia-saas/tests/unit/test_feature_pricing_service.py

"""feature_pricing_service 单元测试。

覆盖：
- 300s TTL 内存缓存（命中不重复 load / 过期重新 load / refresh 强制刷新）
- calculate_price 公式 (dynamic+fixed)*multiplier、price_cap 封顶、round
- disabled / 未知 key 返回 0
- DB 异常 / 空表 → 内置兜底配置（爆款启用且价格与现状一致）
- lookup_model_price 嵌套/扁平结构与旧版回落语义
"""

from __future__ import annotations

import time

import pytest

from packages.domain import feature_pricing_service as fps
from packages.domain.feature_pricing_service import (
    CACHE_TTL_SECONDS,
    FeatureConfig,
    calculate_price,
    get_feature_config,
    is_feature_enabled,
    lookup_model_price,
    refresh_feature_configs,
)


@pytest.fixture(autouse=True)
def _reset_cache():
    """每个用例前后清空模块缓存，避免相互污染。"""
    refresh_feature_configs()
    yield
    refresh_feature_configs()


def _cfg(key="x", **kw) -> FeatureConfig:
    base = dict(
        feature_key=key,
        name=key,
        is_enabled=True,
        fixed_cost=0.2,
        profit_multiplier=2.0,
        dynamic_unit_cost=0.0,
        billing_mode="per_second",
        price_cap=0.0,
        model_pricing={},
        description="",
    )
    base.update(kw)
    return FeatureConfig(**base)


class TestCacheTTL:
    def test_cache_hit_avoids_reload(self, monkeypatch):
        """TTL 内第二次读取不再调 _load_all。"""
        calls = {"n": 0}

        def fake_load():
            calls["n"] += 1
            return {"x": _cfg()}

        monkeypatch.setattr(fps, "_load_all", fake_load)
        get_feature_config("x")
        get_feature_config("x")
        get_feature_config("x")
        assert calls["n"] == 1

    def test_expired_cache_reloads(self, monkeypatch):
        """超过 TTL 后重新 load。"""
        calls = {"n": 0}

        def fake_load():
            calls["n"] += 1
            return {"x": _cfg()}

        monkeypatch.setattr(fps, "_load_all", fake_load)
        get_feature_config("x")
        assert calls["n"] == 1

        # 把缓存时间戳回拨到 TTL 之前
        ts, data = fps._cache
        fps._cache = (ts - CACHE_TTL_SECONDS - 1, data)
        get_feature_config("x")
        assert calls["n"] == 2

    def test_refresh_forces_reload(self, monkeypatch):
        calls = {"n": 0}

        def fake_load():
            calls["n"] += 1
            return {"x": _cfg()}

        monkeypatch.setattr(fps, "_load_all", fake_load)
        get_feature_config("x")
        refresh_feature_configs()
        get_feature_config("x")
        assert calls["n"] == 2

    def test_ttl_constant_is_300(self):
        assert CACHE_TTL_SECONDS == 300.0


class TestCalculatePrice:
    def test_basic_formula(self, monkeypatch):
        # (dynamic 1.0 + fixed 0.2) * 2.0 = 2.4
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(dynamic_unit_cost=1.0)})
        price, bd = calculate_price("x", dynamic_cost=1.0)
        assert price == 2.4
        assert bd["dynamic_cost"] == 1.0
        assert bd["fixed_cost"] == 0.2
        assert bd["profit_multiplier"] == 2.0
        assert bd["final_price"] == 2.4
        assert bd["charged"] is True

    def test_price_cap_clamps(self, monkeypatch):
        # raw = (1+0.2)*2 = 2.4，cap=1.0 → 1.0
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(price_cap=1.0)})
        price, bd = calculate_price("x", dynamic_cost=1.0)
        assert price == 1.0
        assert bd["price_cap"] == 1.0

    def test_no_cap_keeps_raw(self, monkeypatch):
        # cap=0 视为不封顶
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(price_cap=0.0)})
        price, _ = calculate_price("x", dynamic_cost=1.0)
        assert price == 2.4

    def test_rounded_two_decimals(self, monkeypatch):
        monkeypatch.setattr(
            fps,
            "_load_all",
            lambda: {"x": _cfg(fixed_cost=0.1, profit_multiplier=1.0)},
        )
        price, _ = calculate_price("x", dynamic_cost=1.0 / 3.0)
        # 0.3333... + 0.1 = 0.4333 → 0.43
        assert price == 0.43

    def test_negative_dynamic_treated_as_zero(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg()})
        price, _ = calculate_price("x", dynamic_cost=-5.0)
        # (0 + 0.2) * 2 = 0.4
        assert price == 0.4

    def test_disabled_returns_zero(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(is_enabled=False)})
        price, bd = calculate_price("x", dynamic_cost=1.0)
        assert price == 0.0
        assert bd["is_enabled"] is False
        assert bd["charged"] is False

    def test_unknown_key_returns_zero(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg()})
        price, bd = calculate_price("nope", dynamic_cost=1.0)
        assert price == 0.0
        assert bd["charged"] is False


class TestDBFailureFallback:
    def test_load_exception_uses_fallback(self, monkeypatch):
        def boom():
            raise RuntimeError("table does not exist")

        monkeypatch.setattr(fps, "_load_all", boom)
        cfg = get_feature_config("viral_video")
        assert cfg is not None
        assert cfg.is_enabled is True
        assert cfg.fixed_cost == 0.15
        assert cfg.profit_multiplier == 1.3

    def test_empty_table_uses_fallback(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {})
        assert get_feature_config("viral_video").is_enabled is True
        assert get_feature_config("lip_sync").is_enabled is False
        assert get_feature_config("smart_edit").is_enabled is False

    def test_fallback_viral_price_matches_current(self, monkeypatch):
        """兜底爆款价格与旧硬编码现状一致：seedance-2.5/720p/false=70。"""
        monkeypatch.setattr(fps, "_load_all", lambda: {})
        from packages.domain.points_rules import calculate_viral_video_credits

        # 默认全局开关关闭，但纯计费函数价格照常算
        assert calculate_viral_video_credits(15, 1280, 720) == 29.68

    def test_db_row_overrides_fallback(self, monkeypatch):
        monkeypatch.setattr(
            fps,
            "_load_all",
            lambda: {"viral_video": _cfg("viral_video", fixed_cost=0.5, profit_multiplier=2.0, price_cap=50.0)},
        )
        cfg = get_feature_config("viral_video")
        assert cfg.fixed_cost == 0.5
        assert cfg.profit_multiplier == 2.0
        assert cfg.price_cap == 50.0


class TestIsFeatureEnabled:
    def test_disabled_feature(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(is_enabled=False)})
        assert is_feature_enabled("x") is False

    def test_global_switch_off_blocks_enabled_feature(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(is_enabled=True)})
        monkeypatch.setattr(fps, "_global_points_enabled", lambda: False)
        assert is_feature_enabled("x") is False

    def test_both_switches_on(self, monkeypatch):
        monkeypatch.setattr(fps, "_load_all", lambda: {"x": _cfg(is_enabled=True)})
        monkeypatch.setattr(fps, "_global_points_enabled", lambda: True)
        assert is_feature_enabled("x") is True


class TestLookupModelPrice:
    NESTED = {
        "seedance-2.5": {
            "720p": {"false": 70.0, "true": 42.0},
        },
        "wan-3.0": {"480p": {"false": 0.3}},
    }

    def test_nested_exact_hit(self):
        assert lookup_model_price(self.NESTED, "seedance-2.5", "720p", False) == 70.0
        assert lookup_model_price(self.NESTED, "seedance-2.5", "720p", True) == 42.0

    def test_missing_bool_key_returns_none(self):
        # wan-3.0/480p 只有 false，请求 true → None（由调用方回落）
        assert lookup_model_price(self.NESTED, "wan-3.0", "480p", True) is None

    def test_unknown_model_returns_none(self):
        assert lookup_model_price(self.NESTED, "nope", "720p", False) is None

    def test_flat_structure(self):
        flat = {"m|720p|false": 12.5}
        assert lookup_model_price(flat, "m", "720p", False) == 12.5
        assert lookup_model_price(flat, "m", "720p", True) is None