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 名与全局导航栏标题
459 lines
25 KiB
Python
459 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""角色与授权 E2E（二期批次 1~4）。
|
||
|
||
这个脚本验的不是"接口返回 200"，而是**"改完之后，系统真的按新授权做事"** ——
|
||
权限系统最容易出的错是"看起来配好了、其实没生效"，而这个脚本专门盯这一类：
|
||
|
||
| # | 要证明的事 | 怎么证明 |
|
||
|---|---|---|
|
||
| 1 | 自定义角色的**权限码真的在挡** | 给自定义角色只授 `task:view` → 该账号 GET /tasks 通、POST /tasks 得 40301 |
|
||
| 2 | 改授权**立即生效**（不必重新登录） | 用**同一个 token** 改完授权马上再请求，权限随之变化 |
|
||
| 3 | 停用角色**立即生效** | 停用后同一个 token 立刻失去该角色的权限 |
|
||
| 4 | 菜单跟着角色走 | `/auth/me` 的 menus 随角色配置增减（登录响应里也必须有，否则"刷新一次才看得见"） |
|
||
| 5 | 内置角色**只读** | 改/删/授权/停用内置角色一律 40001 |
|
||
| 6 | 越权防线在**服务端** | 学校租户授平台级权限码 → 40301；建"平台运营"类角色 → 40301；发 PLATFORM_ADMIN → 40301 |
|
||
| 7 | 内置角色的权限**以代码为准** | 人为删掉库里的一行授权 → 接口返回的仍是代码那一套；跑一次同步任务 → 库里那行回来 |
|
||
| 8 | 防自锁 | 管理员把自己唯一一条 `role:manage` / `role` 菜单去掉 → 40001（而不是真的锁死） |
|
||
| 9 | 删除守卫 | 还有账号在用的角色删不掉（40045）；解绑后能删，且账号立刻失去权限 |
|
||
|
||
跑法：python scripts/e2e-m10-role.py
|
||
前置：后端 8080 在跑、xt-pg / xt-redis 在跑（与其它 e2e 相同）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
BASE = "http://localhost:8080"
|
||
|
||
PASSED: list[str] = []
|
||
FAILED: list[tuple[str, str]] = []
|
||
CREATED_USERS: list[int] = []
|
||
CREATED_ROLES: list[int] = []
|
||
|
||
|
||
def _load(module_name: str, filename: str):
|
||
path = os.path.join(ROOT, "scripts", filename)
|
||
spec = importlib.util.spec_from_file_location(module_name, path)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
R = _load("e2e_m2_roster", "e2e-m2-roster.py")
|
||
|
||
|
||
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) -> str:
|
||
_, out = R.sql(stmt)
|
||
return out
|
||
|
||
|
||
def body_of(method: str, path: str, token: str | None = None,
|
||
payload: dict | None = None) -> dict:
|
||
_, b = R.call(method, f"{BASE}{path}", token, payload)
|
||
return b
|
||
|
||
|
||
def clear_login_fail(*accounts: str) -> None:
|
||
"""清掉登录失败计数：脚本反复跑会把账号锁掉，而锁定看起来像"密码错了"。"""
|
||
for a in accounts:
|
||
os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning '
|
||
'DEL xt:login:fail:%s > NUL 2>&1' % a)
|
||
|
||
|
||
def login(username: str, password: str, tenant: str | None = "demo"):
|
||
_, b = R.login(BASE, username, password, tenant)
|
||
if b.get("code") != 0:
|
||
return None, b
|
||
return b["data"]["token"], b["data"]
|
||
|
||
|
||
def role_by_code(token: str, code: str) -> dict:
|
||
for r in (body_of("GET", "/api/v1/roles", token).get("data") or []):
|
||
if r.get("roleCode") == code:
|
||
return r
|
||
return {}
|
||
|
||
|
||
def create_role(token: str, code: str, name: str, user_type: int,
|
||
perms: list[str], menus: list[str]) -> dict:
|
||
b = body_of("POST", "/api/v1/roles", token, {
|
||
"roleCode": code, "roleName": name, "userType": user_type,
|
||
"permCodes": perms, "menus": menus,
|
||
})
|
||
if b.get("code") == 0 and b.get("data"):
|
||
CREATED_ROLES.append(b["data"])
|
||
return b
|
||
|
||
|
||
def create_user(token: str, username: str, real_name: str, phone: str,
|
||
role_codes: list[str], user_type: int) -> dict:
|
||
b = body_of("POST", "/api/v1/users", token, {
|
||
"username": username, "realName": real_name, "phone": phone,
|
||
"roleCodes": role_codes, "userType": user_type,
|
||
})
|
||
if b.get("code") == 0 and (b.get("data") or {}).get("id"):
|
||
CREATED_USERS.append(b["data"]["id"])
|
||
return b
|
||
|
||
|
||
def cleanup() -> None:
|
||
"""只删本脚本造的东西（演示租户里别人的数据一律不碰）。"""
|
||
for uid in CREATED_USERS:
|
||
R.sql(f"delete from sys_user_role where user_id={uid};")
|
||
R.sql(f"delete from sys_user where id={uid};")
|
||
for rid in CREATED_ROLES:
|
||
R.sql(f"delete from sys_role_permission where role_id={rid};")
|
||
R.sql(f"delete from sys_role_menu where role_id={rid};")
|
||
R.sql(f"delete from sys_role where id={rid};")
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="角色与授权 E2E")
|
||
ap.parse_args()
|
||
|
||
print("=" * 82)
|
||
print(" 角色与授权 E2E（自定义角色真的生效 / 内置角色只读 / 越权与自锁防线）")
|
||
print("=" * 82)
|
||
|
||
suffix = os.urandom(3).hex()
|
||
role_a_code = f"E2E_ROLE_A_{suffix.upper()}"
|
||
role_b_code = f"E2E_ROLE_B_{suffix.upper()}"
|
||
user_a = f"e2e_role_a_{suffix}"
|
||
user_b = f"e2e_role_b_{suffix}"
|
||
clear_login_fail("admin", "platform", user_a, user_b)
|
||
|
||
at, adm = login("admin", "Admin@123")
|
||
if not at:
|
||
print(f" x 管理员登录失败：{adm}")
|
||
return 2
|
||
admin_uid = adm["user"]["id"]
|
||
|
||
# ══════════════ 1. 权限码目录 ══════════════
|
||
print("\n[1] 权限码目录（角色授权界面的勾选项来源）")
|
||
b = body_of("GET", "/api/v1/permissions", at)
|
||
groups = b.get("data") or []
|
||
check("权限码目录可读且按分组返回", b.get("code") == 0 and len(groups) >= 5,
|
||
json.dumps(b, ensure_ascii=False)[:200])
|
||
items = [i for g in groups for i in (g.get("items") or [])]
|
||
check("目录里 28 个权限码一个不少（删掉 join:audit / cleanup:run 之后）",
|
||
len(items) == 28, f"实际 {len(items)}")
|
||
by_code = {i.get("code"): i for i in items}
|
||
check("每条权限码都带中文名与适用端",
|
||
all(i.get("label") and i.get("platform") for i in items))
|
||
check("⭐ 平台级权限被标出来（tenant:manage → platformOnly=true）",
|
||
by_code.get("tenant:manage", {}).get("platformOnly") is True)
|
||
check("role:manage 不是平台级（学校管理员可以授予）",
|
||
by_code.get("role:manage", {}).get("platformOnly") is False)
|
||
|
||
# ══════════════ 2. 内置角色：只读 + 权限来自代码 ══════════════
|
||
print("\n[2] 内置角色：看得见、改不了")
|
||
roles = body_of("GET", "/api/v1/roles", at).get("data") or []
|
||
check("角色列表可读（内置 4 个 + 可能的自定义角色）", len(roles) >= 4, f"实际 {len(roles)} 个")
|
||
check("⭐ 内置角色一律 editable=false（权限由代码决定）",
|
||
all(r.get("editable") is False for r in roles if r.get("roleType") == 1),
|
||
json.dumps([(r.get("roleCode"), r.get("editable")) for r in roles],
|
||
ensure_ascii=False)[:220])
|
||
sa = role_by_code(at, "SCHOOL_ADMIN")
|
||
check("内置角色的权限码取代码定义（含 role:manage / task:manage）",
|
||
"role:manage" in (sa.get("permCodes") or []) and "task:manage" in (sa.get("permCodes") or []),
|
||
json.dumps(sa, ensure_ascii=False)[:220])
|
||
check("内置角色的菜单取代码定义（含 role 菜单）",
|
||
"role" in (sa.get("menus") or []))
|
||
check("内置角色带 userCount（删除守卫与界面都要用）",
|
||
isinstance(sa.get("userCount"), int) and sa.get("userCount") >= 1,
|
||
f"userCount={sa.get('userCount')}")
|
||
check("⭐ 可分配的内置角色 assignable=true（学校管理员能发给账号）",
|
||
sa.get("assignable") is True, f"assignable={sa.get('assignable')}")
|
||
# ⚠️ demo 租户里有一行历史遗留的 PLATFORM_ADMIN（早于"平台角色只发平台租户"那次修正）。
|
||
# 它正好是这条断言的样本：**看得见、发不出去** —— 界面据此不显示它。
|
||
pa = role_by_code(at, "PLATFORM_ADMIN")
|
||
if pa:
|
||
check("⭐ 学校租户里的平台运营角色 assignable=false（看得见但发不出去）",
|
||
pa.get("assignable") is False, f"assignable={pa.get('assignable')}")
|
||
else:
|
||
check("学校租户里没有平台运营角色（也没问题）", True)
|
||
|
||
# ══════════════ 3. 建自定义角色：服务端校验 ══════════════
|
||
print("\n[3] 建自定义角色（含服务端白名单校验）")
|
||
b = create_role(at, role_a_code, "E2E 年级审核员", 3,
|
||
["task:view", "material:manage", "progress:viewOwn"],
|
||
["dashboard", "task"])
|
||
check("自定义角色创建成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
role_a = role_by_code(at, role_a_code)
|
||
check("列表里能看到它，且 editable=true", role_a.get("editable") is True)
|
||
check("权限码原样保存", sorted(role_a.get("permCodes") or [])
|
||
== sorted(["task:view", "material:manage", "progress:viewOwn"]))
|
||
check("菜单原样保存", sorted(role_a.get("menus") or []) == sorted(["dashboard", "task"]))
|
||
check("userType 落库（账号的用户类型由它推导）", role_a.get("userType") == 3)
|
||
check("新建角色默认 dataScope=4 仅本人（最保守）", role_a.get("dataScope") == 4)
|
||
check("新建角色 userCount=0", role_a.get("userCount") == 0)
|
||
|
||
check("角色码重复 → 40001",
|
||
create_role(at, role_a_code, "重名", 3, ["task:view"], ["task"]).get("code") == 40001)
|
||
check("⭐ 占用内置角色码 → 40001（不能造一个假 SCHOOL_ADMIN）",
|
||
create_role(at, "SCHOOL_ADMIN", "假管理员", 2,
|
||
["task:view"], ["task"]).get("code") == 40001)
|
||
check("角色码格式不合法（小写）→ 40001",
|
||
create_role(at, f"e2e_bad_{suffix}", "小写码", 3,
|
||
["task:view"], ["task"]).get("code") == 40001)
|
||
check("⭐ 学校租户授平台级权限 → 40301",
|
||
create_role(at, f"E2E_ESC_{suffix.upper()}", "越权角色", 2,
|
||
["tenant:manage"], ["tenant"]).get("code") == 40301,
|
||
json.dumps(create_role(at, f"E2E_ESC2_{suffix.upper()}", "越权角色2", 2,
|
||
["tenant:manage"], ["tenant"]), ensure_ascii=False)[:200])
|
||
check("不存在的权限码 → 40001",
|
||
create_role(at, f"E2E_P_{suffix.upper()}", "坏权限", 3,
|
||
["no:such:perm"], ["task"]).get("code") == 40001)
|
||
check("不存在的菜单键 → 40001",
|
||
create_role(at, f"E2E_M_{suffix.upper()}", "坏菜单", 3,
|
||
["task:view"], ["no-such-menu"]).get("code") == 40001)
|
||
check("⭐ 学校租户配平台专用菜单（租户管理）→ 40301",
|
||
create_role(at, f"E2E_MT_{suffix.upper()}", "平台菜单", 3,
|
||
["task:view"], ["tenant"]).get("code") == 40301)
|
||
check("⭐ 学校租户建「平台运营」类角色 → 40301",
|
||
create_role(at, f"E2E_U1_{suffix.upper()}", "平台类角色", 1,
|
||
["task:view"], ["task"]).get("code") == 40301)
|
||
|
||
# ══════════════ 4. 发出去 + 权限真的在挡 ══════════════
|
||
print("\n[4] 把自定义角色发给账号，验证权限码真的在挡")
|
||
phone_a = "139" + "".join(c for c in suffix if c.isdigit()).ljust(8, "0")[:8]
|
||
b = create_user(at, user_a, "E2E角色甲", phone_a, [role_a_code], 3)
|
||
check("给账号发自定义角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
uid_a = (b.get("data") or {}).get("id")
|
||
pwd_a = (b.get("data") or {}).get("initialPassword")
|
||
check("新增账号返回初始密码", bool(pwd_a))
|
||
|
||
check("⭐ 用户类型与角色不符 → 40001（角色说 3，传 4）",
|
||
create_user(at, f"{user_a}_x", "E2E角色甲x", "13900000001",
|
||
[role_a_code], 4).get("code") == 40001)
|
||
check("⭐ 学校租户发 PLATFORM_ADMIN → 40301（原有防线未回归）",
|
||
create_user(at, f"{user_a}_p", "E2E越权", "13900000002",
|
||
["PLATFORM_ADMIN"], 1).get("code") == 40301)
|
||
|
||
ta, prof_a = login(user_a, pwd_a)
|
||
check("自定义角色账号能登录", bool(ta), json.dumps(prof_a, ensure_ascii=False)[:200])
|
||
check("⭐ 登录响应里就带着自定义角色的权限（否则要刷新一次才看得见按钮）",
|
||
"task:view" in (prof_a.get("permissions") or []),
|
||
json.dumps(prof_a.get("permissions"), ensure_ascii=False)[:220])
|
||
check("登录响应里带着自定义角色的菜单",
|
||
"task" in (prof_a.get("menus") or []) and "role" not in (prof_a.get("menus") or []),
|
||
json.dumps(prof_a.get("menus"), ensure_ascii=False)[:220])
|
||
|
||
b = body_of("GET", "/api/v1/auth/me", ta)
|
||
check("/auth/me 的 permissions 与登录响应一致",
|
||
"task:view" in (b.get("data", {}).get("permissions") or []),
|
||
json.dumps(b, ensure_ascii=False)[:220])
|
||
|
||
b = body_of("GET", "/api/v1/tasks", ta)
|
||
check("有 task:view → 任务列表可读", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160])
|
||
b = body_of("POST", "/api/v1/tasks", ta, {
|
||
"taskName": "不该建出来", "batchNo": "202610", "majorName": "数控加工",
|
||
"collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31"})
|
||
check("⭐ 没有 task:manage → 建任务被 40301 拦住",
|
||
b.get("code") == 40301, f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("GET", "/api/v1/roles", ta)
|
||
check("⭐ 没有 role:manage / user:view → 读角色列表 40301",
|
||
b.get("code") == 40301, f"code={b.get('code')}")
|
||
|
||
# ══════════════ 5. 改授权：立即生效（同一个 token）══════════════
|
||
print("\n[5] 改授权立即生效（不需要重新登录）")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_a['id']}/permissions", at,
|
||
{"permCodes": ["roster:view"]})
|
||
check("管理员改角色权限成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160])
|
||
b = body_of("GET", "/api/v1/tasks", ta)
|
||
check("⭐ 用**同一个 token** 立刻失去 task:view（40301，不必重新登录）",
|
||
b.get("code") == 40301, f"code={b.get('code')}")
|
||
b = body_of("GET", "/api/v1/auth/me", ta)
|
||
check("同一 token 的 /auth/me 权限同步变化",
|
||
"task:view" not in (b.get("data", {}).get("permissions") or []),
|
||
json.dumps(b.get("data", {}).get("permissions"), ensure_ascii=False)[:200])
|
||
|
||
body_of("PUT", f"/api/v1/roles/{role_a['id']}/permissions", at,
|
||
{"permCodes": ["task:view", "material:manage"]})
|
||
b = body_of("GET", "/api/v1/tasks", ta)
|
||
check("权限加回来 → 立刻又能读（缓存双向都生效）", b.get("code") == 0, f"code={b.get('code')}")
|
||
|
||
# ══════════════ 6. 菜单随角色走 ══════════════
|
||
print("\n[6] 菜单跟着角色配置走")
|
||
body_of("PUT", f"/api/v1/roles/{role_a['id']}/menus", at, {"menus": ["dashboard", "export"]})
|
||
menus = (body_of("GET", "/api/v1/auth/me", ta).get("data", {}) or {}).get("menus") or []
|
||
check("⭐ 菜单改成 export → /auth/me 里出现 export、task 消失",
|
||
"export" in menus and "task" not in menus, json.dumps(menus, ensure_ascii=False)[:200])
|
||
body_of("PUT", f"/api/v1/roles/{role_a['id']}/menus", at, {"menus": ["dashboard", "task"]})
|
||
menus = (body_of("GET", "/api/v1/auth/me", ta).get("data", {}) or {}).get("menus") or []
|
||
check("菜单改回来 → task 回来", "task" in menus)
|
||
|
||
# ══════════════ 7. 停用角色立即生效 ══════════════
|
||
print("\n[7] 停用角色：立刻不再给权限")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=0", at)
|
||
check("停用成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:160])
|
||
b = body_of("GET", "/api/v1/tasks", ta)
|
||
check("⭐ 同一个 token 立刻失去权限（40301）", b.get("code") == 40301, f"code={b.get('code')}")
|
||
body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=1", at)
|
||
check("重新启用 → 权限回来",
|
||
body_of("GET", "/api/v1/tasks", ta).get("code") == 0)
|
||
check("非法状态值 → 40001",
|
||
body_of("PUT", f"/api/v1/roles/{role_a['id']}/status?status=9", at).get("code") == 40001)
|
||
|
||
# ══════════════ 8. 内置角色不可改 ══════════════
|
||
print("\n[8] 内置角色：改 / 删 / 授权 / 停用 一律拒绝")
|
||
sid = sa["id"]
|
||
check("改内置角色 → 40001",
|
||
body_of("PUT", f"/api/v1/roles/{sid}", at,
|
||
{"roleName": "改个名", "userType": 2}).get("code") == 40001)
|
||
check("改内置角色授权 → 40001",
|
||
body_of("PUT", f"/api/v1/roles/{sid}/permissions", at,
|
||
{"permCodes": ["task:view"]}).get("code") == 40001)
|
||
check("改内置角色菜单 → 40001",
|
||
body_of("PUT", f"/api/v1/roles/{sid}/menus", at,
|
||
{"menus": ["task"]}).get("code") == 40001)
|
||
check("停用内置角色 → 40001",
|
||
body_of("PUT", f"/api/v1/roles/{sid}/status?status=0", at).get("code") == 40001)
|
||
check("删内置角色 → 40001",
|
||
body_of("DELETE", f"/api/v1/roles/{sid}", at).get("code") == 40001)
|
||
check("改一个不存在的角色 → 40001",
|
||
body_of("PUT", "/api/v1/roles/99999999", at,
|
||
{"roleName": "x", "userType": 3}).get("code") == 40001)
|
||
|
||
# ══════════════ 9. 账号列表里的角色名（读库）══════════════
|
||
print("\n[9] 账号列表的角色名读库（自定义角色的名字不在代码里）")
|
||
b = body_of("GET", f"/api/v1/users?keyword={user_a}", at)
|
||
row = next((u for u in (b.get("data", {}).get("list") or []) if u.get("id") == uid_a), {})
|
||
check("⭐ 自定义角色显示的是库里的角色名",
|
||
row.get("roleNames") == ["E2E 年级审核员"],
|
||
json.dumps(row.get("roleNames"), ensure_ascii=False)[:200])
|
||
|
||
# ══════════════ 10. 删除守卫 ══════════════
|
||
print("\n[10] 删除守卫：还有账号在用就删不掉")
|
||
b = body_of("DELETE", f"/api/v1/roles/{role_a['id']}", at)
|
||
check("⭐ 还有账号在用 → 40045", b.get("code") == 40045,
|
||
f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("PUT", f"/api/v1/users/{uid_a}", at, {
|
||
"realName": "E2E角色甲", "phone": phone_a,
|
||
"roleCodes": ["HEAD_TEACHER"], "userType": 3})
|
||
check("把该账号换成内置角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
b = body_of("DELETE", f"/api/v1/roles/{role_a['id']}", at)
|
||
check("解绑后可以删除", b.get("code") == 0, f"code={b.get('code')} msg={b.get('msg')}")
|
||
check("删除后列表里不再有它", not role_by_code(at, role_a_code))
|
||
ta2, _ = login(user_a, pwd_a)
|
||
perms = (body_of("GET", "/api/v1/auth/me", ta2).get("data", {}) or {}).get("permissions") or []
|
||
# ⚠️ 判据要选**只有自定义角色才有**的那条（material:manage）。
|
||
# 第一版断言的是 roster:view / audit:first —— 而 HEAD_TEACHER 本来就有这两条，
|
||
# 于是"换角色成功"被读成了"权限没清掉"（断言写错，不是功能错）。
|
||
check("⭐ 换角色后重新登录：自定义角色的权限没了、新角色的权限在",
|
||
"material:manage" not in perms and "task:view" in perms,
|
||
json.dumps(perms, ensure_ascii=False)[:220])
|
||
|
||
# ══════════════ 11. 防自锁 ══════════════
|
||
print("\n[11] 防自锁：不能把自己管理角色的最后一条路去掉")
|
||
b = create_role(at, role_b_code, "E2E 角色管理员", 2,
|
||
["role:manage", "user:manage", "user:view"],
|
||
["dashboard", "role", "user"])
|
||
check("建「角色管理员」角色成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
role_b = role_by_code(at, role_b_code)
|
||
b = create_user(at, user_b, "E2E角色乙", "13900000003", [role_b_code], 2)
|
||
check("发给账号成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
uid_b = (b.get("data") or {}).get("id")
|
||
pwd_b = (b.get("data") or {}).get("initialPassword")
|
||
tb, prof_b = login(user_b, pwd_b)
|
||
check("该账号能登录，且带着 role:manage",
|
||
"role:manage" in (prof_b.get("permissions") or []),
|
||
json.dumps(prof_b.get("permissions"), ensure_ascii=False)[:220])
|
||
|
||
b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/permissions", tb,
|
||
{"permCodes": ["user:manage", "user:view"]})
|
||
check("⭐ 去掉自己唯一的 role:manage → 40001（防自锁）", b.get("code") == 40001,
|
||
f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/menus", tb,
|
||
{"menus": ["dashboard", "user"]})
|
||
check("⭐ 去掉自己唯一的 role 菜单 → 40001（否则菜单里再也点不到本页）",
|
||
b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/status?status=0", tb)
|
||
check("⭐ 停用自己唯一的角色 → 40001", b.get("code") == 40001,
|
||
f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("PUT", f"/api/v1/users/{uid_b}", tb, {
|
||
"realName": "E2E角色乙", "phone": "13900000003",
|
||
"roleCodes": ["HEAD_TEACHER"], "userType": 3})
|
||
check("⭐ 把自己换成普通角色（会丢掉账号管理）→ 40001",
|
||
b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_b['id']}/permissions", tb,
|
||
{"permCodes": ["role:manage", "user:manage", "user:view", "task:view"]})
|
||
check("加权限（不减少）不受防自锁影响", b.get("code") == 0,
|
||
json.dumps(b, ensure_ascii=False)[:200])
|
||
b = body_of("PUT", f"/api/v1/users/{uid_b}", tb, {
|
||
"realName": "E2E角色乙", "phone": "13900000003",
|
||
"roleCodes": [role_b_code], "userType": 2})
|
||
check("改自己的其它字段（角色不变）不受影响", b.get("code") == 0,
|
||
json.dumps(b, ensure_ascii=False)[:200])
|
||
|
||
# ══════════════ 12. 内置角色授权同步（跨租户定时任务）══════════════
|
||
print("\n[12] 内置角色的权限以代码为准 + 每日同步回填")
|
||
role_id_sa = sa["id"]
|
||
R.sql(f"delete from sys_role_permission where role_id={role_id_sa} and perm_code='task:clone';")
|
||
check("夹具：库里那一行已删掉",
|
||
sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} "
|
||
f"and perm_code='task:clone' and deleted=0;").strip() == "0")
|
||
sa2 = role_by_code(at, "SCHOOL_ADMIN")
|
||
check("⭐ 接口返回的仍是代码那一套（库里缺了也不影响显示与判定）",
|
||
"task:clone" in (sa2.get("permCodes") or []))
|
||
|
||
_, ops = R.login(BASE, "platform", "Platform@123", None)
|
||
ops_token = (ops.get("data") or {}).get("token")
|
||
check("平台运营账号可登录（同步任务是运维动作）", bool(ops_token))
|
||
b = body_of("POST", "/api/v1/schedules/run", ops_token, {"job": "role"})
|
||
check("手工触发「内置角色授权同步」", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
check("⭐ 同步后库里那行回来了",
|
||
sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} "
|
||
f"and perm_code='task:clone' and deleted=0;").strip() == "1")
|
||
body_of("POST", "/api/v1/schedules/run", ops_token, {"job": "role"})
|
||
check("⭐ 再跑一次不产生重复行（幂等）",
|
||
sql(f"select count(*) from sys_role_permission where role_id={role_id_sa} "
|
||
f"and perm_code='task:clone' and deleted=0;").strip() == "1")
|
||
|
||
# ══════════════ 13. 越权：跨租户拿不到别人的角色 ══════════════
|
||
print("\n[13] 跨租户防线")
|
||
b = body_of("PUT", f"/api/v1/roles/{role_b['id']}", ops_token,
|
||
{"roleName": "平台改名", "userType": 2})
|
||
check("⭐ 平台运营不选学校就改学校的角色 → 40001（必须先选定学校）",
|
||
b.get("code") == 40001, f"code={b.get('code')} msg={b.get('msg')}")
|
||
b = body_of("GET", "/api/v1/roles", None)
|
||
check("未登录读角色列表 → 40101", b.get("code") == 40101, f"code={b.get('code')}")
|
||
|
||
cleanup()
|
||
print("\n" + "=" * 82)
|
||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||
print("=" * 82)
|
||
for n, det in FAILED:
|
||
print(f" ✗ {n}\n {det}")
|
||
sys.stdout.flush()
|
||
return 1 if FAILED else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|