#!/usr/bin/env python # -*- coding: utf-8 -*- """ 账号管理（M0 剩余）验收测试。 把《03-接口设计》§5.6~§5.9 的账号接口 + 《05-开发任务拆解》里 M0 的验收标准 变成可重复执行的程序： [x] 账号增删改查、启用/停用、重置密码、修改自己密码 [x] Excel 批量导入（表头不固定时能自动定位，并回传体检报告） [x] 权限：没有 user:view 的人看不到账号列表 [x] **越权防线**：学校管理员不能给自己人分配平台运营角色 [x] **自锁防线**：不能停用自己、不能取消本单位最后一个管理员 用法 ---- python scripts/e2e-m0-user.py python scripts/e2e-m0-user.py --base http://localhost:8080 退出码：0 = 全部通过；1 = 有失败项（可直接用于 CI）。 依赖：cryptography（RSA 加密登录密码）、openpyxl（构造导入用的 Excel） """ 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") # 与 application-dev.yml 的 REDIS_PASSWORD 默认值一致，仅用于 E2E 直写锁键 REDIS_PWD = os.environ.get("REDIS_PASSWORD", "redis_dev_pwd") PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] # 本次测试创建出来的账号，收尾时清理 CREATED_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 show(body: dict, indent: int = 9) -> None: """ 打印完整响应体。 有些失败只看 msg 是查不动的：参数校验失败时 msg 只有一句 "存在阻断性校验问题，无法提交"，真正是哪个字段错了在 data.problems 里。 """ pad = " " * indent text = json.dumps(body, ensure_ascii=False) print(f"{pad}完整响应：{text[:400]}{'…' if len(text) > 400 else ''}") def redis_set(key: str, value: str, ttl_seconds: int) -> bool: """ 直接往 Redis 里写一个键。 只用在「验证登录锁定会被重置密码解除」这一项上： 正常路径下要触发锁定得先连续失败、再通过图形验证码继续失败 —— 脚本没法自动识别验证码图片，所以这里直接把锁键写进去， 模拟"本人密码忘了、反复试错把账号试锁了"这个真实场景。 """ try: r = subprocess.run( ["docker", "exec", "xt-redis", "redis-cli", "-a", REDIS_PWD, "--no-auth-warning", "SET", key, value, "EX", str(ttl_seconds)], capture_output=True, text=True, timeout=25) return r.returncode == 0 and "OK" in (r.stdout or "") except Exception: return False # ── HTTP ──────────────────────────────────────────── def rsa_encrypt(plain: str) -> str: try: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding except ImportError: sys.exit("缺少依赖：pip install cryptography") with open(PUB_KEY, "rb") as f: public_key = serialization.load_pem_public_key(f.read()) cipher = public_key.encrypt(plain.encode("utf-8"), padding.PKCS1v15()) return base64.b64encode(cipher).decode("ascii") def call(method: str, url: str, token: str | None = None, body: dict | None = None, timeout: int = 30) -> tuple[int, dict]: """返回 (http_status, 响应体)。业务失败也是 200 + code!=0。""" 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 post_multipart(url: str, token: str, file_path: str, field: str = "file") -> tuple[int, dict]: """multipart 上传（导入接口用）。""" boundary = "----xt" + uuid.uuid4().hex filename = os.path.basename(file_path) with open(file_path, "rb") as f: content = f.read() parts: list[bytes] = [] parts.append(f"--{boundary}\r\n".encode()) parts.append( f'Content-Disposition: form-data; name="{field}"; filename="{filename}"\r\n'.encode()) parts.append(b"Content-Type: application/vnd.openxmlformats-officedocument." b"spreadsheetml.sheet\r\n\r\n") parts.append(content) parts.append(f"\r\n--{boundary}--\r\n".encode()) payload = b"".join(parts) req = urllib.request.Request( url, data=payload, method="POST", headers={ "Content-Type": f"multipart/form-data; boundary={boundary}", "Authorization": f"Bearer {token}", }) try: with urllib.request.urlopen(req, timeout=60) 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_code: str | None = None) -> tuple[int, dict]: payload = {"username": username, "password": rsa_encrypt(password)} if tenant_code: payload["tenantCode"] = tenant_code return call("POST", f"{base}/api/v1/auth/login", body=payload) 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"), help="学校租户的租户码（用 admin 登录时需要）") ap.add_argument("--admin", default="admin") ap.add_argument("--password", default="Admin@123") args = ap.parse_args() base = args.base.rstrip("/") print("=" * 70) print(f"账号管理 E2E　base={base}　租户={args.tenant}　管理员={args.admin}") print("=" * 70) # ── 0. 登录 ──────────────────────────────────────── print("\n[0] 准备：管理员登录") status, body = login(base, args.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"] admin_uid = body["data"]["user"]["id"] ok(f"管理员登录（uid={admin_uid}，角色={body['data']['roleCodes']}）") # ── 1. 权限 ──────────────────────────────────────── print("\n[1] 访问控制") status, body = call("GET", f"{base}/api/v1/users") check("未登录访问 /users 返回 40101", body.get("code") == 40101, f"实际 code={body.get('code')}") # ── 2. 新增 ──────────────────────────────────────── print("\n[2] 新增账号") suffix = uuid.uuid4().hex[:6] teacher_name = f"测试班主任{suffix}" teacher_user = f"t{suffix}" teacher_phone = "13800138001" status, body = call("POST", f"{base}/api/v1/users", token, { "username": teacher_user, "realName": teacher_name, "phone": teacher_phone, "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) created = body.get("data") or {} if not check("新增班主任账号成功", body.get("code") == 0, f"HTTP {status} code={body.get('code')} msg={body.get('msg')}"): return 1 teacher_uid = created["id"] teacher_pwd = created["initialPassword"] CREATED_IDS.append(teacher_uid) check("返回了初始密码", bool(teacher_pwd), "initialPassword 为空") print(f" → id={teacher_uid} 账号={teacher_user} 初始密码={teacher_pwd}") # 用初始密码登录 status, body = login(base, teacher_user, teacher_pwd, args.tenant) if not check("新账号能用初始密码登录", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"): return 1 teacher_token = body["data"]["token"] check("新账号拿到 HEAD_TEACHER 角色", "HEAD_TEACHER" in (body["data"]["roleCodes"] or []), f"roleCodes={body['data']['roleCodes']}") check("班主任没有 user:view，访问 /users 被拒", call("GET", f"{base}/api/v1/users", teacher_token)[1].get("code") == 40301, "班主任竟然能看账号列表") # ── 3. 字段校验 ──────────────────────────────────── print("\n[3] 字段校验") status, body = call("POST", f"{base}/api/v1/users", token, { "username": "dup" + suffix, "realName": "重名测试", "phone": "13800138002", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) if body.get("code") == 0: CREATED_IDS.append(body["data"]["id"]) status, body = call("POST", f"{base}/api/v1/users", token, { "username": teacher_user, "realName": "重复账号", "phone": "13800138003", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) check("账号重复被拒", body.get("code") != 0, f"重复账号竟然建成功了 code={body.get('code')}") status, body = call("POST", f"{base}/api/v1/users", token, { "username": "bad" + suffix, "realName": "手机号错", "phone": "1380013", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) check("手机号格式错被拒", body.get("code") != 0, f"非法手机号竟然通过了 code={body.get('code')}") # ── 4. 越权防线（最关键）────────────────────────── print("\n[4] 越权防线：学校管理员不能分配平台运营角色") # 学校管理员用自己租户的 token 建账号，并试图授予 PLATFORM_ADMIN。 # 若不拦，这个账号会带上 tenant:manage，而 /tenants 只校验权限码、 # 不校验"是否真的在平台租户下" → 学校管理员即可开通租户、拉全平台租户列表。 status, body = call("POST", f"{base}/api/v1/users", token, { "username": "esc" + suffix, "realName": "提权测试", "phone": "13800138004", "roleCodes": ["PLATFORM_ADMIN"], "userType": 3, }) check("分配 PLATFORM_ADMIN 被拒（40301）", body.get("code") == 40301, f"竟然放行了！code={body.get('code')} msg={body.get('msg')}") check("该账号没有被创建出来", call("GET", f"{base}/api/v1/users?keyword=esc{suffix}", token)[1] .get("data", {}).get("total", -1) == 0, "越权账号已经落库") status, body = call("POST", f"{base}/api/v1/users", token, { "username": "esc2" + suffix, "realName": "提权测试2", "phone": "13800138005", "roleCodes": ["不存在的角色"], "userType": 3, }) check("未知角色被拒", body.get("code") != 0, f"未知角色竟然通过了 code={body.get('code')}") # 用户类型与角色必须自洽（§2.7 的映射：SCHOOL_ADMIN→2、HEAD_TEACHER→3） # 这里是"给了 3 却是学校管理员角色"的组合，必须被拒 status, body = call("POST", f"{base}/api/v1/users", token, { "username": "mix" + suffix, "realName": "类型不符", "phone": "13800138006", "roleCodes": ["SCHOOL_ADMIN"], "userType": 3, }) check("「用户类型 3（班主任）+ 学校管理员角色」被拒", body.get("code") != 0, f"矛盾组合竟然通过了 code={body.get('code')}") # 回归护栏：班主任 = 用户类型 3。 # ⚠️ 曾经因为照 sys_user 的旧列注释（3=社会人员）写校验， # 把「3」当成一期不支持的类型直接拒掉 —— 而 3 恰恰是最常用的班主任。 # 这条断言就是防止那个误判再回来。 status, body = call("POST", f"{base}/api/v1/users", token, { "username": "ht" + suffix, "realName": "班主任回归", "phone": "13800138007", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) check("班主任（用户类型 3）能正常创建 —— 不是「社会人员（二期）」", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") if body.get("code") == 0: CREATED_IDS.append(body["data"]["id"]) # 角色与用户类型两个字段都要回填正确 st2, bd2 = call("GET", f"{base}/api/v1/users?keyword=ht{suffix}", token) row2 = next((h for h in (bd2.get("data") or {}).get("list", []) if h["id"] == body["data"]["id"]), None) check("列表回填的 userType 也是 3", row2 is not None and row2.get("userType") == 3, f"userType={row2.get('userType') if row2 else 'N/A'}") # ── 5. 自锁防线 ──────────────────────────────────── print("\n[5] 自锁防线") status, body = call("PUT", f"{base}/api/v1/users/{admin_uid}/status?status=0", token) check("不能停用当前登录的账号", body.get("code") != 0, f"竟然停用了自己 code={body.get('code')}") status, body = call("DELETE", f"{base}/api/v1/users/{admin_uid}", token) check("不能软删当前登录的账号", body.get("code") != 0, f"竟然删了自己 code={body.get('code')}") # 唯一管理员把自己降级成普通角色 → 会锁死租户，必须拦 # （realName 从 /auth/me 取，不要复用上面那个已经被 DELETE 用掉的 body） _, me = call("GET", f"{base}/api/v1/auth/me", token) admin_name = (((me or {}).get("data") or {}).get("user") or {}).get("realName") or args.admin status, body = call("PUT", f"{base}/api/v1/users/{admin_uid}", token, { "realName": admin_name, "phone": "13800138000", "roleCodes": ["HEAD_TEACHER"], "userType": 3, }) check("不能取消本单位最后一个管理员的角色", body.get("code") != 0, f"竟然把自己降级了 —— 这个租户从此没人能管账号 code={body.get('code')}") # ── 6. 查询 ──────────────────────────────────────── print("\n[6] 查询与分页") status, body = call("GET", f"{base}/api/v1/users?pageNum=1&pageSize=5", token) data = body.get("data") or {} check("列表返回 {list,total,pageNum,pageSize}", isinstance(data.get("list"), list) and "total" in data and data.get("pageNum") == 1 and data.get("pageSize") == 5, f"实际 keys={list(data.keys())}") status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode({"keyword": teacher_name}), token) hits = (body.get("data") or {}).get("list") or [] check("按姓名关键字能搜到", any(h["id"] == teacher_uid for h in hits), f"命中 {len(hits)} 条") status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode({"keyword": teacher_phone}), token) hits = (body.get("data") or {}).get("list") or [] check("按完整手机号能搜到（走 HMAC 等值匹配）", any(h["id"] == teacher_uid for h in hits), f"命中 {len(hits)} 条") status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode({"keyword": teacher_phone[3:8]}), token) hits = (body.get("data") or {}).get("list") or [] check("手机号只给中间几位搜不到（加密的必然取舍，不是 bug）", not any(h["id"] == teacher_uid for h in hits), "居然搜到了 —— 说明手机号不是加密存的") status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode({"roleCode": "HEAD_TEACHER"}), token) hits = (body.get("data") or {}).get("list") or [] check("按角色过滤能搜到", any(h["id"] == teacher_uid for h in hits), f"命中 {len(hits)} 条") # 手机号必须脱敏 status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode({"keyword": teacher_user}), token) row = next((h for h in (body.get("data") or {}).get("list", []) if h["id"] == teacher_uid), None) if not check("列表里的手机号是脱敏值", row is not None and row.get("phoneMasked") == "138****8001", f"phoneMasked={row.get('phoneMasked') if row else 'N/A'}"): show(body) # ── 7. 修改 ──────────────────────────────────────── print("\n[7] 修改账号") status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}", token, { "realName": teacher_name + "改", "phone": teacher_phone, "roleCodes": ["HEAD_TEACHER", "DEPUTY_TEACHER"], "userType": 3, }) check("修改姓名与角色成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") if body.get("code") != 0: show(body) status, body = call("GET", f"{base}/api/v1/users?keyword={teacher_user}", token) row = next((h for h in (body.get("data") or {}).get("list", []) if h["id"] == teacher_uid), None) check("两个角色都生效", row is not None and sorted(row.get("roleCodes") or []) == ["DEPUTY_TEACHER", "HEAD_TEACHER"], f"roleCodes={row.get('roleCodes') if row else 'N/A'}") check("登录名没有被改掉（username 不可改）", row is not None and row.get("username") == teacher_user, f"username={row.get('username') if row else 'N/A'}") # ── 8. 启用 / 停用 ───────────────────────────────── print("\n[8] 启用 / 停用") status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}/status?status=0", token) check("停用成功", body.get("code") == 0, f"code={body.get('code')}") status, body = login(base, teacher_user, teacher_pwd, args.tenant) check("停用后无法登录", body.get("code") != 0, f"停用账号竟然登录成功 code={body.get('code')}") status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}/status?status=1", token) check("启用成功", body.get("code") == 0, f"code={body.get('code')}") status, body = login(base, teacher_user, teacher_pwd, args.tenant) check("启用后能重新登录", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}/status?status=9", token) check("非法状态值被拒", body.get("code") != 0, f"status=9 竟然通过了 code={body.get('code')}") # ── 9. 重置密码 + 锁定解除 ───────────────────────── print("\n[9] 重置密码（含登录锁定解除）") # 9.1 验证码闸门：连续失败 2 次后再登录会要求图形验证码 for _ in range(3): login(base, teacher_user, "WrongPassword123", args.tenant) status, body = login(base, teacher_user, teacher_pwd, args.tenant) check("连续失败后要求图形验证码（40001）", body.get("code") == 40001 and "验证码" in (body.get("msg") or ""), f"code={body.get('code')} msg={body.get('msg')}") # 9.2 锁定与解除 # 正常路径下要触发锁定得"连续失败 + 通过验证码继续失败"， # 脚本没法自动识别验证码图片，所以直接把锁键写进 Redis 来复现真实场景 # （本人忘密码、反复试错把账号试锁了）。 lock_key = f"xt:login:lock:{teacher_user}" if redis_set(lock_key, "1", 600): status, body = login(base, teacher_user, teacher_pwd, args.tenant) check("账号被锁定时登录返回 40013", body.get("code") == 40013, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("POST", f"{base}/api/v1/users/{teacher_uid}/reset-password", token, {"mode": "GENERATE"}) new_pwd = (body.get("data") or {}).get("newPassword") check("重置密码成功并返回新密码", body.get("code") == 0 and bool(new_pwd), f"code={body.get('code')} msg={body.get('msg')}") if body.get("code") != 0: show(body) status, body = login(base, teacher_user, new_pwd, args.tenant) check("重置后立即可登录 —— 锁定被一并解除", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}" "（若为 40013，说明重置密码没有清 LoginGuard）") else: print(" [SKIP] 连不上 xt-redis，跳过「锁定与解除」两项") status, body = call("POST", f"{base}/api/v1/users/{teacher_uid}/reset-password", token, {"mode": "GENERATE"}) new_pwd = (body.get("data") or {}).get("newPassword") check("重置密码成功并返回新密码", body.get("code") == 0 and bool(new_pwd), f"code={body.get('code')} msg={body.get('msg')}") status, body = login(base, teacher_user, teacher_pwd, args.tenant) check("旧密码失效", body.get("code") != 0, f"旧密码还能登录 code={body.get('code')}") status, body = call("POST", f"{base}/api/v1/users/{teacher_uid}/reset-password", token, {"mode": "CUSTOM", "newPassword": "abc"}) check("自定义弱密码被拒", body.get("code") != 0, f"弱密码竟然通过了 code={body.get('code')}") # ── 10. 修改自己的密码 ───────────────────────────── print("\n[10] 修改自己的密码") status, body = login(base, teacher_user, new_pwd, args.tenant) t2 = body["data"]["token"] status, body = call("PUT", f"{base}/api/v1/auth/password", t2, { "oldPassword": rsa_encrypt("WrongOld123"), "newPassword": rsa_encrypt("NewPass456"), }) check("原密码错被拒", body.get("code") != 0, f"原密码错竟然改了密码 code={body.get('code')}") status, body = call("PUT", f"{base}/api/v1/auth/password", t2, { "oldPassword": rsa_encrypt(new_pwd), "newPassword": rsa_encrypt("NewPass456"), }) check("修改自己的密码成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = login(base, teacher_user, "NewPass456", args.tenant) check("能用新密码登录", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") # ── 11. Excel 批量导入 ───────────────────────────── print("\n[11] Excel 批量导入（表头不在第 1 行）") import_path = build_import_xlsx(suffix, base, token) if import_path: status, body = post_multipart(f"{base}/api/v1/users/import", token, import_path) rep = body.get("data") or {} if check("导入接口返回体检报告", body.get("code") == 0, f"HTTP {status} code={body.get('code')} msg={body.get('msg')}"): print(f" → total={rep.get('total')} success={rep.get('success')} " f"failed={rep.get('failed')} headerRow={rep.get('headerRow')}") print(f" → 表头原文：{rep.get('headerText')}") cols = "、".join(f"{c['fieldName']}→{c['column']}列" for c in rep.get("columns") or []) print(f" → 列映射：{cols}") check("自动定位到第 3 行表头（前两行是标题与说明）", rep.get("headerRow") == 3, f"headerRow={rep.get('headerRow')}") check("回传了表头原文", bool(rep.get("headerText")), "headerText 为空 —— 界面无法让用户核对") check("回传了列映射且含姓名/账号/手机号", {"realName", "username", "phone"}.issubset( {c["field"] for c in rep.get("columns") or []}), f"columns={rep.get('columns')}") check("3 条合法行全部导入", rep.get("success") == 3, f"success={rep.get('success')}") check("4 条非法行被逐条报出原因", rep.get("failed") == 4, f"failed={rep.get('failed')}") err = rep.get("errors") or [] print(" → 失败明细：") for e in err: print(f" 第 {e['row']} 行 {e['field']}={e['value']}：{e['reason']}") check("失败明细带行号与原因", all( e.get("row") and e.get("reason") for e in err), f"errors={err}") check("提供了 3 个新账号的初始密码", len(rep.get("initialPasswords") or []) == 3, f"initialPasswords={len(rep.get('initialPasswords') or [])}") for u in rep.get("initialPasswords") or []: CREATED_IDS.append(u["username"]) # 导入的账号能用初始密码登录 if rep.get("initialPasswords"): first = rep["initialPasswords"][0] status, body = login(base, first["username"], first["password"], args.tenant) check("导入的账号能用初始密码登录", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") else: ok("（跳过后续导入断言）") try: os.remove(import_path) except OSError: pass # ── 12. 软删 ─────────────────────────────────────── print("\n[12] 软删") status, body = call("DELETE", f"{base}/api/v1/users/{teacher_uid}", token) check("软删成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}") status, body = call("GET", f"{base}/api/v1/users?keyword={teacher_user}", token) hits = (body.get("data") or {}).get("list") or [] check("软删后列表里查不到", not any(h["id"] == teacher_uid for h in hits), f"还能查到 {len(hits)} 条") status, body = login(base, teacher_user, "NewPass456", args.tenant) check("软删后无法登录", body.get("code") != 0, f"软删账号竟然登录成功 code={body.get('code')}") # ── 13. 平台运营角色不再发给学校租户 ──────────────── print("\n[13] 平台运营角色不给学校租户发放") status, body = call( "GET", f"{base}/api/v1/users?" + urllib.parse.urlencode( {"roleCode": "PLATFORM_ADMIN", "pageSize": 200}), token) platform_holders = (body.get("data") or {}).get("list") or [] check("本学校租户里没有任何账号持有平台运营角色", len(platform_holders) == 0, f"有 {len(platform_holders)} 个：" f"{[h['username'] for h in platform_holders]}") # ── 收尾 ─────────────────────────────────────────── print("\n[14] 清理测试数据") for uid in CREATED_IDS: if isinstance(uid, int): call("DELETE", f"{base}/api/v1/users/{uid}", token) else: st, bd = call("GET", f"{base}/api/v1/users?keyword={uid}&pageSize=50", token) for h in (bd.get("data") or {}).get("list") or []: call("DELETE", f"{base}/api/v1/users/{h['id']}", token) ok(f"已尝试清理 {len(CREATED_IDS)} 个测试账号") # ── 汇总 ─────────────────────────────────────────── print("\n" + "=" * 70) 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("=" * 70) return 1 if FAILED else 0 def build_import_xlsx(suffix: str, base: str, token: str) -> str | None: """ 构造一个「表头不在第 1 行」的导入文件。 真实场景就长这样：上面两行是标题和制表说明，第 3 行才是表头。 顺手混进 4 行有问题的数据，用来验证体检报告能把它们逐条点出来。 """ try: from openpyxl import Workbook except ImportError: print(" [SKIP] 未安装 openpyxl，跳过导入测试") return None # 先取一个真实存在的班级名，避免"组织找不到"这条做成假失败 org_name = "" st, bd = call("GET", f"{base}/api/v1/orgs/tree", token) def walk(nodes): nonlocal org_name for n in nodes or []: if n.get("orgType") == 3 and not org_name: org_name = n["orgName"] walk(n.get("children")) walk((bd.get("data") or []) if st == 200 else []) wb = Workbook() ws = wb.active ws.title = "教师账号" ws.append(["24中技数控1班 教师账号表"]) # 第 1 行：标题 ws.append([None, None, "制表：教务处", None, None]) # 第 2 行：说明 ws.append(["姓名", "账号", "手机号", "班级", "角色"]) # 第 3 行：表头 # 3 条合法 ws.append([f"导入甲{suffix}", f"imp1{suffix}", "13900139001", org_name, "班主任"]) ws.append([f"导入乙{suffix}", f"imp2{suffix}", "13900139002", org_name, None]) ws.append([f"导入丙{suffix}", f"imp3{suffix}", "13900139003", None, "副班主任"]) # 4 条非法 ws.append(["手机号错", f"imp4{suffix}", "1390013", None, None]) ws.append(["账号重复", f"imp1{suffix}", "13900139005", None, None]) ws.append(["角色不存在", f"imp5{suffix}", "13900139006", None, "不存在角色"]) ws.append(["组织找不到", f"imp6{suffix}", "13900139007", "没有这个班", None]) path = os.path.join(os.path.dirname(os.path.abspath(__file__)), f"_tmp_import_{suffix}.xlsx") wb.save(path) print(f" 构造的导入文件：表头在第 3 行，7 行数据（3 合法 / 4 非法）") return path if __name__ == "__main__": raise SystemExit(main())