lowmaster-chen
359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""看激励视频发奖测试(穿山甲 S2S 回调)。
|
|
|
|
回调无 JWT,靠验签——测试用配置里的 mock 密钥自签自验。覆盖:发奖到账、幂等、
|
|
验签失败、每日上限、未知用户、客户端进度查询、回调未配置。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import settings
|
|
from app.integrations import pangle
|
|
from app.core.rewards import (
|
|
DAILY_AD_REWARD_LIMIT,
|
|
VIDEO_ROUND_COOLDOWN_SECONDS,
|
|
VIDEO_ROUND_REQUIRED_COUNT,
|
|
calculate_ad_reward_coin,
|
|
)
|
|
from app.db.session import SessionLocal
|
|
from app.models.ad_reward import AdRewardRecord
|
|
from app.models.user import User
|
|
|
|
|
|
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:
|
|
db = SessionLocal()
|
|
try:
|
|
return db.execute(select(User.id).where(User.phone == phone)).scalar_one()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _signed(user_id: int, trans_id: str, **extra: str) -> dict[str, str]:
|
|
"""构造带合法签名的回调参数(模拟穿山甲服务器)。sign 只对 trans_id 签(穿山甲规范)。"""
|
|
params = {"user_id": str(user_id), "trans_id": trans_id, **extra}
|
|
params["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
|
|
return params
|
|
|
|
|
|
def _callback(client, params: dict[str, str]):
|
|
return client.get("/api/v1/ad/pangle-callback", params=params)
|
|
|
|
|
|
def _coin_balance(client, token: str) -> int:
|
|
return client.get("/api/v1/wallet/account", headers=_auth(token)).json()["coin_balance"]
|
|
|
|
|
|
def test_callback_grants_coins(client) -> None:
|
|
"""验签通过且带 eCPM → 按数值公式发金币到账 + 计数 +1。"""
|
|
phone = "13800003001"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
r = _callback(client, _signed(uid, "trans_a1", reward_name="金币", ecpm="200"))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"is_verify": True, "reason": 0}
|
|
|
|
assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
|
|
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 1
|
|
assert st["daily_limit"] == DAILY_AD_REWARD_LIMIT
|
|
assert st["remaining"] == DAILY_AD_REWARD_LIMIT - 1
|
|
assert st["coin_per_ad"] == 0
|
|
|
|
|
|
def test_callback_without_ecpm_records_exception(client) -> None:
|
|
"""S2S 和客户端会话都没有 eCPM → 不发金币,只记录异常状态。"""
|
|
phone = "13800003011"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
assert _callback(client, _signed(uid, "ecpm_missing")).status_code == 200
|
|
assert _coin_balance(client, token) == 0
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 0
|
|
|
|
|
|
def test_callback_idempotent(client) -> None:
|
|
"""同一 trans_id 二次回调 → 只发一次金币(穿山甲重试不重复发)。"""
|
|
phone = "13800003002"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
p = _signed(uid, "trans_dup", ecpm="200")
|
|
assert _callback(client, p).status_code == 200
|
|
assert _callback(client, p).status_code == 200 # 重试
|
|
|
|
assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1) # 没翻倍
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 1
|
|
|
|
|
|
def test_callback_bad_sign(client) -> None:
|
|
"""签名错误 → 403,不发金币。"""
|
|
phone = "13800003003"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
params = {"user_id": str(uid), "trans_id": "trans_bad", "sign": "deadbeef"}
|
|
r = _callback(client, params)
|
|
assert r.status_code == 403, r.text
|
|
assert _coin_balance(client, token) == 0
|
|
|
|
|
|
def test_daily_count_cap(client, monkeypatch) -> None:
|
|
"""达到每日发奖**次数兜底上限**后继续回调 → 受理但不发(capped),余额封顶。
|
|
|
|
次数上限现为时长闸的兜底(默认 200),这里 monkeypatch 调小到 3 加速(不真跑 200 次);
|
|
本用例不上报观看时长 → 时长主闸不触发,纯验次数兜底。
|
|
"""
|
|
# 次数兜底现走 app_config getter(rebase 合 main 的运营可配后);patch getter 调小到 3
|
|
monkeypatch.setattr("app.core.rewards.get_ad_daily_limit", lambda db: 3)
|
|
phone = "13800003004"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
for i in range(3):
|
|
r = _callback(client, _signed(uid, f"trans_cap_{i}", ecpm="200"))
|
|
assert r.status_code == 200, r.text
|
|
|
|
# 第 4 次:capped,仍 is_verify(不让穿山甲重试),但不加币
|
|
r = _callback(client, _signed(uid, "trans_cap_over", ecpm="200"))
|
|
assert r.status_code == 200
|
|
assert r.json() == {"is_verify": True, "reason": 0}
|
|
|
|
assert _coin_balance(client, token) == sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 3
|
|
assert st["remaining"] == 0
|
|
|
|
|
|
def test_daily_watch_time_cap(client, monkeypatch) -> None:
|
|
"""看广告累计观看时长达每日上限(主闸,这里调小到 100s)后,再回调发奖 → capped 不发金币。
|
|
|
|
时长由前端 watch-report 上报累计;到顶后 ① watch-report/reward-status remaining=0(客户端不再展示),
|
|
② 后端发奖也因时长闸记 capped(双闸,前端绕不过)。次数兜底不动它(此时只看了 1 次,远没到次数上限)。
|
|
"""
|
|
# 时长上限两处引用都要 patch:api 用 rewards.X、grant 闸/today_status 用 ad_reward 内绑定的 X
|
|
monkeypatch.setattr("app.core.rewards.DAILY_AD_WATCH_SECONDS_LIMIT", 100)
|
|
monkeypatch.setattr("app.repositories.ad_reward.DAILY_AD_WATCH_SECONDS_LIMIT", 100)
|
|
phone = "13800003201"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
# 上报一次 100s 观看 → 当日累计达上限,remaining=0
|
|
r = client.post("/api/v1/ad/watch-report", json={"seconds": 100}, headers=_auth(token))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["watched_seconds_today"] == 100
|
|
assert r.json()["watch_seconds_remaining"] == 0
|
|
|
|
# 此时回调发奖 → 时长闸命中 → capped(is_verify 仍 true,不让重试),不加金币
|
|
before = _coin_balance(client, token)
|
|
r = _callback(client, _signed(uid, "trans_time_cap", ecpm="200"))
|
|
assert r.status_code == 200
|
|
assert r.json() == {"is_verify": True, "reason": 0}
|
|
assert _coin_balance(client, token) == before # 未加币
|
|
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["watched_seconds_today"] == 100
|
|
assert st["watch_seconds_limit"] == 100
|
|
assert st["watch_seconds_remaining"] == 0
|
|
|
|
|
|
def test_callback_unknown_user(client) -> None:
|
|
"""验签过但 user_id 不存在 → 不发奖(is_verify false + reason),不崩。"""
|
|
params = _signed(999999, "trans_ghost")
|
|
r = _callback(client, params)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"is_verify": False, "reason": 2}
|
|
|
|
|
|
def test_callback_foreign_env_skipped(client) -> None:
|
|
"""回调 srv_env 与本服 APP_ENV 不符(测试包的 S2S 打到生产)→ 受理但不发币、不串号。
|
|
|
|
防的是跨库串号:同一 user_id 在测试库 / 正式库是两个人,回调地址只配生产,测试包(dev)看广告
|
|
的 S2S 也会打到生产。测试环境 APP_ENV=dev:srv_env=prod 视为「别的环境」→ 跳过,余额不变、
|
|
不写记录;srv_env=dev(本环境)照常发。srv_env 走 gromoreExtra 透传(穿山甲带回 mediaExtra 的字段)。
|
|
兼容:不带 srv_env 的旧客户端按本环境处理(其余用例已覆盖)。
|
|
"""
|
|
phone = "13800003501"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
|
|
# 外环境(prod)回调:受理(is_verify=true 防穿山甲重试)但不发币、不留记录
|
|
foreign = _signed(uid, "trans_foreign_env", ecpm="200",
|
|
gromoreExtra=json.dumps({"srv_env": "prod"}))
|
|
r = _callback(client, foreign)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"is_verify": True, "reason": 0}
|
|
assert _coin_balance(client, token) == 0
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 0
|
|
|
|
# 本环境(dev)回调:照常发币
|
|
same = _signed(uid, "trans_same_env", ecpm="200",
|
|
gromoreExtra=json.dumps({"srv_env": "dev"}))
|
|
r = _callback(client, same)
|
|
assert r.status_code == 200, r.text
|
|
assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 1
|
|
|
|
|
|
def test_reward_status_requires_auth(client) -> None:
|
|
"""客户端进度接口需登录。"""
|
|
r = client.get("/api/v1/ad/reward-status")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_reward_status_initial_round_state(client) -> None:
|
|
"""新用户没看广告 → round_count=0, cooldown_until=None。"""
|
|
phone = "13800003101"
|
|
token = _login(client, phone)
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 0
|
|
assert st["round_count"] == 0
|
|
assert st["cooldown_until"] is None
|
|
|
|
|
|
def test_reward_status_mid_round(client) -> None:
|
|
"""当前 round_size=1:看 1 次 → round_count=0, cooldown_until 进入 3 秒短冷却。"""
|
|
phone = "13800003102"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
_callback(client, _signed(uid, "trans_mid_1", ecpm="200"))
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == 1
|
|
assert st["round_count"] == 0
|
|
assert st["cooldown_until"] is not None
|
|
|
|
|
|
def test_reward_status_round_complete_enters_cooldown(client) -> None:
|
|
"""刚看完一条广告 → round_count=0(下一轮起点) + cooldown_until 是未来 ~3 秒。"""
|
|
from datetime import datetime, timezone
|
|
|
|
phone = "13800003103"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
for i in range(VIDEO_ROUND_REQUIRED_COUNT):
|
|
_callback(client, _signed(uid, f"trans_rc_{i}", ecpm="200"))
|
|
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
|
|
assert st["round_count"] == 0
|
|
assert st["cooldown_until"] is not None
|
|
cd = datetime.fromisoformat(st["cooldown_until"].replace("Z", "+00:00"))
|
|
if cd.tzinfo is None:
|
|
cd = cd.replace(tzinfo=timezone.utc)
|
|
now = datetime.now(timezone.utc)
|
|
delta = (cd - now).total_seconds()
|
|
# 配置 3s,允许 ±30s 容差(本机 / CI 慢 IO)
|
|
assert 0 < delta <= VIDEO_ROUND_COOLDOWN_SECONDS + 30
|
|
|
|
|
|
def test_reward_status_cooldown_expired(client) -> None:
|
|
"""看完一条广告后冷却已过(手动改 created_at 到冷却前)→ cooldown_until 回到 None。"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from sqlalchemy import select, update
|
|
|
|
phone = "13800003104"
|
|
token = _login(client, phone)
|
|
uid = _user_id(phone)
|
|
for i in range(VIDEO_ROUND_REQUIRED_COUNT):
|
|
_callback(client, _signed(uid, f"trans_exp_{i}", ecpm="200"))
|
|
|
|
# 把当日所有 granted 记录的 created_at 推到 11 分钟前(覆盖冷却末尾那条)
|
|
db = SessionLocal()
|
|
try:
|
|
eleven_min_ago = datetime.now(timezone.utc) - timedelta(seconds=VIDEO_ROUND_COOLDOWN_SECONDS + 60)
|
|
db.execute(
|
|
update(AdRewardRecord)
|
|
.where(AdRewardRecord.user_id == uid)
|
|
.values(created_at=eleven_min_ago)
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
|
|
assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
|
|
assert st["round_count"] == 0
|
|
assert st["cooldown_until"] is None
|
|
|
|
|
|
def test_feed_reward_grants_by_10_second_units(client) -> None:
|
|
"""信息流广告完成后:每满 10 秒一份奖励,同一 client_event_id 幂等。"""
|
|
phone = "13800003301"
|
|
token = _login(client, phone)
|
|
|
|
payload = {
|
|
"client_event_id": "feed_evt_0001",
|
|
"ecpm": "200",
|
|
"duration_seconds": 30,
|
|
"adn": "pangle",
|
|
"slot_id": "slot_feed",
|
|
"display_coin": 4,
|
|
}
|
|
r = client.post("/api/v1/ad/feed-reward", json=payload, headers=_auth(token))
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
expected = 4
|
|
assert body["granted"] is True
|
|
assert body["status"] == "granted"
|
|
assert body["unit_count"] == 3
|
|
assert body["coin"] == expected
|
|
assert _coin_balance(client, token) == expected
|
|
|
|
# 重试同一个事件不重复发
|
|
r = client.post("/api/v1/ad/feed-reward", json=payload, headers=_auth(token))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["coin"] == expected
|
|
assert _coin_balance(client, token) == expected
|
|
|
|
|
|
def test_feed_reward_daily_display_cap(client, monkeypatch) -> None:
|
|
"""信息流展示次数到每日上限后继续完成 → capped,不发金币。"""
|
|
monkeypatch.setattr("app.core.rewards.get_ad_daily_limit", lambda db: 1)
|
|
phone = "13800003302"
|
|
token = _login(client, phone)
|
|
|
|
first = {
|
|
"client_event_id": "feed_evt_cap_1",
|
|
"ecpm": "201",
|
|
"duration_seconds": 10,
|
|
}
|
|
second = {
|
|
"client_event_id": "feed_evt_cap_2",
|
|
"ecpm": "201",
|
|
"duration_seconds": 10,
|
|
}
|
|
assert client.post("/api/v1/ad/feed-reward", json=first, headers=_auth(token)).status_code == 200
|
|
before = _coin_balance(client, token)
|
|
r = client.post("/api/v1/ad/feed-reward", json=second, headers=_auth(token))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["granted"] is False
|
|
assert r.json()["status"] == "capped"
|
|
assert r.json()["coin"] == 0
|
|
assert _coin_balance(client, token) == before
|
|
|
|
|
|
def test_callback_disabled_returns_503(client, monkeypatch) -> None:
|
|
"""未配置回调(开关关)时 → 503。"""
|
|
monkeypatch.setattr(settings, "PANGLE_CALLBACK_ENABLED", False)
|
|
# 503 发生在验签/发奖之前,不需要真实用户
|
|
r = _callback(client, _signed(1, "trans_disabled"))
|
|
assert r.status_code == 503, r.text
|