feat: 完成WMS系统权限体系与业务功能迭代
本提交完成了WMS系统的多维度优化升级： 1. 新增RBAC权限系统，支持角色/菜单/按钮三级权限管控 2. 重构库存盘点、物料管理、区域维护等模块的查询筛选与展示逻辑 3. 优化入库/出库/质检等业务流程，完善数据冗余与业务闭环 4. 移除旧装箱表，将装箱逻辑合并到出库主表 5. 新增前端权限判断工具、导出工具与端到端测试用例 6. 补充完善各类注释与数据库字段说明
This commit is contained in:
SunYF
+202
@@ -0,0 +1,202 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""验证 #2/#3/#4 三个 issue 的修复：
|
||||
#2 库存查询支持按入库单号过滤且返回 inboundNo
|
||||
#3 Excel 批量导入：事务化全成功/全失败，支持 batch/sn
|
||||
#4 备料出库/MES：orders 接口 MES 不可达时优雅降级；台账缺失提示友好
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import zipfile
|
||||
|
||||
BASE = "http://127.0.0.1:8890"
|
||||
TMP = "e2e_tmp"
|
||||
os.makedirs(TMP, exist_ok=True)
|
||||
|
||||
def req(method, path, body=None, token=None, raw=False):
|
||||
url = BASE + path
|
||||
data = None
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = "Bearer " + token
|
||||
if body is not None:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
r = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(r, timeout=10) as resp:
|
||||
return resp.status, resp.read().decode("utf-8")
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, e.read().decode("utf-8", "replace")
|
||||
|
||||
def jd(txt):
|
||||
d = json.loads(txt)
|
||||
return d.get("data") if isinstance(d, dict) else None
|
||||
|
||||
def post_multipart(path, token, fields, file_path, file_field="file"):
|
||||
boundary = "----wbtestboundary"
|
||||
parts = []
|
||||
for k, v in fields.items():
|
||||
parts.append(("--" + boundary).encode())
|
||||
parts.append(('Content-Disposition: form-data; name="%s"' % k).encode())
|
||||
parts.append(b"")
|
||||
parts.append(v.encode("utf-8"))
|
||||
parts.append(("--" + boundary).encode())
|
||||
parts.append(('Content-Disposition: form-data; name="%s"; filename="t.xlsx"' % file_field).encode())
|
||||
parts.append(b"Content-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
||||
parts.append(b"")
|
||||
with open(file_path, "rb") as f:
|
||||
parts.append(f.read())
|
||||
parts.append(("--" + boundary + "--").encode())
|
||||
parts.append(b"")
|
||||
data = b"\r\n".join(parts)
|
||||
r = urllib.request.Request(BASE + path, data=data,
|
||||
headers={"Authorization": "Bearer " + token,
|
||||
"Content-Type": "multipart/form-data; boundary=" + boundary},
|
||||
method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(r, timeout=15) as resp:
|
||||
return resp.status, resp.read().decode("utf-8")
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, e.read().decode("utf-8", "replace")
|
||||
|
||||
# ---------- 构造最小合法 xlsx（inlineStr，免依赖） ----------
|
||||
def sheet_xml(rows):
|
||||
out = ['<?xml version="1.0" encoding="UTF-8" standalone="yes"?>']
|
||||
out.append('<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><sheetData>')
|
||||
for r in rows:
|
||||
out.append("<row>")
|
||||
for c in r:
|
||||
s = str(c).replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
out.append('<c t="inlineStr"><is><t xml:space="preserve">%s</t></is></c>' % s)
|
||||
out.append("</row>")
|
||||
out.append("</sheetData></worksheet>")
|
||||
return "".join(out)
|
||||
|
||||
def make_xlsx(path, rows):
|
||||
ct = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"><Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/><Default Extension="xml" ContentType="application/xml"/><Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/><Override PartName="/xl/worksheets/sheet1.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/></Types>'
|
||||
rels = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/></Relationships>'
|
||||
wb = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets><sheet name="Sheet1" sheetId="1" r:id="rId1"/></sheets></workbook>'
|
||||
wbrels = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet1.xml"/></Relationships>'
|
||||
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as z:
|
||||
z.writestr("[Content_Types].xml", ct)
|
||||
z.writestr("_rels/.rels", rels)
|
||||
z.writestr("xl/workbook.xml", wb)
|
||||
z.writestr("xl/_rels/workbook.xml.rels", wbrels)
|
||||
z.writestr("xl/worksheets/sheet1.xml", sheet_xml(rows))
|
||||
|
||||
print("=== 0) login ===")
|
||||
st, txt = req("POST", "/api/auth/login", {"username": "admin", "password": "123456"})
|
||||
TOKEN = jd(txt).get("token")
|
||||
print(" token:", "OK" if TOKEN else "FAIL")
|
||||
|
||||
MAT_B = "TEST_BATCH_E2E"
|
||||
MAT_S = "TEST_SN_E2E"
|
||||
|
||||
def ensure_material(code, mode):
|
||||
st, txt = req("GET", "/api/material/query?keyword=" + code, token=TOKEN)
|
||||
lst = jd(txt)
|
||||
if isinstance(lst, dict):
|
||||
for m in lst.get("list", []):
|
||||
if m.get("code") == code:
|
||||
return
|
||||
req("POST", "/api/material/create", {"code": code, "name": code, "manageMode": mode, "unit": "PCS"}, token=TOKEN)
|
||||
|
||||
ensure_material(MAT_B, 1)
|
||||
ensure_material(MAT_S, 2)
|
||||
|
||||
print("\n=== #2 库存查询按入库单号过滤 + 返回 inboundNo ===")
|
||||
import datetime
|
||||
uq = datetime.datetime.now().strftime("%H%M%S%f")
|
||||
batch2 = "B-E2E-" + uq
|
||||
st, txt = req("POST", "/api/inbound/create",
|
||||
{"inboundType": "purchase", "materialCode": MAT_B, "batchNo": batch2,
|
||||
"quantity": 4, "zoneCode": "Z01", "operator": "tester"}, token=TOKEN)
|
||||
ib_no = jd(txt).get("inboundNo")
|
||||
print(" 入库单号:", ib_no, "| 批次:", batch2)
|
||||
st, txt = req("GET", "/api/stock/query?inboundNo=" + ib_no, token=TOKEN)
|
||||
d = jd(txt)
|
||||
rows = d.get("list", []) if isinstance(d, dict) else []
|
||||
hit = [r for r in rows if r.get("inboundNo") == ib_no]
|
||||
print(" 按该单号查到库存行:", len(hit), "| 首行 inboundNo=", hit[0].get("inboundNo") if hit else None, "| 首行批次=", hit[0].get("batchNo") if hit else None)
|
||||
st, txt = req("GET", "/api/stock/query?inboundNo=NOPE_NO_SUCH", token=TOKEN)
|
||||
d = jd(txt)
|
||||
empty = (d.get("list") == []) if isinstance(d, dict) else False
|
||||
print(" 按不存在单号查询为空:", empty)
|
||||
|
||||
print("\n=== #3 Excel 批量导入（事务 全成功/全失败） ===")
|
||||
# 合法 batch
|
||||
make_xlsx(TMP + "/batch_ok.xlsx", [
|
||||
["物料编码", "批次号", "数量", "生产日期", "供应商", "区域", "备注"],
|
||||
[MAT_B, "B-E2E-OK1", "5", "2026-09-02", "SUP", "Z01", ""],
|
||||
[MAT_B, "B-E2E-OK2", "3", "2026-09-02", "SUP", "Z01", ""],
|
||||
])
|
||||
st, txt = post_multipart("/api/inbound/batch-excel", TOKEN, {"mode": "batch"}, TMP + "/batch_ok.xlsx")
|
||||
d = jd(txt)
|
||||
print(" 合法 batch -> success=%s failed=%s" % (d.get("success"), d.get("failed")))
|
||||
# 非法 batch（一行物料不存在）
|
||||
make_xlsx(TMP + "/batch_bad.xlsx", [
|
||||
["物料编码", "批次号", "数量", "生产日期", "供应商", "区域", "备注"],
|
||||
[MAT_B, "B-E2E-FAIL", "5", "2026-09-02", "SUP", "Z01", ""],
|
||||
["UNKNOWN_MAT_X", "X", "1", "2026-09-02", "SUP", "Z01", ""],
|
||||
])
|
||||
st, txt = post_multipart("/api/inbound/batch-excel", TOKEN, {"mode": "batch"}, TMP + "/batch_bad.xlsx")
|
||||
d = jd(txt)
|
||||
print(" 非法 batch -> success=%s failed=%s err0=%s" % (d.get("success"), d.get("failed"), (d.get("errors") or [{}])[0].get("message")))
|
||||
st, txt = req("GET", "/api/stock/query?keyword=B-E2E-FAIL", token=TOKEN)
|
||||
d = jd(txt)
|
||||
not_written = (d.get("list") == []) if isinstance(d, dict) else False
|
||||
print(" 非法行对应批次未被写入(整体回退):", not_written)
|
||||
# 合法 sn
|
||||
make_xlsx(TMP + "/sn_ok.xlsx", [
|
||||
["物料编码", "SN", "区域", "生产日期", "供应商"],
|
||||
[MAT_S, "SNE2E001", "Z01", "", ""],
|
||||
[MAT_S, "SNE2E002", "Z01", "", ""],
|
||||
])
|
||||
st, txt = post_multipart("/api/inbound/batch-excel", TOKEN, {"mode": "sn"}, TMP + "/sn_ok.xlsx")
|
||||
d = jd(txt)
|
||||
print(" 合法 sn -> success=%s failed=%s" % (d.get("success"), d.get("failed")))
|
||||
# 非法 sn（文件内重复 SN）
|
||||
make_xlsx(TMP + "/sn_bad.xlsx", [
|
||||
["物料编码", "SN", "区域", "生产日期", "供应商"],
|
||||
[MAT_S, "SNE2E_DUP", "Z01", "", ""],
|
||||
[MAT_S, "SNE2E_DUP", "Z01", "", ""],
|
||||
])
|
||||
st, txt = post_multipart("/api/inbound/batch-excel", TOKEN, {"mode": "sn"}, TMP + "/sn_bad.xlsx")
|
||||
d = jd(txt)
|
||||
print(" 非法 sn(重复) -> success=%s failed=%s" % (d.get("success"), d.get("failed")))
|
||||
st, txt = req("GET", "/api/stock/query?keyword=SNE2E_DUP", token=TOKEN)
|
||||
d = jd(txt)
|
||||
sn_not_written = (d.get("list") == []) if isinstance(d, dict) else False
|
||||
print(" 重复 SN 未被写入(整体回退):", sn_not_written)
|
||||
|
||||
print("\n=== #4 备料出库/MES 关联 ===")
|
||||
st, txt = req("GET", "/api/orders", token=TOKEN)
|
||||
od = jd(txt)
|
||||
print(" /api/orders (MES 不可达时应优雅返回空列表, 非 502): HTTP", st, "data是列表:", isinstance(od, list))
|
||||
st, txt = req("POST", "/api/outbound/create",
|
||||
{"orderNo": "NOPE_ORDER_X", "materialCode": MAT_B, "batchNo": "B-E2E-01",
|
||||
"qty": 1, "operator": "tester"}, token=TOKEN)
|
||||
d = json.loads(txt)
|
||||
print(" 台账缺失提示含'通用出库':", "通用出库" in (d.get("message") or ""), "| msg=", d.get("message"))
|
||||
|
||||
print("\n=== 清理 E2E 测试数据 ===")
|
||||
import subprocess
|
||||
cleanup_sql = (
|
||||
"DELETE FROM inbound_details WHERE material_code IN ('TEST_BATCH_E2E','TEST_SN_E2E');"
|
||||
"DELETE FROM inbound_orders WHERE material_code IN ('TEST_BATCH_E2E','TEST_SN_E2E');"
|
||||
"DELETE FROM inventories WHERE batch_no LIKE 'B-E2E-%' OR sn_code LIKE 'SNE2E%';"
|
||||
"DELETE FROM materials WHERE code IN ('TEST_BATCH_E2E','TEST_SN_E2E');"
|
||||
)
|
||||
sql = "BEGIN;" + cleanup_sql + "COMMIT;"
|
||||
try:
|
||||
subprocess.run(["psql", "-h", "127.0.0.1", "-U", "postgres", "-d", "bj_power_wms",
|
||||
"-c", sql], check=True,
|
||||
env={**os.environ, "PGPASSWORD": "postgres"},
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
print(" 已清理测试物料/库存/入库单")
|
||||
except Exception as e:
|
||||
print(" 清理失败(可手动):", e)
|
||||
|
||||
print("\n=== DONE ===")
|
||||
Reference in New Issue
Block a user