feat(user): 账号管理(CRUD、Excel 导入、改密)并修掉 5 处缺陷

后端补齐 /users 全套(列表/新增/修改/软删/启停/重置密码/批量导入)与
PUT /auth/password;前端加「组织与账号 → 账号管理」一屏(左组织树 + 右列表、
新增编辑弹窗、导入体检报告、初始密码只显示一次的强确认)。

验收:e2e-m0-user.py 55/55;e2e-m0.py 回归 36/36;typecheck 0 错、build 通过。
账号管理页已在浏览器肉眼确认(首屏与切页均可见、数据与分页正确)。

本轮修掉的 5 个真问题 —— 都不是功能没做完,而是做完了但错的:

1) 学校租户也发放 PLATFORM_ADMIN 角色行
   学校管理员在新增账号时传 roleCodes=[PLATFORM_ADMIN] 就能造出带
   tenant:manage 的账号,而 /tenants 只校验权限码、不校验是否真在平台租户下
   → 任何学校管理员都能开通租户、拉全平台租户列表(租户级越权到平台级)。
   RoleProvisioner 不再给非平台租户发该角色行 + 服务端按租户白名单校验。

2) UserUpsertReq.username 的 @NotBlank 对修改也生效
   PUT /users/{id} 永远返回 40001,且明细只说存在阻断性校验问题,
   看不出是登录名 —— 改账号功能整体不可用。去掉该注解,
   必填性改由 UserService.create 判定。

3) Main.vue 的 <transition appear>
   首次渲染被当成一次过渡,元素被加上 zoom-fade-enter-from(opacity:0)后
   永不清理 → DOM 内容齐全、接口数据正常、文字可选中,但整块内容区肉眼不可见。
   极易误判成页面没渲染而往路由/接口方向排查。去掉 appear。

4) assertAnotherAdminExists 在停用/删除路径上不可达
   能调这两个接口的人自己必然是活跃管理员,所以除目标外还有管理员恒成立;
   而真正会锁死租户的路径 —— 把管理员改成普通角色(目标可以是操作者自己)——
   当时没有拦。补 losingAdminRole() + UserMapper.countAdminRoles(),
   并在方法注释里写明哪条路径真生效、哪条只是防御性的。

5) sys_user.user_type 有两套互相冲突的枚举
   DDL/实体注释写「1教师/管理员 2填报人(学生) 3社会人员」,而权威契约
   (03-接口设计 §2.7)与 BuiltinRole.UserType 是
   「1平台运营 2学校管理员 3班主任 4学生」—— 2 和 3 完全错位。
   后果:userType=3(班主任,最高频的账号)被当成社会人员(二期)直接拒掉,
   建班主任账号会失败;前端还把学校管理员显示成填报人(学生)。
   改法:后端从角色反推用户类型再比对;新增 V3 迁移更正库注释
   (V1 已应用不能改,Flyway 会校验 checksum);更正 PRD 与 schema DDL 的同一行;
   前端「用户类型」从手选改为由角色推导,从根上消除矛盾组合;
   E2E 加 3 条回归断言锁住这个语义。

顺带修掉:
  · components/index.ts 里 6 个指向已删组件的悬空导出(构建不报错、用到才炸)
  · tsconfig 的 lib 停在 ES2020 而代码用了 Array.prototype.at
  · 两处 .at() 的空值风险(useBreakpoint / useTabsStore)
  · dashboard 的 userStore.menus 拼错,导致平台判断恒为 false
  · http.ts 里两个永不命中的错误码常量
  · 账号列表的 fixed 列缺 :scroll 配置,会浮在其它列上面
  · .gitignore 补 __pycache__

新增脚本:
  · scripts/e2e-m0-user.py            账号管理 55 项验收(含越权与自锁的负向断言)
  · scripts/dev-clean-test-users.py   清理 E2E 残留的测试账号
This commit is contained in:
2026-09-28 22:25:01 +08:00
parent 87ece5a971
commit 8480f431ba
52 changed files with 4264 additions and 122 deletions
+101
View File
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-
"""
清理 E2E 崩掉时残留在 demo 租户里的测试账号。
背景:e2e-m0-user.py 在 [5] 自锁防线那一段曾因脚本自身 bug 抛异常退出,
没有走到 [14] 的清理,于是 demo 租户里留下了 `dup*` / `t*` / `imp*` 之类的测试账号。
它们会出现在账号管理页上,必须清掉。
"""
import base64
import json
import os
import re
import sys
import urllib.error
import urllib.request
sys.stdout.reconfigure(encoding='utf-8')
BASE = "http://localhost:8080"
PUB = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"deploy", "rsa-public-key-dev.pem")
# E2E 造账号时用的前缀(见 scripts/e2e-m0-user.py)
TEST_PATTERNS = [
re.compile(r'^dup[0-9a-f]{6}$'), # 重名测试
re.compile(r'^t[0-9a-f]{6}$'), # 测试班主任
re.compile(r'^ht[0-9a-f]{6}$'), # 班主任回归护栏
re.compile(r'^imp[123][0-9a-f]{6}$'), # 导入进来的
re.compile(r'^esc2?[0-9a-f]{6}$'), # 提权测试(正常不该存在)
re.compile(r'^mix[0-9a-f]{6}$'), # 类型不符测试(正常不该存在)
re.compile(r'^bad[0-9a-f]{6}$'), # 手机号错(正常不该存在)
]
def rsa(plain: str) -> str:
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
with open(PUB, "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, path: str, token: str | None = None, body: dict | None = None):
headers = {"Content-Type": "application/json; charset=utf-8"}
if token:
headers["Authorization"] = f"Bearer {token}"
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=25) as r:
return json.loads(r.read().decode())
except urllib.error.HTTPError as e:
try:
return json.loads(e.read().decode())
except Exception:
return {"code": e.code}
# 登录
r = call("POST", "/api/v1/auth/login",
body={"username": "admin", "password": rsa("Admin@123"), "tenantCode": "demo"})
if r.get("code") != 0:
sys.exit(f"登录失败:{r}")
token = r["data"]["token"]
admin_uid = r["data"]["user"]["id"]
print(f" 已登录(admin uid={admin_uid})")
# 拉全量
r = call("GET", "/api/v1/users?pageSize=200", token)
users = (r.get("data") or {}).get("list") or []
print(f" demo 租户共 {len(users)} 个账号")
victims = []
for u in users:
if u["id"] == admin_uid:
continue
uname = u.get("username") or ""
if any(p.match(uname) for p in TEST_PATTERNS):
victims.append(u)
if not victims:
print(" ✓ 没有残留的测试账号")
sys.exit(0)
print(f" 发现 {len(victims)} 个残留测试账号:")
for v in victims:
print(f" id={v['id']:<4} username={v.get('username'):<14} realName={v.get('realName')}")
print()
for v in victims:
res = call("DELETE", f"/api/v1/users/{v['id']}", token)
flag = "✓" if res.get("code") == 0 else "✗"
print(f" {flag} 删除 {v.get('username')} code={res.get('code')}")
# 复核
r = call("GET", "/api/v1/users?pageSize=200", token)
left = (r.get("data") or {}).get("list") or []
print()
print(f" 清理后剩 {len(left)} 个账号:")
for u in left:
print(f" id={u['id']:<4} username={u.get('username'):<14} realName={u.get('realName')} 角色={u.get('roleNames')}")
+661
View File
@@ -0,0 +1,661 @@
#!/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())