lowmaster-chen
e1bd0e3ef7
改了什么：新增新版大盘日期窗口聚合、广告收益拆分、比价耗时字段、美团 CPS 拉单入库与大盘展示，并同步反馈审核和邀请奖励下线口径。 验证：python -m pytest tests/test_admin_read.py tests/test_admin_write.py tests/test_compare_record.py tests/test_invite.py tests/test_feedback.py -q。
376 lines
15 KiB
Python
376 lines
15 KiB
Python
"""好友邀请测试:邀请码、绑定、幂等、自邀/无效码屏蔽、指纹兜底归因。
|
|
|
|
用 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_records_relation_without_invite_coins(client) -> None:
|
|
"""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
|
|
|
|
# 邀请金币已下线,双方余额不因邀请变化
|
|
assert _coin_balance(client, b) == 0
|
|
assert _coin_balance(client, a) == 0
|
|
|
|
# A 的战绩:已邀 1 人,邀请金币保持 0
|
|
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_duplicate_relation(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"
|
|
# 邀请金币已下线,双方余额不因邀请变化
|
|
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、金币为 0。"""
|
|
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"] == 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()
|
|
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
|