#!/usr/bin/env python # -*- coding: utf-8 -*- """ M1-06 填报二维码 验收测试。 覆盖《03-接口设计》§7.12： [x] GET /tasks/{id}/qrcode?type=MINI → scene + qrcodeUrl（真实调微信生成） [x] GET /qrcodes/{scene} 匿名取图，返回真实图片二进制 [x] 同一个 scene 的码会被缓存，不重复消耗微信日额度 [x] 匿名图片接口**只读缓存**，不现场调微信（防刷额度） [x] orgId 必须在本任务的覆盖范围内 [x] type=URL（H5 短链）明确不支持，且不返回假的「短链」 [x] 权限 ⚠️ 这个 E2E 会**真实调用微信接口**（生成小程序码）。 它是本仓库唯一一条打外网的测试 —— 需要本机能出网、且凭据有效。 如果失败于 40164（IP 不在白名单），那是环境问题不是代码问题，脚本会明确指出。 为什么值得这样测： 小程序码是学生端的唯一入口。它失败的表现不是报错，而是"码扫不开 / 扫开是空白页"， 而那时码已经发到几百人的班级群里了 —— 必须在上线前真的生成一次、真的取回图片。 用法 python scripts/e2e-m1-qrcode.py 依赖：cryptography；且需要能 `docker exec xt-pg` / `xt-redis` 退出码：0 = 全部通过 """ from __future__ import annotations import argparse import base64 import json import os import subprocess import sys import urllib.error import urllib.request import uuid if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem") 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") REDIS_CONTAINER = os.environ.get("XT_REDIS_CONTAINER", "xt-redis") REDIS_PWD = os.environ.get("REDIS_PASSWORD", "redis_dev_pwd") PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] CTX: dict[str, str | None] = {"base": None, "token": None} CREATED_TASK_IDS: list[int] = [] CREATED_USER_IDS: list[int] = [] CREATED_SCENES: list[str] = [] def ok(name: str) -> None: PASSED.append(name) print(f" [OK] {name}") def fail(name: str, detail: str) -> None: FAILED.append((name, detail)) print(f" [FAIL] {name}\n {detail}") def check(name: str, cond: bool, detail: str = "") -> bool: if cond: ok(name) else: fail(name, detail 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 redis(args: list[str]) -> tuple[bool, str]: try: r = subprocess.run( ["docker", "exec", "-e", f"REDISCLI_AUTH={REDIS_PWD}", REDIS_CONTAINER, "redis-cli"] + args, capture_output=True, text=True, timeout=30) return r.returncode == 0, (r.stdout or r.stderr).strip() except Exception as e: return False, str(e) def cleanup_all() -> None: if CREATED_TASK_IDS: ids = ",".join(str(i) for i in CREATED_TASK_IDS) for stmt in [ f"DELETE FROM collect_task_org WHERE task_id IN ({ids})", f"DELETE FROM collect_material_def WHERE task_id IN ({ids})", f"DELETE FROM collect_form_schema WHERE task_id IN ({ids})", f"DELETE FROM collect_roster WHERE task_id IN ({ids})", f"DELETE FROM collect_task WHERE id IN ({ids})", ]: sql(stmt) for scene in CREATED_SCENES: redis(["DEL", f"xt:wxmp:code:develop:{scene}"]) redis(["DEL", f"xt:wxmp:code:trial:{scene}"]) redis(["DEL", f"xt:wxmp:code:release:{scene}"]) base, token = CTX.get("base"), CTX.get("token") if base and token: for uid in CREATED_USER_IDS: try: call("DELETE", f"{base}/api/v1/users/{uid}", token) except Exception: pass def rsa_encrypt(plain: str) -> str: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding with open(PUB_KEY, "rb") as f: pub = serialization.load_pem_public_key(f.read()) return base64.b64encode(pub.encrypt(plain.encode(), padding.PKCS1v15())).decode() def call(method: str, url: str, token: str | None = None, body: dict | None = None, timeout: int = 40) -> tuple[int, dict]: headers = {"Content-Type": "application/json; charset=utf-8"} if token: headers["Authorization"] = f"Bearer {token}" data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: raw = e.read().decode("utf-8") try: return e.code, json.loads(raw) except json.JSONDecodeError: return e.code, {"raw": raw} def fetch_full(url: str, token: str | None = None, timeout: int = 40): """ 取二进制并连响应头一起返回。 ⚠️ 成功时第 4 个返回值是**空串**而不是 None —— 这样调用方写 `f"HTTP {st} {err[:120]}"` 在成功路径上也不会炸。 之前返回 None 时，恰恰是"本该通过的断言"把脚本崩掉了（成功时 err 为 None）， 属于自己给自己挖坑。 """ headers = {} if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(url, headers=headers, method="GET") try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, dict(resp.headers), resp.read(), "" except urllib.error.HTTPError as e: return e.code, dict(e.headers), b"", e.read().decode("utf-8", "replace")[:300] def fetch_image(url: str, token: str | None = None, timeout: int = 40): """取二进制（图片），不走 JSON。""" st, headers, body, err = fetch_full(url, token, timeout) return st, headers.get("Content-Type", ""), body, err def login(base: str, username: str, password: str, tenant: str | None = None): payload = {"username": username, "password": rsa_encrypt(password)} if tenant: payload["tenantCode"] = tenant return call("POST", f"{base}/api/v1/auth/login", body=payload) def is_image(b: bytes) -> str: if len(b) > 12 and b[0] == 0xFF and b[1] == 0xD8: return "JPEG" if b[:4] == b"\x89PNG": return "PNG" if b[:3] == b"GIF": return "GIF" return "" def main() -> int: ap = argparse.ArgumentParser(description="M1-06 填报二维码 E2E") ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080")) ap.add_argument("--tenant", default=os.environ.get("XT_TENANT", "demo")) ap.add_argument("--password", default="Admin@123") args = ap.parse_args() base = args.base.rstrip("/") print("=" * 72) print(f"M1-06 填报二维码 E2E　base={base}　租户={args.tenant}") print(" ⚠️ 这条会真实调用微信接口生成小程序码") print("=" * 72) suffix = uuid.uuid4().hex[:6] # ── 0. 准备 ──────────────────────────────────────── print("\n[0] 准备") status, body = login(base, "admin", args.password, args.tenant) if body.get("code") != 0: fail("管理员登录", f"HTTP {status} code={body.get('code')} msg={body.get('msg')}") return 1 token = body["data"]["token"] CTX["base"], CTX["token"] = base, token ok("管理员登录") status, body = call("POST", f"{base}/api/v1/tasks", token, { "taskName": f"E2E二维码 {suffix}", "collectMode": 1, "batchNo": "2004", "majorName": "数控加工", "certType": 1, "startTime": "2026-09-28", "endTime": "2026-12-31", }) if body.get("code") != 0: fail("建任务", f"code={body.get('code')} msg={body.get('msg')}") return 1 task_id = body["data"]["id"] task_code = body["data"]["taskCode"] CREATED_TASK_IDS.append(task_id) ok(f"建任务（id={task_id} code={task_code}）") # 取一个班级，用于 orgId 用例 status, body = call("GET", f"{base}/api/v1/orgs/tree", token) classes: list[dict] = [] def walk(nodes): for n in nodes or []: if n.get("orgType") == 3: classes.append(n) walk(n.get("children")) walk(body.get("data")) if not classes: fail("取班级", "demo 租户里没有班级，无法测 orgId 用例") return 1 ok(f"取到班级：{classes[0]['orgName']}（id={classes[0]['id']}）") # ── 1. 生成小程序码 ───────────────────────────────── print("\n[1] 生成小程序码（真实调微信）") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/qrcode?type=MINI", token) if not check("生成成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): msg = body.get("msg") or "" if "40164" in msg: print(" ↑ 这是环境问题：调用方 IP 不在小程序的 IP 白名单里。") print(" 请在微信小程序后台「开发管理 → 开发设置 → IP 白名单」加上本机出口 IP。") return 1 d = body.get("data") or {} scene = d.get("scene") or "" CREATED_SCENES.append(scene) print(f" scene={scene} qrcodeUrl={d.get('qrcodeUrl')}") check("返回 scene", scene.startswith(f"t1-k{task_id}"), f"scene={scene}") check("scene 不超过微信 32 字符上限", len(scene) <= 32, f"长度 {len(scene)}") check("qrcodeUrl 指向本服务的图片接口", (d.get("qrcodeUrl") or "").endswith("/api/v1/qrcodes/" + scene), f"{d.get('qrcodeUrl')}") check("shortUrl 为 null（没有短链服务，不拿长链接冒充）", d.get("shortUrl") is None, f"{d.get('shortUrl')}") check("expireTime 为 null（二维码永久有效，截止由小程序端提示）", d.get("expireTime") is None, f"{d.get('expireTime')}") # ── 2. 匿名取图 ───────────────────────────────────── print("\n[2] 匿名取二维码图片") st, ct, data, err = fetch_image(f"{base}/api/v1/qrcodes/{scene}") if not check("匿名可访问（ 带不了 token，必须匿名）", st == 200, f"HTTP {st} {err}"): return 1 fmt = is_image(data) check(f"返回的是真实图片（{len(data)} 字节，{fmt}）", fmt != "", f"前 16 字节：{data[:16].hex()}") check("Content-Type 是图片", ct.startswith("image/"), f"Content-Type={ct}") check("图不小于 1KB（防止拿到一个空图或错误页）", len(data) > 1024, f"{len(data)} 字节") st2, hdrs, _, _ = fetch_full(f"{base}/api/v1/qrcodes/{scene}") check("带 Cache-Control（同一个 scene 的图不会变，让浏览器缓存，免得列表页每次拉 90KB）", "max-age" in (hdrs.get("Cache-Control") or ""), f"Cache-Control={hdrs.get('Cache-Control')}") # ── 3. 缓存 ───────────────────────────────────────── print("\n[3] 缓存：不重复消耗微信额度") okk, out = redis(["--scan", "--pattern", f"xt:wxmp:code:*:{scene}"]) check("小程序码已进 Redis 缓存", okk and scene in out, f"scan → {out}") okk, out = redis(["--scan", "--pattern", "xt:wxmp:token:*"]) check("access_token 也缓存了（日调用限额，不能每次都取）", okk and out.strip() != "", f"scan → {out}") okk, out = redis(["TTL", f"xt:wxmp:token:{out.strip().splitlines()[0].split(':')[-1]}"]) \ if out.strip() else (False, "") if okk and out.strip().isdigit(): check("token 的 TTL 是 7000 秒（比官方 7200 少 200，避开边界）", 6900 <= int(out.strip()) <= 7000, f"TTL={out.strip()}") # ── 4. 匿名接口只读缓存 ───────────────────────────── print("\n[4] 匿名图片接口只读缓存（防刷微信额度）") fake_scene = "t99999-k99999" st, headers, data, err = fetch_full(f"{base}/api/v1/qrcodes/{fake_scene}") # ⚠️ 项目约定：业务错误返回 **HTTP 200**，真正的错误码在响应体的 code 里 # （见 ErrorCode.httpStatus()：只有 401/403/429 与 5xxxx 才占用真 HTTP 状态码）。 # 所以这里必须解析 body —— 只看 HTTP 状态码会把"业务失败"当成"成功"。 parsed = None try: parsed = json.loads(data.decode("utf-8")) except Exception: pass check("没生成过的 scene 返回失败而不是现生成", parsed is not None and parsed.get("code") == 40001 and "还没有生成过" in (parsed.get("msg") or ""), f"HTTP {st} body={data[:180]!r}") okk, out = redis(["--scan", "--pattern", f"*{fake_scene}*"]) check("确实没有为它生成任何东西（额度没被消耗）", okk and out.strip() == "", f"scan → {out}") okk, out = redis(["EXISTS", f"xt:wxmp:code:develop:{fake_scene}"]) check("Redis 里也没有为它留下 key", okk and out.strip() == "0", f"EXISTS → {out}") # ── 5. orgId ──────────────────────────────────────── print("\n[5] orgId：必须在本任务的覆盖范围里") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/qrcode?type=MINI&orgId={classes[0]['id']}", token) check("班级不在覆盖范围时被拒", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") check("拒绝理由指路到「配置班级」", "配置班级" in (body.get("msg") or ""), f"msg={body.get('msg')}") status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, { "orgs": [{"orgId": classes[0]["id"]}], }) if check("把该班加进覆盖范围", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/qrcode?type=MINI&orgId={classes[0]['id']}", token) check("加进覆盖范围后可以生成带班级的码", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") d2 = body.get("data") or {} org_scene = d2.get("scene") or "" CREATED_SCENES.append(org_scene) check("scene 带上班级后缀 -o{orgId}", org_scene.endswith(f"-o{classes[0]['id']}"), f"scene={org_scene}") st, ct, data, err = fetch_image(f"{base}/api/v1/qrcodes/{org_scene}") check("带班级的码也能取到图片", st == 200 and is_image(data) != "", f"HTTP {st} {err[:120]}") # ── 6. type=URL ───────────────────────────────────── print("\n[6] H5 短链：明确不支持") status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/qrcode?type=URL", token) check("type=URL 被明确拒绝", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") check("拒绝理由说清原因（短链服务 + H5 方案未定），不是笼统的「不支持」", "短链" in (body.get("msg") or "") and "H5" in (body.get("msg") or ""), f"msg={body.get('msg')}") # ── 7. 权限 ───────────────────────────────────────── print("\n[7] 权限") status, body = call("GET", f"{base}/api/v1/tasks/99999999/qrcode", token) check("任务不存在时给出明确提示", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}") ht_user = f"htq{suffix}" status, body = call("POST", f"{base}/api/v1/users", token, { "username": ht_user, "realName": f"E2E二维码班主任{suffix}", "phone": "13900139501", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) if check("造班主任账号", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): uid, pwd = body["data"]["id"], body["data"]["initialPassword"] CREATED_USER_IDS.append(uid) status, body = login(base, ht_user, pwd, args.tenant) ht_token = body["data"]["token"] # 班主任不该看到"别人的任务"的码；本任务没把他设为初审人，所以列表里没有它， # 但 qrcode 接口用的是 task:view + 租户内可见 —— 这里验证的是他能生成（他有 task:view） status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/qrcode?type=MINI", ht_token) check("班主任可生成自己班上任务的码（task:view）", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") # ── 8. 清理 ──────────────────────────────────────── print("\n[8] 清理测试数据") cleanup_all() ok(f"已清理 {len(CREATED_TASK_IDS)} 个任务、{len(CREATED_USER_IDS)} 个账号、{len(CREATED_SCENES)} 个二维码缓存") print("\n" + "=" * 72) print(f"通过 {len(PASSED)}　失败 {len(FAILED)}　共 {len(PASSED) + len(FAILED)}") if FAILED: print("\n失败项：") for name, detail in FAILED: print(f" · {name}\n {detail}") print("=" * 72) return 1 if FAILED else 0 if __name__ == "__main__": try: code = main() except BaseException: print("\n ⚠ 脚本异常退出，正在清理已造出的测试数据…") cleanup_all() raise raise SystemExit(code)