feat(invite): 被邀请人列表接口 + 指纹归因(任务3) (#31)

- GET /invitees: 分页 + 名字降级兜底(昵称→微信昵称→脱敏手机号)
- 指纹归因: 落地页采集 + (IP,设备型号)反查撞库 + 时间窗口闸
- test_invite: +5 个列表测试(脱敏/倒序/昵称优先/分页/空), 修指纹测试跨用例串味

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Reviewed-on: #31
Co-authored-by: xiebing <xiebing@wonderable.ai>
Co-committed-by: xiebing <xiebing@wonderable.ai>
This commit was merged in pull request #31.
This commit is contained in:
xiebing
2026-06-09 21:48:48 +08:00
committed by marco
parent a828b51d9f
commit b2f5a53dd8
11 changed files with 777 additions and 23 deletions
+218 -1
View File
@@ -1,4 +1,4 @@
"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽。
"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽、指纹兜底归因
用 sms mock 登录拿 token(同 test_welfare),再跑邀请闭环。
"""
@@ -6,12 +6,16 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from app.core.rewards import (
INVITE_FP_WINDOW_DAYS,
INVITE_INVITEE_COINS,
INVITE_INVITER_COINS,
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
@@ -158,3 +162,216 @@ def test_old_user_not_eligible(client) -> None:
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"
# 双方各发金币
assert _coin_balance(client, a) == INVITE_INVITER_COINS
assert _coin_balance(client, b) == INVITE_INVITEE_COINS
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)
names = {it["display_name"] for it in body["items"]}
assert names == {"138****2051", "138****2052"}
assert all(it["avatar_url"] is None for it in body["items"])
assert all(it["coins"] == INVITE_INVITER_COINS 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()
items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
assert items[0]["display_name"] == "138****2062" # C 刚邀,在前
assert items[1]["display_name"] == "138****2061" # B 2 小时前,在后
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