shaguabijia-app-server/tests/test_invite.py

"""好友邀请测试:邀请码、绑定(双方不发钱)、幂等、自邀/无效码屏蔽、指纹兜底归因。

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

from datetime import datetime, timedelta, timezone

from sqlalchemy import select

from app.core.rewards import (
    INVITE_FP_WINDOW_DAYS,
    INVITE_NEW_USER_WINDOW_HOURS,
)
from app.db.session import SessionLocal
from app.models.invite_fingerprint import InviteFingerprint
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_no_coins_either_side(client) -> None:
    """v3:B 用 A 的码绑定 → 双方都不发钱(被邀请人无奖励、邀请人改比价后发现金)。"""
    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"] == 0

    # 绑定后双方金币都不增(邀请人收益改走"好友比价发 2 元邀请奖励金")
    assert _coin_balance(client, b) == 0
    assert _coin_balance(client, a) == 0

    # A 的战绩:已邀 1 人;金币口径收益恒 0(邀请人收益走邀请奖励金,见 try_reward_on_compare)
    r = client.get("/api/v1/invite/me", headers=_auth(a))
    assert r.json()["invited_count"] == 1
    assert r.json()["coins_earned"] == 0


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


# =====================================================================
# 任务 3:指纹归因(剪贴板兜底,见 brain [[invite-three-tasks]])
# =====================================================================

def test_landing_track_records_fingerprint(client) -> None:
    """落地页 POST /landing-track 成功存指纹(无需鉴权)。"""
    a = _login(client, "13800002040")
    a_code = _my_code(client, a)
    # 浏览器无 token 调
    r = client.post(
        "/api/v1/invite/landing-track",
        json={"ref": a_code, "screen": "1080x2400"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "ok"


def test_landing_track_invalid_code(client) -> None:
    """无效邀请码 → invalid_code(silent 返,不影响落地页下载流程)。"""
    r = client.post(
        "/api/v1/invite/landing-track",
        json={"ref": "ZZZZZZ", "screen": "1080x2400"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "invalid_code"


def test_bind_by_fingerprint_success(client) -> None:
    """B 落地页存指纹 → B 登录后用指纹兜底绑定成功(模拟剪贴板被覆盖丢失)。"""
    a = _login(client, "13800002041")
    a_code = _my_code(client, a)

    # 1. B 浏览器(无 token)访问落地页 → 存指纹
    r1 = client.post(
        "/api/v1/invite/landing-track",
        json={"ref": a_code, "screen": "1234x5678"},  # 用独特 screen 避免跟其它测试撞
    )
    assert r1.json()["status"] == "ok"

    # 2. B 注册登录(剪贴板被覆盖,没拿到邀请码)
    b = _login(client, "13800002042")

    # 3. B 不带 invite_code,只带 fingerprint 走兜底
    r2 = client.post(
        "/api/v1/invite/bind",
        json={
            "channel": "fingerprint",
            "fingerprint": {"screen": "1234x5678", "device_model": ""},
        },
        headers=_auth(b),
    )
    assert r2.status_code == 200, r2.text
    assert r2.json()["status"] == "success"
    # v3:绑定双方都不发钱(被邀请人无奖励、邀请人改比价发现金)
    assert _coin_balance(client, a) == 0
    assert _coin_balance(client, b) == 0


def test_bind_by_fingerprint_not_found(client) -> None:
    """没有匹配的指纹记录 → fp_not_found,不发奖。"""
    b = _login(client, "13800002043")
    r = client.post(
        "/api/v1/invite/bind",
        json={
            "channel": "fingerprint",
            "fingerprint": {"screen": "0000x9999", "device_model": "no_such_model"},
        },
        headers=_auth(b),
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "fp_not_found"
    assert r.json()["coins_awarded"] == 0
    assert _coin_balance(client, b) == 0


def test_bind_by_fingerprint_outside_window(client) -> None:
    """指纹记录超出 INVITE_FP_WINDOW_DAYS 窗口 → fp_not_found。

    用独特设备型号隔离:测试环境所有请求 IP 都是 testclient、落地页默认无 UA(解析出的
    型号="")，而指纹反查按 (IP, 型号) 匹配 → 会串到别的测试残留的、没过期的 model=""
    指纹上(误判 success)。这里给落地页传一个独特 UA(解析出独特型号 OUTWIN9X),反查只可能
    命中本测试自己的指纹,"过期就反查不到"才验得准。
    """
    a = _login(client, "13800002044")
    a_code = _my_code(client, a)

    # 落地页存指纹(独特 UA → 独特型号 OUTWIN9X),再手动把 created_at 改到窗口外
    unique_screen = "2222x3333"
    unique_ua = "Mozilla/5.0 (Linux; Android 13; OUTWIN9X Build/TQ) AppleWebKit/537.36"
    r1 = client.post(
        "/api/v1/invite/landing-track",
        json={"ref": a_code, "screen": unique_screen},
        headers={"user-agent": unique_ua},
    )
    assert r1.json()["status"] == "ok"

    with SessionLocal() as db:
        fp = db.execute(
            select(InviteFingerprint).where(InviteFingerprint.screen == unique_screen)
        ).scalar_one()
        fp.created_at = datetime.now(timezone.utc) - timedelta(days=INVITE_FP_WINDOW_DAYS + 1)
        db.commit()

    b = _login(client, "13800002045")
    r = client.post(
        "/api/v1/invite/bind",
        json={
            "channel": "fingerprint",
            "fingerprint": {"screen": unique_screen, "device_model": "OUTWIN9X"},
        },
        headers=_auth(b),
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "fp_not_found"
    assert _coin_balance(client, a) == 0
    assert _coin_balance(client, b) == 0


def test_bind_no_code_no_fingerprint(client) -> None:
    """既没邀请码也没指纹 → invalid_code(无效请求)。"""
    b = _login(client, "13800002046")
    r = client.post(
        "/api/v1/invite/bind",
        json={"channel": "fingerprint"},  # 没 invite_code、没 fingerprint
        headers=_auth(b),
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "invalid_code"
    assert _coin_balance(client, b) == 0


# =====================================================================
# 被邀请人列表 GET /invitees(邀请页小窗 + 完整列表页共用)
# =====================================================================

def test_invitees_basic(client) -> None:
    """A 邀 B、C → 列表返 2 条、total=2、名字=被邀请人默认昵称(创建即有)、头像 null、金币对。"""
    a = _login(client, "13800002050")
    a_code = _my_code(client, a)
    for phone in ("13800002051", "13800002052"):
        u = _login(client, phone)
        client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(u))

    r = client.get("/api/v1/invite/invitees", headers=_auth(a))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["total"] == 2
    assert body["has_more"] is False
    assert len(body["items"]) == 2
    # 创建即分配默认昵称 → 名字=被邀请人昵称(取实际值对比)、头像 null(不依赖顺序用 set)
    with SessionLocal() as db:
        expected_names = {
            get_user_by_phone(db, "13800002051").nickname,
            get_user_by_phone(db, "13800002052").nickname,
        }
    names = {it["display_name"] for it in body["items"]}
    assert names == expected_names
    assert all(it["avatar_url"] is None for it in body["items"])
    # v2:绑定时邀请人那份为 0(收益改"好友比价才发奖"),故每条 coins 字段=0
    assert all(it["coins"] == 0 for it in body["items"])


def test_invitees_order_desc(client) -> None:
    """按邀请时间倒序:最近邀请的在最前(手动拉开时间,避开 SQLite 秒级精度)。"""
    from app.models.invite import InviteRelation
    a = _login(client, "13800002060")
    a_code = _my_code(client, a)
    b = _login(client, "13800002061")
    c = _login(client, "13800002062")
    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(c))
    with SessionLocal() as db:
        ub = get_user_by_phone(db, "13800002061")
        rel_b = db.execute(
            select(InviteRelation).where(InviteRelation.invitee_user_id == ub.id)
        ).scalar_one()
        rel_b.created_at = datetime.now(timezone.utc) - timedelta(hours=2)
        db.commit()
        nick_b = ub.nickname
        nick_c = get_user_by_phone(db, "13800002062").nickname
    items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
    # 创建即有默认昵称 → display_name=被邀请人昵称;本用例核心验证倒序(C 刚邀在前、B 2h 前在后)
    assert items[0]["display_name"] == nick_c
    assert items[1]["display_name"] == nick_b


def test_invitees_nickname_priority(client) -> None:
    """设了昵称的被邀请人 → 名字用昵称(优先于脱敏手机号)。"""
    a = _login(client, "13800002070")
    a_code = _my_code(client, a)
    b = _login(client, "13800002071")
    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
    with SessionLocal() as db:
        u = get_user_by_phone(db, "13800002071")
        u.nickname = "小明"
        db.commit()
    items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
    assert items[0]["display_name"] == "小明"


def test_invitees_pagination(client) -> None:
    """分页:limit=1 → 1 条 + has_more=True;offset=1 → 第 2 条 + has_more=False。"""
    a = _login(client, "13800002080")
    a_code = _my_code(client, a)
    for phone in ("13800002081", "13800002082"):
        u = _login(client, phone)
        client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(u))

    p1 = client.get("/api/v1/invite/invitees?limit=1&offset=0", headers=_auth(a)).json()
    assert len(p1["items"]) == 1 and p1["total"] == 2 and p1["has_more"] is True
    p2 = client.get("/api/v1/invite/invitees?limit=1&offset=1", headers=_auth(a)).json()
    assert len(p2["items"]) == 1 and p2["has_more"] is False


def test_invitees_empty_and_auth(client) -> None:
    """没邀请人 → 空列表;不带 token → 401。"""
    a = _login(client, "13800002090")
    body = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()
    assert body["total"] == 0 and body["items"] == [] and body["has_more"] is False
    assert client.get("/api/v1/invite/invitees").status_code == 401