#!/usr/bin/env python # -*- coding: utf-8 -*- """ 扫码落地页（GET /api/v1/mp/scene-context）验收测试。 覆盖 FR-105（统一小程序 + 扫码带参 scene 识别租户）： [x] 正常 scene：返回租户名、任务名、状态、能否填报、给学生的话 [x] 非法 scene / 不存在的租户 / 不存在的任务：各自给出**不同的**提示 [x] 租户级 scene（不带任务）：提示去扫任务码 [x] 带班级：班级名只在"该班确实在任务覆盖范围里"时才返回 [x] 跨租户：A 租户的 scene 拼上 B 租户的任务 id → 必须当作"不存在" [x] 状态文案随任务状态变化（收集中/已暂停/已截止） [x] **限流**：scene 可穷举，必须能挡住批量扫描 [x] 这个接口**匿名**就能访问（学生登录前就要用） 为什么这些值得逐条断言： 这是学生扫码后的第一屏，而且它是**匿名**接口、scene 又可穷举（t1-k1、t1-k2…）。 一旦哪个分支写错，后果不是"报错"，而是"把不该给的信息给出去了"， 或者"扫对了码却告诉他这个码不是本系统的" —— 后者会让学生直接放弃。 用法 python scripts/e2e-m1-scene.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.parse 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]] = [] CREATED_TASK_IDS: list[int] = [] CREATED_TENANT_IDS: list[int] = [] 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_roster WHERE task_id IN ({ids})", f"DELETE FROM collect_task WHERE id IN ({ids})", ]: sql(stmt) for tid in CREATED_TENANT_IDS: for stmt in [ f"DELETE FROM sys_user_role WHERE user_id IN (SELECT id FROM sys_user WHERE tenant_id = {tid})", f"DELETE FROM sys_user WHERE tenant_id = {tid}", f"DELETE FROM sys_org WHERE tenant_id = {tid}", f"DELETE FROM sys_role WHERE tenant_id = {tid}", f"DELETE FROM collect_task WHERE tenant_id = {tid}", f"DELETE FROM sys_tenant WHERE id = {tid}", ]: sql(stmt) redis(["DEL", "xt:mp:scene:rate:127.0.0.1"]) redis(["DEL", "xt:mp:scene:rate:localhost"]) 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 = 30) -> 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 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 scene(base: str, value: str, token: str | None = None) -> dict: """调 scene-context，返回 data。""" url = f"{base}/api/v1/mp/scene-context?scene={urllib.parse.quote(value)}" _, body = call("GET", url, token) return body.get("data") or {} def scene_raw(base: str, value: str) -> tuple[int, dict]: url = f"{base}/api/v1/mp/scene-context?scene={urllib.parse.quote(value)}" return call("GET", url) def main() -> int: ap = argparse.ArgumentParser(description="扫码落地页 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"扫码落地页 E2E　base={base}") print(" （本接口是匿名的 —— 学生登录前就要用；测试也刻意不带 token）") print("=" * 72) suffix = uuid.uuid4().hex[:6] redis(["DEL", "xt:mp:scene:rate:127.0.0.1"]) # ── 0. 准备（用管理员造数据，但 scene 查询都不带 token）──── print("\n[0] 准备") status, body = login(base, "admin", args.password, args.tenant) if body.get("code") != 0: fail("管理员登录", f"code={body.get('code')} msg={body.get('msg')}") return 1 token = body["data"]["token"] tenant_id = (body["data"].get("tenant") or {}).get("id") ok(f"管理员登录（租户 id={tenant_id}）") # 本次要用的任务 status, body = call("POST", f"{base}/api/v1/tasks", token, { "taskName": f"E2E落地页 {suffix}", "collectMode": 1, "batchNo": "2005", "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"] CREATED_TASK_IDS.append(task_id) ok(f"建任务（id={task_id}，草稿状态）") 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 租户里没有班级") return 1 # ── 1. 正常 scene（匿名）───────────────────────────── print("\n[1] 正常 scene（匿名，不带 token）") d = scene(base, f"t{tenant_id}-k{task_id}") check("返回租户名（学生要能确认扫对了学校）", bool(d.get("tenantName")), f"tenantName={d.get('tenantName')}") check("返回任务名", d.get("taskName") == f"E2E落地页 {suffix}", f"taskName={d.get('taskName')}") check("草稿状态 → 尚未发布", d.get("taskStatusText") == "尚未发布", f"{d.get('taskStatusText')}") check("草稿不可填报", d.get("canSubmit") is False, f"canSubmit={d.get('canSubmit')}") check("提示语是「尚未开始」（不是笼统的不可填报）", "尚未开始" in (d.get("message") or ""), f"message={d.get('message')}") check("valid 为 true", d.get("valid") is True, f"valid={d.get('valid')}") # ── 2. 各种非法输入要给不同的提示 ───────────────────── print("\n[2] 非法输入：提示要各不相同（否则学生不知道该找谁）") d_garbage = scene(base, "garbage") d_no_tenant = scene(base, "t99999999-k1") d_no_task = scene(base, f"t{tenant_id}-k99999999") check("乱码 → 不是本系统的二维码", "不是本系统生成" in (d_garbage.get("message") or ""), f"{d_garbage.get('message')}") check("不存在的租户 → 单位不存在", "单位不存在" in (d_no_tenant.get("message") or ""), f"{d_no_tenant.get('message')}") check("不存在的任务 → 任务不存在", "任务不存在" in (d_no_task.get("message") or ""), f"{d_no_task.get('message')}") msgs = {d_garbage.get("message"), d_no_tenant.get("message"), d_no_task.get("message")} check("三种失败的提示互不相同（不是同一句话敷衍）", len(msgs) == 3, f"{msgs}") # ── 3. 租户级 scene ────────────────────────────────── print("\n[3] 租户级 scene（M0 开通租户时约定的 t{tenantId}）") d_tenant = scene(base, f"t{tenant_id}") check("租户级码有效但提示去扫任务码", d_tenant.get("valid") is True and "填报二维码" in (d_tenant.get("message") or ""), f"valid={d_tenant.get('valid')} message={d_tenant.get('message')}") check("租户级码没有任务信息", d_tenant.get("taskId") is None, f"{d_tenant.get('taskId')}") # ── 4. 带班级 ──────────────────────────────────────── print("\n[4] 带班级的 scene") d_not_covered = scene(base, f"t{tenant_id}-k{task_id}-o{classes[0]['id']}") check("班级不在覆盖范围时不返回 orgName（不能凭空显示一个班）", d_not_covered.get("orgName") is None, f"orgName={d_not_covered.get('orgName')}") call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", token, {"orgs": [{"orgId": classes[0]["id"]}]}) d_covered = scene(base, f"t{tenant_id}-k{task_id}-o{classes[0]['id']}") check("班级在覆盖范围时返回班级名", d_covered.get("orgName") == classes[0]["orgName"], f"orgName={d_covered.get('orgName')} 期望={classes[0]['orgName']}") check("带班不影响任务信息", d_covered.get("taskId") == task_id, f"{d_covered.get('taskId')}") # ── 5. 跨租户 ──────────────────────────────────────── print("\n[5] 跨租户：A 租户的 scene 拼 B 租户的任务 id") status, plat = login(base, "platform", "Platform@123") other_task_id = None if plat.get("code") == 0: plat_token = plat["data"]["token"] other_code = f"scn{suffix}" status, body = call("POST", f"{base}/api/v1/tenants", plat_token, { "tenantName": f"E2E扫码隔离 {suffix}", "tenantCode": other_code, "tenantType": 1, "contactName": "测试", "contactPhone": "13900139601", "expireTime": "2027-12-31", "adminUsername": "admin", "adminRealName": "隔离管理员", "adminPhone": "13900139602", }) if body.get("code") == 0: other_tenant_id = body["data"]["id"] other_pwd = body["data"]["adminInitialPassword"] CREATED_TENANT_IDS.append(other_tenant_id) login(base, "admin", other_pwd, other_code) _, ob = login(base, "admin", other_pwd, other_code) if ob.get("code") == 0: ot = ob["data"]["token"] _, tb = call("POST", f"{base}/api/v1/tasks", ot, { "taskName": f"别校的任务 {suffix}", "collectMode": 1, "batchNo": "2011", "majorName": "x", "certType": 1, "startTime": "2026-09-28", "endTime": "2026-12-31", }) if tb.get("code") == 0: other_task_id = tb["data"]["id"] sql(f"DELETE FROM collect_task WHERE id = {other_task_id}") # 用 SQL 收尾更省事 if other_task_id: d_cross = scene(base, f"t{tenant_id}-k{other_task_id}") check("用 A 租户的租户号 + B 租户的任务 id → 任务不存在（不跨租户泄露）", "任务不存在" in (d_cross.get("message") or ""), f"message={d_cross.get('message')}") d_cross2 = scene(base, f"t{CREATED_TENANT_IDS[0]}-k{task_id}") check("反过来（B 的租户号 + A 的任务 id）也不泄露", "任务不存在" in (d_cross2.get("message") or ""), f"message={d_cross2.get('message')}") else: fail("跨租户用例", "第二个租户没造出来，这条隔离断言没执行") # ── 6. 状态文案随状态变化 ──────────────────────────── print("\n[6] 状态文案：五种状态给五句不同的话") sql(f"INSERT INTO collect_roster (tenant_id, task_id, real_name, id_card_no, id_card_hash) " f"VALUES ({tenant_id}, {task_id}, 'E2E学生A', 'seed', 'seed')") sql(f"INSERT INTO collect_material_def (tenant_id, task_id, material_code, material_name, required) " f"VALUES ({tenant_id}, {task_id}, 'IDCARD', '身份证', 1)") call("POST", f"{base}/api/v1/tasks/{task_id}/publish", token, {"force": True}) d_collecting = scene(base, f"t{tenant_id}-k{task_id}") check("收集中 → 可填报", d_collecting.get("canSubmit") is True, f"canSubmit={d_collecting.get('canSubmit')} status={d_collecting.get('taskStatusText')}") check("收集中 → 提示「开始填报」", "开始填报" in (d_collecting.get("message") or ""), f"{d_collecting.get('message')}") call("POST", f"{base}/api/v1/tasks/{task_id}/pause", token, {}) d_paused = scene(base, f"t{tenant_id}-k{task_id}") check("已暂停 → 不可填报", d_paused.get("canSubmit") is False, f"{d_paused.get('canSubmit')}") check("已暂停 → 提示等恢复通知（不是「已截止」，学生的下一步动作完全不同）", "暂停" in (d_paused.get("message") or ""), f"{d_paused.get('message')}") call("POST", f"{base}/api/v1/tasks/{task_id}/resume", token) call("POST", f"{base}/api/v1/tasks/{task_id}/close", token, {}) d_closed = scene(base, f"t{tenant_id}-k{task_id}") check("已截止 → 不可填报", d_closed.get("canSubmit") is False, f"{d_closed.get('canSubmit')}") check("已截止 → 提示联系班主任补交（§7.12：码不做失效，靠这句话说明状态）", "截止" in (d_closed.get("message") or "") and "班主任" in (d_closed.get("message") or ""), f"{d_closed.get('message')}") check("截止后仍能查到任务（码永久有效）", d_closed.get("taskName") == f"E2E落地页 {suffix}", f"{d_closed.get('taskName')}") # ── 7. 限流 ────────────────────────────────────────── print("\n[7] 限流：scene 可穷举，必须挡得住批量扫描") rate_key = None okk, out = redis(["--scan", "--pattern", "xt:mp:scene:rate:*"]) if okk and out.strip(): rate_key = out.strip().splitlines()[0] if rate_key: redis(["DEL", rate_key]) codes = [] for i in range(25): st, b = scene_raw(base, f"t{tenant_id}-k{task_id}") codes.append((st, b.get("code"))) # 前 20 次应成功（code=0），第 21 次起应被限流 passed_n = sum(1 for st, c in codes if c == 0) limited_n = sum(1 for st, c in codes if c == 40071) check("前 20 次正常通过", passed_n == 20, f"通过 {passed_n} 次") check("第 21 次起被限流（40071）", limited_n == 5, f"被限 {limited_n} 次") check("限流用的是项目的 GATE_RATE_LIMITED(40071)，不是自制错误码", any(c == 40071 for _, c in codes), f"{set(c for _, c in codes)}") redis(["DEL", rate_key]) _, after = scene_raw(base, f"t{tenant_id}-k{task_id}") check("清掉计数器后立刻恢复（说明是限流不是封禁）", after.get("code") == 0, f"code={after.get('code')}") else: fail("限流用例", "没找到限流键，说明限流可能没生效") # ── 8. 清理 ────────────────────────────────────────── print("\n[8] 清理测试数据") cleanup_all() ok(f"已清理 {len(CREATED_TASK_IDS)} 个任务、{len(CREATED_TENANT_IDS)} 个测试租户") 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)