"""邀请 v3 比价发奖测试:好友完成成功比价 → 邀请人得邀请奖励金(独立账户),幂等只发一次, 账户隔离不串金币/现金。被邀请人绑定不得任何奖励由 test_invite.py 覆盖。 """ from __future__ import annotations from sqlalchemy import select from app.core.rewards import INVITE_COMPARE_REWARD_CENTS from app.db.session import SessionLocal from app.models.wallet import CoinAccount, InviteCashTransaction 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 _my_code(client, token: str) -> str: return client.get("/api/v1/invite/me", headers=_auth(token)).json()["invite_code"] def _bind(client, token: str, code: str): return client.post("/api/v1/invite/bind", json={"invite_code": code}, headers=_auth(token)) def _report_compare(client, token: str, trace_id: str, status: str = "success"): return client.post( "/api/v1/compare/record", json={"trace_id": trace_id, "status": status}, headers=_auth(token), ) def _invite_cash(phone: str) -> int: """被试用户的邀请奖励金余额(分)。专用查询端点见小步3,这里直接查 DB。""" with SessionLocal() as db: u = get_user_by_phone(db, phone) acc = db.get(CoinAccount, u.id) if u else None return acc.invite_cash_balance_cents if acc else 0 def test_compare_reward_granted_to_inviter(client) -> None: """B 被 A 邀请后完成一次成功比价 → A 得邀请奖励金 2 元(独立账户),B 不得该奖。""" a = _login(client, "13800003001") b = _login(client, "13800003002") _bind(client, b, _my_code(client, a)) assert _invite_cash("13800003001") == 0 # 发奖前 r = _report_compare(client, b, "trace-reward-1") assert r.status_code == 200, r.text assert _invite_cash("13800003001") == INVITE_COMPARE_REWARD_CENTS # 邀请人到账 assert _invite_cash("13800003002") == 0 # 被邀请人不得此奖 def test_compare_reward_idempotent(client) -> None: """好友比价多次 → 只发一次(compare_reward_granted 幂等)。""" a = _login(client, "13800003003") b = _login(client, "13800003004") _bind(client, b, _my_code(client, a)) _report_compare(client, b, "trace-idem-1") _report_compare(client, b, "trace-idem-2") # 第二次比价(不同 trace) _report_compare(client, b, "trace-idem-1") # 重复上报同 trace assert _invite_cash("13800003003") == INVITE_COMPARE_REWARD_CENTS # 仍只发一次 def test_compare_no_relation_no_reward(client) -> None: """没有邀请关系的人比价 → 不发奖(没人是他的邀请人)。""" x = _login(client, "13800003005") r = _report_compare(client, x, "trace-norel-1") assert r.status_code == 200, r.text assert _invite_cash("13800003005") == 0 def test_compare_failed_no_reward(client) -> None: """失败的比价(status=failed)不触发发奖。""" a = _login(client, "13800003006") b = _login(client, "13800003007") _bind(client, b, _my_code(client, a)) _report_compare(client, b, "trace-fail-1", status="failed") assert _invite_cash("13800003006") == 0 def test_compare_reward_isolated_from_coin_cash(client) -> None: """账户隔离:邀请奖励金进 invite_cash,不串 cash_balance_cents;流水落 invite_cash_transaction。""" a = _login(client, "13800003008") b = _login(client, "13800003009") _bind(client, b, _my_code(client, a)) _report_compare(client, b, "trace-iso-1") with SessionLocal() as db: ua = get_user_by_phone(db, "13800003008") acc = db.get(CoinAccount, ua.id) assert acc.invite_cash_balance_cents == INVITE_COMPARE_REWARD_CENTS # 奖励金到账 assert acc.cash_balance_cents == 0 # 没串进金币现金 txns = db.execute( select(InviteCashTransaction).where( InviteCashTransaction.user_id == ua.id, InviteCashTransaction.biz_type == "invite_reward", ) ).scalars().all() assert len(txns) == 1 assert txns[0].amount_cents == INVITE_COMPARE_REWARD_CENTS