xiaotutongxue/scripts/e2e-m3-h5-auth.py

#!/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}")

    # ── 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_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())