shaguabijia-app-server/tests/test_welfare.py

"""福利模块测试:钱包 / 签到 / 一次性任务。

用 sms mock 登录拿 token,再跑各接口闭环。
"""
from __future__ import annotations

from app.core.rewards import (
    COIN_PER_CENT,
    COIN_PER_YUAN,
    MIN_EXCHANGE_COIN,
    SIGNIN_REWARDS,
    TASK_ENABLE_NOTIFICATION,
    TASK_REWARDS,
)
from app.db.session import SessionLocal
from app.repositories import wallet as crud_wallet
from app.repositories.user import get_user_by_phone


def _login(client, phone: str) -> str:
    """sms mock 登录,返回 access_token。"""
    client.post("/api/v1/auth/sms/send", json={"phone": phone})
    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
    assert r.status_code == 200, r.text
    return r.json()["access_token"]


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def test_account_auto_created_empty(client) -> None:
    """首次访问账户:自动建空账户,余额全 0。"""
    token = _login(client, "13800001001")
    r = client.get("/api/v1/wallet/account", headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body == {"coin_balance": 0, "cash_balance_cents": 0, "total_coin_earned": 0}


def test_signin_flow(client) -> None:
    """签到状态 → 签到 → 余额到账 → 重复签到 409 → 流水有记录。"""
    token = _login(client, "13800001002")

    # 1. 签到前状态:未签、可签、今天是第 1 档
    r = client.get("/api/v1/signin/status", headers=_auth(token))
    assert r.status_code == 200, r.text
    st = r.json()
    assert st["today_signed"] is False
    assert st["can_claim"] is True
    assert st["today_cycle_day"] == 1
    assert st["today_coin"] == SIGNIN_REWARDS[0]
    assert len(st["steps"]) == len(SIGNIN_REWARDS)
    assert st["steps"][0]["status"] == "today"

    # 2. 执行签到,发第 1 档金币
    r = client.post("/api/v1/signin", headers=_auth(token))
    assert r.status_code == 200, r.text
    res = r.json()
    assert res["cycle_day"] == 1
    assert res["streak"] == 1
    assert res["coin_awarded"] == SIGNIN_REWARDS[0]
    assert res["coin_balance"] == SIGNIN_REWARDS[0]

    # 3. 账户余额到账
    r = client.get("/api/v1/wallet/account", headers=_auth(token))
    assert r.json()["coin_balance"] == SIGNIN_REWARDS[0]

    # 4. 状态变为已签
    r = client.get("/api/v1/signin/status", headers=_auth(token))
    st = r.json()
    assert st["today_signed"] is True
    assert st["can_claim"] is False
    assert st["consecutive_days"] == 1
    assert st["steps"][0]["status"] == "claimed"

    # 5. 重复签到 → 409
    r = client.post("/api/v1/signin", headers=_auth(token))
    assert r.status_code == 409

    # 6. 金币流水有一笔 signin 入账
    r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
    assert r.status_code == 200, r.text
    page = r.json()
    assert page["next_cursor"] is None
    assert len(page["items"]) == 1
    txn = page["items"][0]
    assert txn["amount"] == SIGNIN_REWARDS[0]
    assert txn["biz_type"] == "signin"
    assert txn["balance_after"] == SIGNIN_REWARDS[0]


def test_task_claim_flow(client) -> None:
    """任务列表 → 领"打开消息提醒" → 余额到账 → 重复领 409。"""
    token = _login(client, "13800001003")
    coin = TASK_REWARDS[TASK_ENABLE_NOTIFICATION]

    # 1. 任务列表:未领取
    r = client.get("/api/v1/tasks", headers=_auth(token))
    assert r.status_code == 200, r.text
    items = {it["task_key"]: it for it in r.json()["items"]}
    assert items[TASK_ENABLE_NOTIFICATION]["claimed"] is False
    assert items[TASK_ENABLE_NOTIFICATION]["coin"] == coin

    # 2. 领奖
    r = client.post(f"/api/v1/tasks/{TASK_ENABLE_NOTIFICATION}/claim", headers=_auth(token))
    assert r.status_code == 200, r.text
    assert r.json()["coin_awarded"] == coin
    assert r.json()["coin_balance"] == coin

    # 3. 重复领 → 409
    r = client.post(f"/api/v1/tasks/{TASK_ENABLE_NOTIFICATION}/claim", headers=_auth(token))
    assert r.status_code == 409

    # 4. 列表变为已领取
    r = client.get("/api/v1/tasks", headers=_auth(token))
    items = {it["task_key"]: it for it in r.json()["items"]}
    assert items[TASK_ENABLE_NOTIFICATION]["claimed"] is True

    # 5. 未知任务 → 404
    r = client.post("/api/v1/tasks/no_such_task/claim", headers=_auth(token))
    assert r.status_code == 404


def test_exchange_info(client) -> None:
    token = _login(client, "13800001004")
    r = client.get("/api/v1/wallet/exchange-info", headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["coin_per_yuan"] == COIN_PER_YUAN
    assert body["min_coin"] == MIN_EXCHANGE_COIN
    assert body["step_coin"] == COIN_PER_CENT


def test_exchange_flow(client) -> None:
    """先供款够兑换下限的金币 → 兑换 1 元 → 金币扣、现金加 → 现金流水有记录。

    打开消息提醒任务已降到 1000 金币(不再 = 兑换下限), 不能再靠领任务供款;
    直接 grant_coins 注入 MIN_EXCHANGE_COIN(= COIN_PER_YUAN = 10000)当种子。
    """
    phone = "13800001005"
    token = _login(client, phone)
    # 供款: 直接注入 1 元额度金币(替代原先靠 notification=10000 领任务供款)
    with SessionLocal() as db:
        user = get_user_by_phone(db, phone)
        assert user is not None
        crud_wallet.grant_coins(db, user.id, MIN_EXCHANGE_COIN, biz_type="test_seed", remark="测试供款")
        db.commit()

    # 兑换 10000 金币 → 100 分
    r = client.post(
        "/api/v1/wallet/exchange",
        json={"coin_amount": COIN_PER_YUAN},
        headers=_auth(token),
    )
    assert r.status_code == 200, r.text
    res = r.json()
    assert res["coin_amount"] == COIN_PER_YUAN
    assert res["cash_added_cents"] == 100
    assert res["coin_balance"] == 0
    assert res["cash_balance_cents"] == 100

    # 账户余额同步
    r = client.get("/api/v1/wallet/account", headers=_auth(token))
    assert r.json()["coin_balance"] == 0
    assert r.json()["cash_balance_cents"] == 100

    # 金币流水多了一笔 exchange_out(负)
    r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
    types = [t["biz_type"] for t in r.json()["items"]]
    assert "exchange_out" in types
    out_txn = next(t for t in r.json()["items"] if t["biz_type"] == "exchange_out")
    assert out_txn["amount"] == -COIN_PER_YUAN

    # 现金流水有一笔 exchange_in(正)
    r = client.get("/api/v1/wallet/cash-transactions", headers=_auth(token))
    page = r.json()
    assert len(page["items"]) == 1
    assert page["items"][0]["amount_cents"] == 100
    assert page["items"][0]["biz_type"] == "exchange_in"


def test_exchange_insufficient_and_invalid(client) -> None:
    token = _login(client, "13800001006")

    # 余额 0 → 兑换 1 元 → 409 余额不足
    r = client.post(
        "/api/v1/wallet/exchange",
        json={"coin_amount": COIN_PER_YUAN},
        headers=_auth(token),
    )
    assert r.status_code == 409

    # 低于最小额 → 400
    r = client.post(
        "/api/v1/wallet/exchange",
        json={"coin_amount": MIN_EXCHANGE_COIN - COIN_PER_CENT},
        headers=_auth(token),
    )
    assert r.status_code == 400

    # 非整分倍数(带零头)→ 400
    r = client.post(
        "/api/v1/wallet/exchange",
        json={"coin_amount": COIN_PER_YUAN + 1},
        headers=_auth(token),
    )
    assert r.status_code == 400


def test_savings_summary_and_battle(client) -> None:
    """首次访问触发 demo seeder,聚合接口返回真实算出的数字。"""
    token = _login(client, "13800001007")

    r = client.get("/api/v1/savings/summary", headers=_auth(token))
    assert r.status_code == 200, r.text
    s = r.json()
    assert s["order_count"] > 0
    assert s["total_saved_cents"] > 0
    # 平均每单 = 总额 // 单数
    assert s["avg_saved_cents"] == s["total_saved_cents"] // s["order_count"]

    r = client.get("/api/v1/savings/battle", headers=_auth(token))
    assert r.status_code == 200, r.text
    b = r.json()
    assert b["streak_days"] >= 1
    assert b["week_saved_cents"] >= 0
    assert 0 <= b["beat_percent"] <= 100


def test_savings_seeder_idempotent(client) -> None:
    """重复访问不应重复灌数据:两次 summary 完全一致。"""
    token = _login(client, "13800001008")
    first = client.get("/api/v1/savings/summary", headers=_auth(token)).json()
    second = client.get("/api/v1/savings/summary", headers=_auth(token)).json()
    assert first == second


def test_savings_records_pagination(client) -> None:
    token = _login(client, "13800001009")
    # 第一页
    r = client.get("/api/v1/savings/records?limit=5", headers=_auth(token))
    assert r.status_code == 200, r.text
    page1 = r.json()
    assert len(page1["items"]) == 5
    assert page1["next_cursor"] is not None
    # id 倒序
    ids = [it["id"] for it in page1["items"]]
    assert ids == sorted(ids, reverse=True)
    # 第二页接着游标走,不与第一页重叠
    r = client.get(
        f"/api/v1/savings/records?limit=5&cursor={page1['next_cursor']}",
        headers=_auth(token),
    )
    page2 = r.json()
    assert max(it["id"] for it in page2["items"]) < min(ids)


def test_endpoints_require_auth(client) -> None:
    """不带 token 的福利接口统一 401。"""
    assert client.get("/api/v1/wallet/account").status_code == 401
    assert client.get("/api/v1/signin/status").status_code == 401
    assert client.post("/api/v1/signin").status_code == 401
    assert client.get("/api/v1/tasks").status_code == 401
    assert client.get("/api/v1/wallet/cash-transactions").status_code == 401
    assert client.post("/api/v1/wallet/exchange", json={"coin_amount": 10000}).status_code == 401
    assert client.get("/api/v1/savings/summary").status_code == 401
    assert client.get("/api/v1/savings/battle").status_code == 401
    assert client.get("/api/v1/savings/records").status_code == 401