shaguabijia-app-server/tests/test_invite.py

"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽。

用 sms mock 登录拿 token(同 test_welfare),再跑邀请闭环。
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone

from app.core.rewards import (
    INVITE_INVITEE_COINS,
    INVITE_INVITER_COINS,
    INVITE_NEW_USER_WINDOW_HOURS,
)
from app.db.session import SessionLocal
from app.repositories.user import get_user_by_phone


def _login(client, phone: str) -> str:
    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 _coin_balance(client, token: str) -> int:
    r = client.get("/api/v1/wallet/account", headers=_auth(token))
    assert r.status_code == 200, r.text
    return r.json()["coin_balance"]


def _my_code(client, token: str) -> str:
    r = client.get("/api/v1/invite/me", headers=_auth(token))
    assert r.status_code == 200, r.text
    return r.json()["invite_code"]


def test_invite_me_returns_stable_code(client) -> None:
    """/invite/me 返回邀请码 + 分享链接;同一用户多次取码稳定。"""
    token = _login(client, "13800002001")
    r = client.get("/api/v1/invite/me", headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["invite_code"]                      # 非空
    assert f"ref={body['invite_code']}" in body["share_url"]
    assert body["invited_count"] == 0
    assert body["coins_earned"] == 0
    # 再取一次,码不变(懒生成 + 持久化)
    assert _my_code(client, token) == body["invite_code"]


def test_bind_flow_both_get_coins(client) -> None:
    """B 用 A 的码绑定 → 双方各得 1 万金币;A 战绩 +1。"""
    a = _login(client, "13800002002")
    b = _login(client, "13800002003")
    a_code = _my_code(client, a)

    assert _coin_balance(client, a) == 0
    assert _coin_balance(client, b) == 0

    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
    assert r.status_code == 200, r.text
    res = r.json()
    assert res["status"] == "success"
    assert res["coins_awarded"] == INVITE_INVITEE_COINS

    # 双方金币到账
    assert _coin_balance(client, b) == INVITE_INVITEE_COINS
    assert _coin_balance(client, a) == INVITE_INVITER_COINS

    # A 的战绩:已邀 1 人,累计获得 = 邀请人那份
    r = client.get("/api/v1/invite/me", headers=_auth(a))
    assert r.json()["invited_count"] == 1
    assert r.json()["coins_earned"] == INVITE_INVITER_COINS


def test_bind_idempotent_no_double_reward(client) -> None:
    """同一被邀请人二次绑定 → already_bound,不重复发奖。"""
    a = _login(client, "13800002004")
    b = _login(client, "13800002005")
    a_code = _my_code(client, a)

    r1 = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
    assert r1.json()["status"] == "success"
    bal_b = _coin_balance(client, b)
    bal_a = _coin_balance(client, a)

    # 再绑一次(甚至换个码也不行,已被绑过)
    c = _login(client, "13800002006")
    c_code = _my_code(client, c)
    r2 = client.post("/api/v1/invite/bind", json={"invite_code": c_code}, headers=_auth(b))
    assert r2.json()["status"] == "already_bound"
    assert r2.json()["coins_awarded"] == 0

    # 余额没变(没二次发奖),C 也没拿到邀请奖励
    assert _coin_balance(client, b) == bal_b
    assert _coin_balance(client, a) == bal_a
    assert _coin_balance(client, c) == 0


def test_self_invite_blocked(client) -> None:
    """填自己的邀请码 → self_invite,不发奖。"""
    a = _login(client, "13800002007")
    a_code = _my_code(client, a)
    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(a))
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "self_invite"
    assert _coin_balance(client, a) == 0


def test_invalid_code(client) -> None:
    """无效邀请码 → invalid_code,不发奖。"""
    b = _login(client, "13800002008")
    r = client.post("/api/v1/invite/bind", json={"invite_code": "ZZZZZZ"}, headers=_auth(b))
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "invalid_code"
    assert _coin_balance(client, b) == 0


def test_manual_channel_recorded(client) -> None:
    """channel=manual 也能绑定(手动填码兜底路径)。"""
    a = _login(client, "13800002009")
    b = _login(client, "13800002010")
    a_code = _my_code(client, a)
    r = client.post(
        "/api/v1/invite/bind",
        json={"invite_code": a_code, "channel": "manual"},
        headers=_auth(b),
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "success"


def test_invite_requires_auth(client) -> None:
    """不带 token → 401。"""
    assert client.get("/api/v1/invite/me").status_code == 401
    assert client.post("/api/v1/invite/bind", json={"invite_code": "ABCDEF"}).status_code == 401


def test_old_user_not_eligible(client) -> None:
    """新用户闸:被邀请人注册超过窗口 → not_eligible,双方都不发奖。"""
    a = _login(client, "13800002030")
    a_code = _my_code(client, a)
    b = _login(client, "13800002031")
    # 把 b 的注册时间改成超出窗口 → 变"老用户"
    with SessionLocal() as db:
        u = get_user_by_phone(db, "13800002031")
        assert u is not None
        u.created_at = datetime.now(timezone.utc) - timedelta(hours=INVITE_NEW_USER_WINDOW_HOURS + 1)
        db.commit()

    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "not_eligible"
    assert r.json()["coins_awarded"] == 0
    assert _coin_balance(client, b) == 0
    assert _coin_balance(client, a) == 0