SunYF/bj_power
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2544c6141dfb23527bee5c9234835cd7b2dfa348
bj_power/e2e_union_verify.py
T

211 lines
11 KiB
Python
Raw Normal View History

feat: 完成产线与仓储系统多模块迭代升级
2026-09-04 14:18:53 +08:00
# -*- coding: utf-8 -*-
"""双系统(管理界面统一/去英文/WMS操作日志)端到端验证。
覆盖：WMS 角色 options+自定义角色分配+操作日志写读+权限负向；MES 权限码拆分迁移+角色树配接口+账号CRUD+内置保护+日志时间筛选。
"""
import json
import os
import subprocess
import urllib.request
PSQL = r"C:/Program Files/PostgreSQL/17/bin/psql.exe"
WMS = "http://127.0.0.1:8890"
MES = "http://127.0.0.1:8888"
PASS = 0
FAIL = 0
def psql(db, sql):
env = dict(os.environ)
env["PGPASSWORD"] = "postgres"
r = subprocess.run([PSQL, "-U", "postgres", "-h", "127.0.0.1", "-d", db, "-t", "-c", sql],
env=env, capture_output=True)
out = r.stdout.decode("utf-8", "replace").strip()
err = r.stderr.decode("utf-8", "replace").strip()
if r.returncode != 0:
print("PSQL-ERR(%s) rc=%d | %s | %s" % (db, r.returncode, sql[:80], err))
return out
def api(base, method, path, body=None, token=None, headers=None):
data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
h = {"Content-Type": "application/json"}
if token:
h["Authorization"] = "Bearer " + token
if headers:
h.update(headers)
req = urllib.request.Request(base + path, data=data, headers=h, method=method)
try:
with urllib.request.urlopen(req, timeout=15) as r:
raw = r.read().decode("utf-8", "replace")
try:
return r.status, json.loads(raw)
except Exception:
return r.status, {"code": 0, "data": raw}
except urllib.error.HTTPError as e:
try:
return e.code, json.loads(e.read().decode("utf-8", "replace"))
except Exception:
return e.code, {}
except Exception as e:
return 0, {"code": -1, "message": str(e)}
def report(name, ok, extra=""):
global PASS, FAIL
if ok:
PASS += 1
print("PASS | " + name + (" | " + str(extra) if extra else ""))
else:
FAIL += 1
print("FAIL | " + name + (" | " + str(extra) if extra else ""))
# ========== 预清理 ==========
psql("bj_power_wms", "DELETE FROM users WHERE username IN ('union-admin'); DELETE FROM roles WHERE code='union-role';")
psql("bj_power_mes", "DELETE FROM \"user\" WHERE username='union-e2e'; DELETE FROM role WHERE code='union-e2e-role';")
# ========== WMS ==========
print("\n==== WMS ====")
st, r = api(WMS, "POST", "/api/auth/login", {"username": "admin", "password": "123456"})
tok = (r.get("data") or {}).get("token", "")
report("W1 admin登录", st == 200 and bool(tok))
st, r = api(WMS, "POST", "/api/rbac/seed", {}, tok)
report("W2 rbac/seed", st == 200 and r.get("code") == 0, r.get("message", ""))
st, r = api(WMS, "GET", "/api/roles/options", token=tok)
codes = [x.get("code") for x in (r.get("data") or {}).get("list", [])]
report("W3 角色options(user:manage可读)", st == 200 and "admin" in codes, codes[:6])
st, r = api(WMS, "GET", "/api/permissions", token=tok)
pcodes = [x.get("code") for x in (r.get("data") or {}).get("list", [])]
report("W4 权限含 eventlog:manage", "eventlog:manage" in pcodes)
# 创建自定义角色并绑定用户（修复“自定义角色无法分配”断点）
st, r = api(WMS, "POST", "/api/roles", {"name": "联调角色", "code": "union-role", "remark": "e2e",
"permissionCodes": ["dashboard:view", "inventory:view"]}, tok)
roleId = 0
rl = (r.get("data") or {})
roleId = rl.get("id") or 0
report("W5 新增自定义角色", st == 200 and roleId > 0, r.get("message", ""))
st, r = api(WMS, "POST", "/api/user/create", {"username": "union-admin", "password": "union1234",
"realName": "联调", "roleId": roleId, "dept": "测试"}, tok)
report("W6 新用户绑定自定义角色roleId", st == 200 and (r.get("data") or {}).get("roleId") == roleId,
r.get("message", ""))
# 后端应回填 role 字符串为自定义角色 code（非默认 operator）
urow = psql("bj_power_wms", "SELECT role FROM users WHERE username='union-admin'")
report("W7 role字符串与roleId一致(自定义code)", urow == "union-role", urow)
# 操作日志链路：user.create / role.create 已落库
cnt = psql("bj_power_wms", "SELECT count(*) FROM event_log WHERE event_type LIKE 'user.%' AND operator='admin'")
report("W8 日志已写(user.* by admin)", int(cnt or 0) >= 1, cnt)
st, r = api(WMS, "GET", "/api/event/logs?page=1&pageSize=5&eventType=user.&operator=admin", token=tok)
d = r.get("data") or {}
report("W9 日志查询接口(按类型/操作人)", st == 200 and r.get("code") == 0 and d.get("total", 0) >= 1,
json.dumps(d, ensure_ascii=False)[:120])
# 时间范围过滤（当天）
import time as _t
t0 = int(_t.time()) - 86400
t1 = int(_t.time()) + 60
st, r = api(WMS, "GET", "/api/event/logs?from=%d&to=%d" % (t0, t1), token=tok)
report("W10 日志时间范围过滤", st == 200 and (r.get("data") or {}).get("total", 0) >= 1)
# 负向：operator(store1) 无 eventlog:manage / user:manage
st, r = api(WMS, "POST", "/api/auth/login", {"username": "store1", "password": "123456"})
tok2 = (r.get("data") or {}).get("token", "")
st, r = api(WMS, "GET", "/api/event/logs", token=tok2)
report("W11 operator 无日志权限(403)", st == 403, st)
st, r = api(WMS, "POST", "/api/roles/options", token=tok2) # GET 不匹配 -> 直接看 GET
st, r = api(WMS, "GET", "/api/roles/options", token=tok2)
report("W12 operator 无角色options(user:manage)", st == 403, st)
# 编辑自定义角色（role.update 日志）→ 再删除（先删用户）
st, r = api(WMS, "POST", "/api/roles/update", {"id": roleId, "name": "联调角色2",
"remark": "e2e-upd", "permissionCodes": ["dashboard:view"]}, tok)
report("W13 编辑角色权限", st == 200, r.get("message", ""))
st, r = api(WMS, "POST", "/api/roles/update", {"id": roleId, "name": "X", "permissionCodes": []}, tok)
cnt2 = psql("bj_power_wms", "SELECT count(*) FROM event_log WHERE event_type='role.update' AND operator='admin'")
report("W14 role.update 已写日志", int(cnt2 or 0) >= 1, cnt2)
st, r = api(WMS, "POST", "/api/user/delete", {"id": 0}, tok) # 参数错误对照
report("W15 删除参数校验", st == 400 or st == 500 or r.get("code") != 0)
# 清理 WMS union 数据
uid = psql("bj_power_wms", "SELECT id FROM users WHERE username='union-admin'")
if uid:
api(WMS, "POST", "/api/user/delete", {"id": int(uid)}, tok)
api(WMS, "POST", "/api/roles/delete", {"id": roleId}, tok)
report("W16 清理完成", psql("bj_power_wms", "SELECT count(*) FROM roles WHERE code='union-role'") == "0")
# ========== MES ==========
print("\n==== MES ====")
st, r = api(MES, "POST", "/api/v1/login", {"username": "admin", "password": "123456"})
mtok = (r.get("data") or {}).get("accessToken", "")
report("M1 admin登录", st == 200 and bool(mtok))
st, r = api(MES, "GET", "/api/v1/userinfo", token=mtok)
d = r.get("data") or {}
menuPaths = [m.get("path") for m in d.get("menus", [])]
report("M2 userinfo菜单含 /account /role", "/account" in menuPaths and "/role" in menuPaths, menuPaths)
report("M3 无旧 /rbac 菜单", "/rbac" not in menuPaths)
report("M4 roleCode=SUPER_ADMIN", d.get("roleCode") == "SUPER_ADMIN", d.get("roleCode"))
# 角色 CRUD + 权限码拆分生效（后端按钮码校验走 sys.role:*）
st, r = api(MES, "POST", "/api/v1/roles", {"name": "联调角色E2E", "code": "union-e2e-role",
"remark": "e2e", "permissionCodes": ["produce.workorder", "produce.trace"]}, mtok)
report("M5 新增角色(写接口正常)", st == 200, r.get("message", ""))
rid = psql("bj_power_mes", "SELECT id FROM role WHERE code='union-e2e-role'")
rid = int(rid) if rid else 0
report("M6 角色已落库", rid > 0, rid)
# 编码不可改（MES 业务错误为 HTTP200 + body message，按文案断言）
st, r = api(MES, "PUT", "/api/v1/roles", {"id": rid, "name": "改名", "code": "changed-code",
"remark": "", "permissionCodes": ["produce.workorder"]}, mtok)
msg7 = (r.get("message") or "") + (r.get("msg") or "")
report("M7 角色编码不可修改被拒", "不可修改" in msg7 and "编码" in msg7, msg7)
st, r = api(MES, "PUT", "/api/v1/roles", {"id": rid, "name": "联调角色E2E改名", "code": "union-e2e-role",
"remark": "e2e-upd", "permissionCodes": ["produce.workorder"]}, mtok)
report("M8 编辑角色名称/权限", st == 200, r.get("message", ""))
# 账号 CRUD 分配新角色
st, r = api(MES, "POST", "/api/v1/users", {"username": "union-e2e", "password": "union1234",
"name": "联调E2E", "roleId": rid, "status": "ENABLED",
"canLoginWorkstation": False, "stations": []}, mtok)
report("M9 新增账号绑自定义角色", st == 200, r.get("message", ""))
uid2 = psql("bj_power_mes", "SELECT id FROM \"user\" WHERE username='union-e2e'")
report("M10 账号已落库", bool(uid2))
st, r = api(MES, "PUT", "/api/v1/users", {"id": int(uid2), "username": "union-e2e", "name": "联调E2E",
"roleId": rid, "status": "DISABLED",
"canLoginWorkstation": False, "stations": []}, mtok)
report("M11 停用账号", st == 200, r.get("message", ""))
st, r = api(MES, "PUT", "/api/v1/users", {"id": int(uid2), "username": "union-e2e", "name": "联调E2E",
"roleId": rid, "status": "ENABLED",
"canLoginWorkstation": False, "stations": []}, mtok)
report("M12 重新启用", st == 200)
st, r = api(MES, "DELETE", "/api/v1/users/%s" % uid2, token=mtok)
report("M13 删除账号", st == 200, r.get("message", ""))
# 内置角色删除保护
superId = psql("bj_power_mes", "SELECT id FROM role WHERE code='SUPER_ADMIN'")
st, r = api(MES, "DELETE", "/api/v1/roles/%s" % superId, token=mtok)
msg14 = (r.get("message") or "") + (r.get("msg") or "")
report("M14 SUPER_ADMIN不可删", "内置角色" in msg14 and "不允许删除" in msg14, msg14)
operId = psql("bj_power_mes", "SELECT id FROM role WHERE code='OPERATOR'")
st, r = api(MES, "DELETE", "/api/v1/roles/%s" % operId, token=mtok)
msg15 = (r.get("message") or "") + (r.get("msg") or "")
report("M15 OPERATOR不可删(新保护)", "内置角色" in msg15 and "不允许删除" in msg15, msg15)
# 事件日志时间范围 from/to
import time as _tt
now_ms = int(_tt.time() * 1000)
st, r = api(MES, "GET", "/api/v1/event-logs?page=1&pageSize=5&from=%d&to=%d" % (now_ms - 7 * 86400000, now_ms), token=mtok)
d = r.get("data") or {}
report("M16 日志时间范围过滤", st == 200 and r.get("code") == 0 and d.get("total", 0) >= 0,
json.dumps(d, ensure_ascii=False)[:120])
# 前缀筛选
st, r = api(MES, "GET", "/api/v1/event-logs?eventType=work.order.&pageSize=3", token=mtok)
report("M17 日志事件族前缀筛选", st == 200 and r.get("code") == 0, r.get("message", ""))
# 清理 MES union-e2e-role
psql("bj_power_mes", "DELETE FROM role WHERE code='union-e2e-role';")
print("\n================ RESULT: PASS=%d FAIL=%d ================" % (PASS, FAIL))
Reference in New Issue Copy Permalink