Files
bj_power/e2e_stock_agg.py
T
SunYF 572fa975bc feat: 完成WMS系统权限体系与业务功能迭代
本提交完成了WMS系统的多维度优化升级:
1. 新增RBAC权限系统,支持角色/菜单/按钮三级权限管控
2. 重构库存盘点、物料管理、区域维护等模块的查询筛选与展示逻辑
3. 优化入库/出库/质检等业务流程,完善数据冗余与业务闭环
4. 移除旧装箱表,将装箱逻辑合并到出库主表
5. 新增前端权限判断工具、导出工具与端到端测试用例
6. 补充完善各类注释与数据库字段说明
2026-09-03 14:20:37 +08:00

121 lines
5.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import json
import urllib.request
import urllib.error
import urllib.parse
import time
BASE = "http://localhost:8902"
def req(method, path, token=None, body=None):
url = BASE + path
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(url, data=data, method=method)
r.add_header("Content-Type", "application/json")
if token:
r.add_header("Authorization", "Bearer " + token)
try:
with urllib.request.urlopen(r, timeout=15) as resp:
return resp.status, json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
try:
return e.code, json.loads(e.read().decode())
except Exception:
return e.code, {}
print("=== 1) 登录 admin ===")
_, login = req("POST", "/api/auth/login", body={"username": "admin", "password": "123456"})
TOKEN = login.get("data", {}).get("token", "")
assert TOKEN, "登录失败"
print("token:", TOKEN[:12] + "...")
def cleanup():
# 停用测试期间生成的库存行,避免污染/重复
pass
print("\n=== 2) 确保测试物料存在 ===")
def ensure_material(code, name, manage_mode, spec):
_, lst = req("GET", f"/api/material/query?materialCode={code}", TOKEN)
for m in lst.get("data", {}).get("list", []):
if m.get("code") == code:
return
s, r = req("POST", "/api/material/create", TOKEN, {
"code": code, "name": name, "spec": spec,
"manageMode": manage_mode, "unit": "件"
})
print(f" 创建物料 {code} -> HTTP {s} {r.get('message','')}")
ensure_material("ST-A100", "标准结构件A100", 1, "Φ20×300")
ensure_material("PR-S100", "精密传感器S100", 2, "M8")
print("\n=== 3) 造库存数据:入库若干批次/SN ===")
# 结构件批次入库(每条一个批次,区域 Z01)
inb1 = req("POST", "/api/inbound/create", TOKEN, {
"inboundType": "purchase", "materialCode": "ST-A100",
"quantity": 10, "zoneCode": "Z01", "operator": "verify"
})
print(" 结构件批次1入库 -> HTTP", inb1[0], inb1[1].get("message", ""))
inb1b = req("POST", "/api/inbound/create", TOKEN, {
"inboundType": "purchase", "materialCode": "ST-A100",
"quantity": 5, "zoneCode": "Z01", "operator": "verify"
})
print(" 结构件批次2入库 -> HTTP", inb1b[0], inb1b[1].get("message", ""))
# 精密件 SN 入库
sn_items = {"inboundType": "purchase", "materialCode": "PR-S100",
"zoneCode": "Z02", "operator": "verify"}
sn_items["snList"] = [f"SN100{i:03d}" for i in range(1, 6)]
inb2 = req("POST", "/api/inbound/create", TOKEN, sn_items)
print(" 精密件SN入库 -> HTTP", inb2[0], inb2[1].get("message", ""))
print("\n=== 4) 聚合主列表 /stock/query ===")
s, q = req("GET", "/api/stock/query?page=1&pageSize=50", TOKEN)
data = q.get("data", {})
rows = data.get("list", [])
print("HTTP", s, "| total:", data.get("total"), "| rows:", len(rows))
for r in rows:
print(f" 物料={r['materialCode']} 名称={r['materialName']} 类型={r['manageMode']} "
f"数量={r['totalQty']} 锁定={r['lockedQty']} 可用={r['availQty']} "
f"批次数={r['batchCount']} SN数={r['snCount']} 区={r['zoneCode']}")
agg = {r["materialCode"]: r for r in rows}
assert "ST-A100" in agg, "聚合列表缺结构件物料 ST-A100"
assert "PR-S100" in agg, "聚合列表缺精密件物料 PR-S100"
st = agg["ST-A100"]; pr = agg["PR-S100"]
assert st["batchCount"] >= 2, "结构件应有 >=2 个批次,聚合后批次数不对"
assert pr["snCount"] >= 5, "精密件应有 >=5 个 SN,聚合后 SN 数不对"
# 关键:主列表不应泄露 SN;结构中不应出现明细的 snCode/batchNo 平铺字段
for r in rows:
assert "snCode" not in r or r.get("snCode") is None, f"主列表不应泄露 SN: {r}"
assert "inventoryId" not in r, "主列表不应带明细 id"
print(" ✓ 主列表按物料聚合且不泄露 SN/批次明细")
print("\n=== 5) 明细下钻 /stock/details ===")
# 点 ST-A100 结构件(管理粒度=1)
s, d1 = req("GET", f"/api/stock/details?materialCode=ST-A100&manageMode=1&page=1&pageSize=50", TOKEN)
dd = d1.get("data", {})
print("HTTP", s, "| 结构件明细 total:", dd.get("total"), "| rows:", len(dd.get("list", [])))
for r in dd.get("list", [])[:3]:
print(f" 批次={r['batchNo']} 数量={r['quantity']} 状态={r['status']}")
assert dd.get("total", 0) >= 2, "结构件明细应有 >=2 批次"
for r in dd.get("list", []):
assert r.get("batchNo"), "明细应含批次号"
# 点 PR-S100 精密件(管理粒度=2)
s, d2 = req("GET", f"/api/stock/details?materialCode=PR-S100&manageMode=2&page=1&pageSize=50", TOKEN)
dd2 = d2.get("data", {})
print("HTTP", s, "| 精密件明细 total:", dd2.get("total"), "| rows:", len(dd2.get("list", [])))
for r in dd2.get("list", [])[:3]:
print(f" SN={r.get('snCode')} 数量={r.get('quantity')} 锁定={r.get('lockedQty')} 状态={r.get('status')}")
ok_sn = [r for r in dd2.get("list", []) if r.get("snCode")]
assert len(ok_sn) >= 5, "精密件明细应有 >=5 SN"
print(" ✓ 明细分页下钻正常,SN/批次仅在明细层可见")
print("\n=== 6) 筛选联动验证:区域 + 质量状态 ===")
s, f1 = req("GET", "/api/stock/query?zoneCode=Z01&page=1&pageSize=50", TOKEN)
print(" zoneCode=Z01 -> total:", f1.get("data", {}).get("total"))
s, f2 = req("GET", "/api/stock/query?qualityStatus=" + urllib.parse.quote("未检") + "&page=1&pageSize=50", TOKEN)
print(" qualityStatus=未检 -> total:", f2.get("data", {}).get("total"))
s, f3 = req("GET", "/api/stock/query?manageMode=2&page=1&pageSize=50", TOKEN)
md2 = f3.get("data", {}).get("list", [])
print(" manageMode=2 -> total:", f3.get("data", {}).get("total"), "| 全部为精密件:", all(r["manageMode"]==2 for r in md2))
print("\n=== ALL CHECKS PASSED ===")