lovife
1f3c485ceb
角色与授权（交接摘要 §一·三十五）： - 权限判定合并两个来源：内置角色按代码（BuiltinRole）、自定义角色按库 （sys_role_permission）。实现放 module 层的 DbRoleGrantSource，security 只依赖 接口 RoleGrantSource —— 否则角色服务要调 permissionChecker，两个包迟早成环 - 权限判定必须加缓存：此前是零 IO，而压测基线是"SQL 条数恒定"。缓存放 Redis， 空结果也缓存（否则已删角色的 token 会把库打穿）、Redis 挂了退化成直读库、TTL 只兜底 - V18 建 sys_role.user_type（账号的 user_type 一直从角色反推，自定义角色不在代码映射里） 与 sys_role_menu；8 个接口 + 错误码 40045；契约 119 -> 129 - 内置角色只读（权限码是前后端契约，做成可改 = 一次误操作就能把管理员锁在系统外）； 平台级能力两道防线（界面禁用 + 服务端 40301）；防自锁；改完立即生效（写授权清缓存） - RoleProvisioner 从"已存在就 continue"改成幂等回填 + 每日 03:20 定时任务 —— 修的是"代码改了权限、已开通租户永远停在发放那一刻"，这种漂移不报错也没测试会红 - scripts/e2e-m10-role.py 77/77 需求收敛（交接摘要 §一·三十六，本节的数字为准）： - 删副班主任 / 审核员两个内置角色（6 -> 4）、历史整理器（module/cleanup）、 加入申请（module/join + 学生端 pages/join）、collect_task_org.deputy_user_ids 整列； 审核范围三级回退 -> 两级，连带 6 个 mapper 的 LIKE '%,id,%' 分支 - V19 迁移处理存量：先解 sys_user_role 绑定再软删角色；账号只在"除这两个角色外没有 别的角色"时才软删。必须迁而不能只删代码 —— 角色码删掉后 mergePermissions 返回空集， 存量账号会"能登录、处处 40301"，不抛异常不打日志 - 契约 129 -> 119、错误码 59 -> 54、权限码 30 -> 28、PC 菜单 12 -> 10、 小程序页 10 -> 9、原型 32 -> 29 屏、E2E 39 -> 37 个（断言 1695 条） - UserService.list 对不认识的角色码由"静默丢掉过滤条件"改成失败关闭 （之前传个不存在的 roleCode 会返回全校账号，且看起来完全正常） - 品牌元信息同步改名「深学在线」：pom 名称、/doc.html 标题、配置与启动类注释、 存储自检探针内容；学生端 app 名与全局导航栏标题
686 lines
33 KiB
Python
686 lines
33 KiB
Python
#!/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] 查询与分页")
|
||
|
||
# ⚠️ 这条是 2026-10 需求收敛时补的防线回归：
|
||
# 删掉「副班主任」角色后，前端一度还在传 roleCode=DEPUTY_TEACHER，
|
||
# 而旧实现对**不认识的角色码是静默忽略过滤条件** → 返回全校账号（含学生），
|
||
# 且不报任何错。现在必须**失败关闭**（空列表）。
|
||
status, body = call("GET", f"{base}/api/v1/users?roleCode=DEPUTY_TEACHER", token)
|
||
ghost = (body.get("data") or {}).get("list")
|
||
check("⭐ 按已删除的角色码过滤 → 返回空（不是「忽略过滤条件」返回全校）",
|
||
body.get("code") == 0 and ghost == [],
|
||
f"code={body.get('code')} 条数={len(ghost) if isinstance(ghost, list) else ghost}")
|
||
|
||
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] 修改账号")
|
||
# ⚠️ 2026-10 需求收敛：删掉「副班主任」后，user_type=3 的内置角色只剩班主任一个，
|
||
# 所以这里改成单角色改名，并**另加一条**断言把"多角色必须同属一个用户类型"这条
|
||
# 规则本身验出来（以前拿「班主任 + 副班主任」验的是"两个同类角色都生效"）。
|
||
status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}", token, {
|
||
"realName": teacher_name + "改",
|
||
"phone": teacher_phone,
|
||
"roleCodes": ["HEAD_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 [])
|
||
== ["HEAD_TEACHER"],
|
||
f"roleCodes={row.get('roleCodes') if row else 'N/A'}")
|
||
|
||
status, body = call("PUT", f"{base}/api/v1/users/{teacher_uid}", token, {
|
||
"realName": teacher_name + "改",
|
||
"phone": teacher_phone,
|
||
"roleCodes": ["HEAD_TEACHER", "SCHOOL_ADMIN"],
|
||
"userType": 3,
|
||
})
|
||
check("⭐ 多角色必须同属一个用户类型（班主任 + 学校管理员 → 40001）",
|
||
body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}")
|
||
|
||
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())
|