group_fqcd_jr/tests/unit/service/test_customer_benefit_service.py

"""客户权益服务的定向测试：分层判定、累积展开、升级提示、鉴权。

权益条目由 `tools/seed_customer_benefits.py` 灌入（54 条）。本测试**不查库**，
用替身 session 直接给条目，保证分层与展开逻辑可独立验证。
"""

from decimal import Decimal
from typing import Any

import pytest

from app.core.contracts import RequestContext
from app.core.errors import ForbiddenAgentError
from app.model.benefit import CustomerBenefit
from app.service.customer_benefit_service import (
    TIERS,
    CustomerBenefitService,
    resolve_tier,
    tier_chain,
)


def benefit(code: str, tier: str, order: int) -> CustomerBenefit:
    return CustomerBenefit(
        id=order, benefit_code=code, customer_tier=tier, category="financial",
        name=f"{tier}-{order}", description="d", display_order=order,
        status="active", created_at=None, updated_at=None,  # type: ignore[arg-type]
    )


#: 与种子同一形状：每层条数不同，便于断言累积后的总数。
ALL: list[CustomerBenefit] = [
    *[benefit(f"g{i}", "gold", i) for i in range(1, 10)],        # 9
    *[benefit(f"p{i}", "platinum", 100 + i) for i in range(1, 12)],  # 11
    *[benefit(f"d{i}", "diamond", 200 + i) for i in range(1, 14)],   # 13
    *[benefit(f"v{i}", "private", 300 + i) for i in range(1, 22)],   # 21
]


class FakeSession:
    """只实现 `scalars`：把 `in_(codes)` 近似成"返回全部"，由服务侧过滤数量断言。"""

    def __init__(self, rows: list[CustomerBenefit]) -> None:
        self.rows = rows
        self.last_codes: tuple[str, ...] = ()

    async def scalars(self, statement: Any) -> list[CustomerBenefit]:
        # 从编译后的 SQL 参数里取层级码，保证"只取该层级及以下"确实生效。
        params = statement.compile().params
        codes = tuple(v for k, v in params.items() if "customer_tier" in str(k))
        flat: list[str] = []
        for c in codes:
            if isinstance(c, str):
                flat.append(c)
            else:
                flat.extend(str(x) for x in c)
        self.last_codes = tuple(flat)
        return [r for r in self.rows if r.customer_tier in self.last_codes]


# --- 分层判定 ---------------------------------------------------------------


@pytest.mark.parametrize(
    ("amount", "expected"),
    [
        (None, None),
        ("0", None),
        ("499999.99", None),      # 差 1 分不到金卡
        ("500000", "gold"),       # 门槛含等号
        ("1999999.99", "gold"),
        ("2000000", "platinum"),
        ("5999999", "platinum"),
        ("6000000", "diamond"),
        ("9999999", "diamond"),
        ("10000000", "private"),  # 1000 万整
        ("99999999", "private"),
    ],
)
def test_resolve_tier_boundaries(amount: str | None, expected: str | None) -> None:
    """门槛含等号、边界不外溢 —— 这类 off-by-one 在金额分层里最容易错。"""
    spec = resolve_tier(Decimal(amount) if amount is not None else None)
    assert (spec.code if spec else None) == expected


def test_tiers_are_ordered_high_to_low() -> None:
    """`TIERS` 必须从高到低：`resolve_tier` 取第一个命中，`tier_chain` 依赖该顺序。"""
    thresholds = [t.min_total_asset for t in TIERS]
    assert thresholds == sorted(thresholds, reverse=True)


def test_tier_chain_is_accumulating_and_ordered_low_to_high() -> None:
    assert tier_chain(next(t for t in TIERS if t.code == "gold")) == ("gold",)
    assert tier_chain(next(t for t in TIERS if t.code == "platinum")) == ("gold", "platinum")
    assert tier_chain(next(t for t in TIERS if t.code == "private")) == (
        "gold", "platinum", "diamond", "private",
    )


# --- 累积展开 ---------------------------------------------------------------


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("amount", "tier", "count"),
    [
        ("500000", "gold", 9),
        ("2000000", "platinum", 20),    # 9 + 11
        ("6000000", "diamond", 33),     # 9 + 11 + 13
        ("10000000", "private", 54),    # 9 + 11 + 13 + 21
    ],
)
async def test_benefits_accumulate_with_tier(amount: str, tier: str, count: int) -> None:
    """文档写明"含全部下级权益，新增以下" ⇒ 高等级必须拿到低等级的全部条目。"""
    session = FakeSession(ALL)
    data = await CustomerBenefitService(session).entitlements(  # type: ignore[arg-type]
        total_asset=Decimal(amount)
    )
    assert data["tier"] == tier
    assert len(data["benefits"]) == count
    # 低等级条目必须在场（累积的直接证据）
    assert any(b["tier"] == "gold" for b in data["benefits"])


@pytest.mark.asyncio
async def test_below_lowest_threshold_gets_no_benefits_but_keeps_upgrade_hint() -> None:
    """低于 50 万是"普通客户"，不是错误：权益空，但仍告诉他要多少才升级。"""
    session = FakeSession(ALL)
    data = await CustomerBenefitService(session).entitlements(  # type: ignore[arg-type]
        total_asset=Decimal("100")
    )
    assert data["tier"] is None
    assert data["tier_label"] == "普通客户"
    assert data["benefits"] == []
    assert data["next_tier"] == {
        "tier": "gold", "tier_label": "金卡", "min_total_asset": "500000.00",
    }
    # 未达门槛时不应去查权益表
    assert session.last_codes == ()


@pytest.mark.asyncio
async def test_top_tier_has_no_next_tier() -> None:
    data = await CustomerBenefitService(FakeSession(ALL)).entitlements(  # type: ignore[arg-type]
        total_asset=Decimal("50000000")
    )
    assert data["tier"] == "private"
    assert data["next_tier"] is None


# --- 鉴权 -------------------------------------------------------------------


class FakeCtx:
    def __init__(self, permissions: set[str]) -> None:
        self.user_id = "9102"
        self.permissions = permissions
        self.roles: set[str] = set()
        self.portal = "api"


@pytest.mark.asyncio
async def test_missing_permission_is_denied_before_touching_data(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """权限检查必须在读取任何客户数据**之前**（避免"拒绝请求却已经查了库"）。"""
    service = CustomerBenefitService(FakeSession(ALL))  # type: ignore[arg-type]
    called = {"n": 0}

    async def spy(customer_id: int) -> Decimal | None:
        called["n"] += 1
        return Decimal("10000000")

    monkeypatch.setattr(service, "_total_asset", spy)

    ctx = RequestContext.model_construct(
        user_id=9102, permissions=frozenset(), roles=frozenset(),
        portal="api", trace_id="t", request_id="r",
    )
    with pytest.raises(ForbiddenAgentError):
        await service.entitlements_for(ctx)
    assert called["n"] == 0