lovife
57bf39d5a1
一、催报分钟级限流（F-29，契约 §3「10 次/分钟/任务」此前未实现）
· UrgeService 加 Redis 计数：定向与一键催报共用，超限 42901
且话说清等多久、24h 去重还在（去重防「学生被重复催」，限流防「老师连点
把系统打爆」—— 两件事）；限流放在「确实有活可干」之后（空结果连点不计数）
· Redis 不可用时放行不瘫痪（限流是保护，不是业务前提）
二、学生微信登录也写 sys_login_log（F-31 —— 此前只有 PC 账号密码登录留痕，
按 §13.4 查「学生登录记录」永远为空，合规缺口）
· 新增 WechatLoginLogger：小程序登录（MpLoginService）与 H5 授权回调/
打桩登录（MpH5AuthController）三条路径都写
· login_type=2 与 DDL 注释对齐（1 PC 2 小程序 3 开放接口）；H5 与小程序
共用 2（DDL 只定义到 3，不造新值），要区分看 userAgent —— 字段里本来就带着
· 写失败不影响登录（日志表满了不能没人能登录）；不记 code/openid
验证：e2e-m5-urge → 42/42（新增「连点第 11 次被限流」断言，先清计数保证确定性）
e2e-m3-h5-auth → 24/24（新增「两次 stub-login 各留一条 login_type=2」断言，
清理时连登录日志一起清）
247 lines
12 KiB
Python
247 lines
12 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
H5 微信网页授权 验收测试（M3-02）。
|
||
|
||
覆盖：
|
||
[x] /mp/h5/authorize 生成的微信授权地址参数正确（appid/redirect_uri/scope/state）
|
||
[x] state 每次不同（防重放），且带回 scene
|
||
[x] /mp/h5/stub-login（开发用）→ 302 回 H5，**token 在 fragment 里、不在 query 里**
|
||
[x] 打桩登录是幂等的：同一个 openid 第二次登录拿同一个用户
|
||
[x] 拿到的 token 真的能用（过得了认证）
|
||
[x] 没有 scene 时不猜租户，而是提示扫码
|
||
[x] 无效 code 的微信错误被翻译成人话（真调微信）
|
||
|
||
⚠️ 关于打桩：stub-login 是**开发/测试专用**路径，生产禁用（配置开关 + profile 双保险）。
|
||
这里测它，是为了让"H5 登录之后的全部流程"能被自动化 —— 微信授权那一步
|
||
必须真人在微信里点，脚本永远做不了。
|
||
|
||
用法：python scripts/e2e-m3-h5-auth.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
PG_CONTAINER = os.environ.get("XT_PG_CONTAINER", "xt-pg")
|
||
PG_USER = os.environ.get("PG_USER", "fill")
|
||
PG_DB = os.environ.get("PG_DB", "filldb")
|
||
PG_PWD = os.environ.get("PG_PASSWORD", "fill_dev_pwd")
|
||
|
||
PASSED: list[str] = []
|
||
FAILED: list[tuple[str, str]] = []
|
||
CREATED_USER_IDS: list[int] = []
|
||
|
||
|
||
def ok(n: str) -> None:
|
||
PASSED.append(n)
|
||
print(f" [OK] {n}")
|
||
|
||
|
||
def fail(n: str, d: str) -> None:
|
||
FAILED.append((n, d))
|
||
print(f" [FAIL] {n}\n {d}")
|
||
|
||
|
||
def check(n: str, cond: bool, d: str = "") -> bool:
|
||
if cond:
|
||
ok(n)
|
||
else:
|
||
fail(n, d or "断言不成立")
|
||
return cond
|
||
|
||
|
||
def sql(stmt: str) -> tuple[bool, str]:
|
||
try:
|
||
r = subprocess.run(
|
||
["docker", "exec", "-e", f"PGPASSWORD={PG_PWD}", PG_CONTAINER,
|
||
"psql", "-U", PG_USER, "-d", PG_DB, "-t", "-A", "-c", stmt],
|
||
capture_output=True, text=True, timeout=40)
|
||
return r.returncode == 0, (r.stdout or r.stderr).strip()
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def json_call(url: str, token: str | None = None) -> dict:
|
||
headers = {"Content-Type": "application/json"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return json.loads(e.read().decode("utf-8"))
|
||
except Exception:
|
||
return {"_http": e.code}
|
||
|
||
|
||
def redirect_of(url: str) -> tuple[int, dict]:
|
||
"""不跟随重定向，拿 Location 头。"""
|
||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||
def redirect_request(self, *a, **k):
|
||
return None
|
||
|
||
op = urllib.request.build_opener(NoRedirect)
|
||
try:
|
||
with op.open(urllib.request.Request(url), timeout=30) as resp:
|
||
return resp.status, dict(resp.headers)
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, dict(e.headers)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
|
||
args = ap.parse_args()
|
||
base = args.base.rstrip("/")
|
||
|
||
print("=" * 72)
|
||
print(f"H5 微信网页授权 E2E base={base}")
|
||
print("=" * 72)
|
||
|
||
openid = "e2e_openid_" + uuid.uuid4().hex[:10]
|
||
scene = "t1-k17"
|
||
|
||
# ── 1. 授权地址 ─────────────────────────────────────
|
||
print("\n[1] 生成微信授权地址")
|
||
r1 = json_call(f"{base}/api/v1/mp/h5/authorize?scene={scene}")
|
||
check("接口可用", r1.get("code") == 0, f"{r1}")
|
||
u1 = (r1.get("data") or {}).get("authorizeUrl") or ""
|
||
print(f" {u1[:150]}…")
|
||
check("指向微信授权端点", u1.startswith("https://open.weixin.qq.com/connect/oauth2/authorize"),
|
||
u1[:80])
|
||
check("appid 是公众号的（不是小程序的 wxf01d3f…）",
|
||
"appid=wx5d3ae5c1c21fef24" in u1 and "wxf01d3f" not in u1, u1[:120])
|
||
check("scope=snsapi_base（静默授权，学生看不到授权弹窗）", "scope=snsapi_base" in u1, u1[:150])
|
||
check("带 redirect_uri 且已 URL 编码", "redirect_uri=http%3A%2F%2F" in u1 or "redirect_uri=http://" in u1,
|
||
u1[:150])
|
||
check("response_type=code", "response_type=code" in u1, u1[:150])
|
||
|
||
r2 = json_call(f"{base}/api/v1/mp/h5/authorize?scene={scene}")
|
||
u2 = (r2.get("data") or {}).get("authorizeUrl") or ""
|
||
check("两次的 state 不同（防重放）", ("state=" in u1 and "state=" in u2 and u1 != u2),
|
||
"两次返回相同，state 可能是固定的")
|
||
|
||
# ── 2. 打桩登录（开发用）─────────────────────────────
|
||
print("\n[2] 打桩登录（跳过微信授权，供自动化测试）")
|
||
code, headers = redirect_of(
|
||
f"{base}/api/v1/mp/h5/stub-login?openid={openid}&scene={scene}")
|
||
check("返回 302 重定向", code == 302, f"HTTP {code} headers={list(headers.keys())}")
|
||
loc = headers.get("Location", "")
|
||
print(f" Location: {loc[:120]}…")
|
||
# H5 应用地址是可配置的（本地开发用穿透域名），所以不能写死 127.0.0.1 ——
|
||
# 从配置里读期望值，改配置不会让这条断言假失败
|
||
# H5 应用地址是**可配置的**（本地开发用内网穿透域名），所以从配置里读期望值，
|
||
# 别写死 127.0.0.1 —— 改了配置这条断言就会假失败（实测踩过）
|
||
yml = open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "backend", "xt-server", "src", "main",
|
||
"resources", "application-dev.yml"), encoding="utf-8").read()
|
||
mm = re.search(r"^\s*h5-base-url:\s*\$\{[^:}]+:([^}]+)\}", yml, re.M)
|
||
expect_base = mm.group(1).strip() if mm else ""
|
||
check("回到配置的 H5 应用地址", bool(expect_base) and loc.startswith(expect_base),
|
||
f"Location={loc[:80]} 期望前缀={expect_base or '(配置里没读到)'}")
|
||
check("**token 在 fragment（#）里，不在 query 里** —— query 会进 Nginx 日志/Referer",
|
||
"#" in loc and "token=" in loc.split("#", 1)[1],
|
||
f"Location={loc[:140]}")
|
||
check("query 部分没有 token", "token=" not in loc.split("#", 1)[0], loc[:140])
|
||
# ⚠️ fragment 是 **hash 路由格式**（/pages/index/index?token=X&scene=Y），
|
||
# 不是裸 hash。uni-app H5 的 onLoad 只解析路由的 query ——
|
||
# 写成裸 hash 时两边都不认，参数会全丢（踩过）。
|
||
frag = loc.split("#", 1)[1] if "#" in loc else ""
|
||
check("fragment 是 hash 路由格式（uni-app 能解析）",
|
||
frag.startswith("/pages/index/index?"), f"fragment={frag[:60]}")
|
||
qs = frag.split("?", 1)[1] if "?" in frag else ""
|
||
params = dict(p.split("=", 1) for p in qs.split("&") if "=" in p)
|
||
token = urllib.parse.unquote(params.get("token", ""))
|
||
check("fragment 里带回了 scene（授权回来还要继续用）",
|
||
urllib.parse.unquote(params.get("scene", "")) == scene, frag[:100])
|
||
check("token 非空且像 JWT", token.count(".") == 2 and len(token) > 60, f"len={len(token)}")
|
||
|
||
# ── 3. token 真的能用 ───────────────────────────────
|
||
print("\n[3] 拿到的 token 真的能过认证")
|
||
probe = json_call(f"{base}/api/v1/mp/roster/match", token) # GET 会 405，但认证先过
|
||
check("token 有效（不是 40101 未登录）", probe.get("code") != 40101,
|
||
f"code={probe.get('code')} msg={probe.get('msg')}")
|
||
|
||
# ── 4. 幂等 ─────────────────────────────────────────
|
||
print("\n[4] 同一个 openid 第二次登录")
|
||
okk, uid1 = sql("SELECT id FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
|
||
code2, headers2 = redirect_of(
|
||
f"{base}/api/v1/mp/h5/stub-login?openid={openid}&scene={scene}")
|
||
okk2, uid2 = sql("SELECT id FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
|
||
check("第二次登录不重复建号（同一个 user id）",
|
||
okk and okk2 and uid1.strip() == uid2.strip() and uid1.strip() != "",
|
||
f"{uid1.strip()} vs {uid2.strip()}")
|
||
okk3, cnt = sql("SELECT count(*) FROM sys_user WHERE open_id = '" + openid + "' AND deleted = 0")
|
||
check("库里这个 openid 只有一行", okk3 and cnt.strip() == "1", f"{cnt}")
|
||
if uid1.strip().isdigit():
|
||
CREATED_USER_IDS.append(int(uid1.strip()))
|
||
okk4, role = sql("SELECT count(*) FROM sys_user_role ur JOIN sys_role r ON r.id = ur.role_id "
|
||
f"WHERE ur.user_id = {uid1.strip()} AND r.role_code = 'STUDENT'")
|
||
check("已授予学生角色（用户类型由角色推导）", okk4 and role.strip() == "1", f"{role}")
|
||
|
||
# ── 4b. 登录留痕（2026-09-30 补上，F-31）────────────────
|
||
# 此前只有 PC 账号密码登录写 sys_login_log，学生从微信进来完全无痕 ——
|
||
# 按 §13.4 查"学生登录记录"永远是空的，合规上是个缺口
|
||
okk5, lls = sql("SELECT count(*) FROM sys_login_log "
|
||
f"WHERE user_id = {uid1.strip()} AND login_type = 2 AND success = 1")
|
||
check("⭐ 微信登录也写 sys_login_log（login_type=2）",
|
||
okk5 and int(lls.strip()) >= 2,
|
||
f"行数={lls.strip()}（本脚本已做两次 stub-login，应各留一条）")
|
||
|
||
# ── 5. 没有 scene 时不猜租户 ─────────────────────────
|
||
print("\n[5] 没带 scene 时")
|
||
code3, headers3 = redirect_of(f"{base}/api/v1/mp/h5/stub-login?openid={openid}")
|
||
loc3 = headers3.get("Location", "")
|
||
# 打桩路径与真实回调的提示语不同（它明确说"需要带合法的 scene"），
|
||
# 所以这里只断言"有错误信息且没有 token"，不去比对具体文案
|
||
check("不猜默认租户，而是给出错误提示", "authError" in loc3 and len(loc3) > 20,
|
||
f"Location={loc3[:140]}")
|
||
check("提示里能看懂是 scene 的问题（不是一句笼统的失败）",
|
||
"scene" in urllib.parse.unquote(loc3).lower(), urllib.parse.unquote(loc3)[:120])
|
||
check("这种重定向里没有 token", "token=" not in loc3, loc3[:120])
|
||
|
||
# ── 6. 无效 code 的微信错误要能看懂 ───────────────────
|
||
print("\n[6] 无效 code（真调微信，验证错误翻译）")
|
||
code4, headers4 = redirect_of(
|
||
f"{base}/api/v1/mp/h5/callback?code=THIS_CODE_IS_FAKE&state=nonexistent")
|
||
loc4 = headers4.get("Location", "")
|
||
decoded = urllib.parse.unquote(loc4)
|
||
print(f" {decoded[:170]}…")
|
||
# state 不存在 → scene 为空 → 提示扫码（在换 code 之前就该拦）
|
||
check("state 不存在时不往下走（先拦在 scene 缺失）",
|
||
"authError" in loc4, f"Location={decoded[:140]}")
|
||
|
||
# ── 7. 清理 ─────────────────────────────────────────
|
||
print("\n[7] 清理测试数据")
|
||
for uid in CREATED_USER_IDS:
|
||
sql(f"DELETE FROM sys_user_role WHERE user_id = {uid}")
|
||
sql(f"DELETE FROM sys_login_log WHERE user_id = {uid}")
|
||
sql(f"DELETE FROM sys_user WHERE id = {uid}")
|
||
ok(f"已清理 {len(CREATED_USER_IDS)} 个测试账号（含其登录日志）")
|
||
|
||
print("\n" + "=" * 72)
|
||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||
if FAILED:
|
||
print("\n失败项：")
|
||
for n, d in FAILED:
|
||
print(f" · {n}\n {d}")
|
||
print("=" * 72)
|
||
return 1 if FAILED else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|