左辰勇
71135c4e3a
start_play 无锁 check-then-insert，并发 /start 会算出同一个 seq、各拿一个 play_token 绕过次数上限刷金币；加 (user_id, seq) 唯一键，撞键即降级为不放视频。 grant_play 改成 status='playing'→'granted' 条件更新（compare-and-set）， 播完与关闭抢跑/重试只会命中一条，不再二次铸币。含迁移与测试。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
264 lines
10 KiB
Python
264 lines
10 KiB
Python
"""新手引导视频:次数上限 + 发币幂等。
|
|
|
|
这条链路直接铸币,两处并发缺陷曾经都是真漏洞,所以本文件的重点不是走通 happy path,
|
|
而是**并发失败分支**:
|
|
- `/start` 并发算出同一个 seq → (user_id, seq) 唯一键挡下,降级为"这次不放视频";
|
|
- `/reward` 并发抢跑 → status 条件更新只让一个 rowcount=1,输的那个不得入账。
|
|
|
|
真并发在 SQLite 测试库里复现不了(读事务会直接把写方锁死,而不是让它读到旧快照),
|
|
所以用"把对手已经落库的状态先摆好、再让被测方按旧读数往下走"来模拟,断言的是同一件事:
|
|
输的一方必须空手而归,且账不能乱。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models.guide_video import GuideVideoPlay
|
|
from app.models.wallet import CoinAccount, CoinTransaction
|
|
from app.repositories import guide_video as crud_guide
|
|
from app.repositories.user import get_user_by_phone
|
|
|
|
VIDEO_URL = "/media/guide_video/pytest_guide.mp4"
|
|
|
|
|
|
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 _user_id(phone: str) -> int:
|
|
with SessionLocal() as db:
|
|
user = get_user_by_phone(db, phone)
|
|
assert user is not None
|
|
return user.id
|
|
|
|
|
|
def _coin_balance(user_id: int) -> int:
|
|
with SessionLocal() as db:
|
|
acc = db.get(CoinAccount, user_id)
|
|
return acc.coin_balance if acc else 0
|
|
|
|
|
|
def _guide_txns(user_id: int) -> list[CoinTransaction]:
|
|
"""该用户的引导视频金币流水(按 id 升序)。"""
|
|
with SessionLocal() as db:
|
|
return list(db.execute(
|
|
select(CoinTransaction)
|
|
.where(
|
|
CoinTransaction.user_id == user_id,
|
|
CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
|
|
)
|
|
.order_by(CoinTransaction.id)
|
|
).scalars().all())
|
|
|
|
|
|
def _assert_ledger_balanced(user_id: int) -> None:
|
|
"""coin_balance 必须恒等于流水总和 —— 双发/丢更新都会先在这里露馅。"""
|
|
with SessionLocal() as db:
|
|
total = db.execute(
|
|
select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
|
|
CoinTransaction.user_id == user_id
|
|
)
|
|
).scalar_one()
|
|
acc = db.get(CoinAccount, user_id)
|
|
assert acc is not None
|
|
assert acc.coin_balance == total, f"余额 {acc.coin_balance} != 流水总和 {total}"
|
|
|
|
|
|
@pytest.fixture()
|
|
def guide_configured():
|
|
"""给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
|
|
with SessionLocal() as db:
|
|
crud_guide.set_video(db, VIDEO_URL, admin_id=1)
|
|
cfg = crud_guide.get_config(db)
|
|
yield cfg
|
|
with SessionLocal() as db:
|
|
crud_guide.set_video(db, None, admin_id=1)
|
|
|
|
|
|
# ===== 基本闭环 =====
|
|
|
|
|
|
def test_start_miss_when_no_video_configured(client) -> None:
|
|
"""没配片 → should_play=False,客户端照旧放广告(本功能不配视频就等于没上线)。"""
|
|
token = _login(client, "13920000001")
|
|
r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["should_play"] is False
|
|
assert body["play_token"] == ""
|
|
|
|
|
|
def test_start_then_reward_grants_once(client, guide_configured) -> None:
|
|
"""开播 → 上报 → 到账 120;重复上报不再加钱,只回已发金额。"""
|
|
cfg = guide_configured
|
|
phone = "13920000002"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
|
|
body = r.json()
|
|
assert body["should_play"] is True
|
|
assert body["video_url"] == VIDEO_URL
|
|
assert body["seq"] == 1
|
|
assert body["reward_coin"] == cfg["reward_coin"]
|
|
play_token = body["play_token"]
|
|
assert play_token
|
|
|
|
r = client.post(
|
|
"/api/v1/guide-video/reward",
|
|
json={"play_token": play_token, "completed": True},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"granted": True, "coin": cfg["reward_coin"], "status": "granted"}
|
|
assert _coin_balance(uid) == cfg["reward_coin"]
|
|
|
|
# 重复上报(客户端超时重试 / 播完与 ✕ 都报了一次)
|
|
r = client.post(
|
|
"/api/v1/guide-video/reward",
|
|
json={"play_token": play_token, "completed": False},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.json() == {"granted": False, "coin": cfg["reward_coin"], "status": "already_granted"}
|
|
assert _coin_balance(uid) == cfg["reward_coin"], "重复上报不得二次入账"
|
|
assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
|
|
_assert_ledger_balanced(uid)
|
|
|
|
|
|
def test_reward_rejects_unknown_and_others_token(client, guide_configured) -> None:
|
|
"""乱填 token / 拿别人的 token 都发不出币(grant 按 user_id + token 双条件定位)。"""
|
|
victim = _login(client, "13920000003")
|
|
r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(victim))
|
|
stolen = r.json()["play_token"]
|
|
|
|
attacker_phone = "13920000004"
|
|
attacker = _login(client, attacker_phone)
|
|
attacker_uid = _user_id(attacker_phone)
|
|
|
|
r = client.post(
|
|
"/api/v1/guide-video/reward",
|
|
json={"play_token": stolen, "completed": True},
|
|
headers=_auth(attacker),
|
|
)
|
|
assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
|
|
|
|
r = client.post(
|
|
"/api/v1/guide-video/reward",
|
|
json={"play_token": "deadbeef" * 4, "completed": True},
|
|
headers=_auth(attacker),
|
|
)
|
|
assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
|
|
assert _coin_balance(attacker_uid) == 0
|
|
|
|
|
|
def test_start_stops_at_max_plays(client, guide_configured) -> None:
|
|
"""开播即计次:用满 max_plays 后不再下发,客户端回到广告链路。"""
|
|
max_plays = int(guide_configured["max_plays"])
|
|
token = _login(client, "13920000005")
|
|
|
|
for i in range(1, max_plays + 1):
|
|
body = client.post(
|
|
"/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
|
|
).json()
|
|
assert body["should_play"] is True, f"第 {i} 次应该还能放"
|
|
assert body["seq"] == i
|
|
assert body["remaining"] == max_plays - i
|
|
|
|
body = client.post(
|
|
"/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
|
|
).json()
|
|
assert body["should_play"] is False
|
|
assert body["remaining"] == 0
|
|
|
|
|
|
# ===== 并发失败分支(回归) =====
|
|
|
|
|
|
def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypatch) -> None:
|
|
"""并发 /start 抢到同一个 seq 时,输的一方降级成"不放视频",而不是多拿一个 token。
|
|
|
|
模拟:对手已经提交了 seq=1,而本次请求读到的还是旧计数(used=0)—— 这正是无锁
|
|
check-then-insert 的race window。没有 (user_id, seq) 唯一键的话,这里会插成第二行、
|
|
换回第二个 play_token,3 次上限就能被并发无限绕过(改包即可 N 倍刷币)。
|
|
"""
|
|
phone = "13920000006"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
# 对手先落一行 seq=1
|
|
first = client.post(
|
|
"/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
|
|
).json()
|
|
assert first["should_play"] is True and first["seq"] == 1
|
|
|
|
# 本次请求读到的是过期计数 → 仍会算出 seq=1
|
|
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
|
|
with SessionLocal() as db:
|
|
result = crud_guide.start_play(db, uid)
|
|
|
|
assert result["should_play"] is False, "撞 seq 唯一键后必须降级,不能再发一个 token"
|
|
assert result["play_token"] == ""
|
|
|
|
monkeypatch.undo()
|
|
with SessionLocal() as db:
|
|
rows = db.execute(
|
|
select(func.count()).select_from(GuideVideoPlay).where(
|
|
GuideVideoPlay.user_id == uid
|
|
)
|
|
).scalar_one()
|
|
assert rows == 1, "抢输的那次不该留下播放行"
|
|
|
|
|
|
def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> None:
|
|
"""并发 /reward 抢输的一方不得入账 —— 条件更新(status 进 WHERE)的失败分支。
|
|
|
|
模拟的是真并发的**必要条件**:输的一方在对手提交之前就已经读到 status='playing',
|
|
之后带着这个旧认知继续往下走(「播完」与「✕ 关闭」抢跑、或超时重试都会造出这一幕)。
|
|
这里靠 Session 的 identity map 固化那次旧读数 —— 后续同一 Session 再查同一行,拿回的
|
|
还是这份旧快照,等价于 PG READ COMMITTED 下两个事务都读到 playing。
|
|
|
|
旧实现在这里会照发第二笔:它先读再判再写,而 `except IntegrityError` 兜底根本不可能
|
|
触发 —— 发币走 UPDATE 撞不到 uq_guide_video_play_token,biz_type='guide_video' 的流水
|
|
也不在 ux_coin_transaction_task_ref 的谓词(biz_type LIKE 'task%')覆盖内。
|
|
"""
|
|
coin = int(guide_configured["reward_coin"])
|
|
phone = "13920000007"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
play_token = client.post(
|
|
"/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
|
|
).json()["play_token"]
|
|
|
|
db_loser = SessionLocal()
|
|
try:
|
|
# 输的一方先读到 playing —— 并发下两个请求都会读到它
|
|
stale = db_loser.execute(
|
|
select(GuideVideoPlay).where(GuideVideoPlay.play_token == play_token)
|
|
).scalar_one()
|
|
assert stale.status == "playing"
|
|
|
|
# 对手抢先发币并提交
|
|
with SessionLocal() as db_winner:
|
|
won = crud_guide.grant_play(db_winner, uid, play_token=play_token, completed=True)
|
|
assert won == {"granted": True, "coin": coin, "status": "granted"}
|
|
|
|
# 输的一方带着旧认知继续:必须空手而归
|
|
lost = crud_guide.grant_play(db_loser, uid, play_token=play_token, completed=False)
|
|
assert lost == {"granted": False, "coin": coin, "status": "already_granted"}
|
|
finally:
|
|
db_loser.close()
|
|
|
|
assert _coin_balance(uid) == coin, "一次播放只能发一次币"
|
|
assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
|
|
_assert_ledger_balanced(uid)
|