SunYF
572fa975bc
本提交完成了WMS系统的多维度优化升级： 1. 新增RBAC权限系统，支持角色/菜单/按钮三级权限管控 2. 重构库存盘点、物料管理、区域维护等模块的查询筛选与展示逻辑 3. 优化入库/出库/质检等业务流程，完善数据冗余与业务闭环 4. 移除旧装箱表，将装箱逻辑合并到出库主表 5. 新增前端权限判断工具、导出工具与端到端测试用例 6. 补充完善各类注释与数据库字段说明
80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
BASE = "http://localhost:8910"
|
|
MAT = "SYF_SN_DETAIL_1"
|
|
INBOUND_NO = None
|
|
|
|
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=10) as resp:
|
|
return resp.status, json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read().decode())
|
|
|
|
print("=== 登录 ===")
|
|
_, login = req("POST", "/api/auth/login", body={"username": "admin", "password": "123456"})
|
|
TOKEN = login.get("data", {}).get("token", "")
|
|
assert TOKEN, "登录失败"
|
|
print("token 长度:", len(TOKEN))
|
|
|
|
print("\n=== 1) 建精密件物料(manageMode=2) ===")
|
|
_, cmat = req("POST", "/api/material/create", TOKEN, {
|
|
"code": MAT, "name": "明细分页测试料", "manageMode": 2, "unit": "pcs"
|
|
})
|
|
print("create material:", cmat.get("code"), cmat.get("message"))
|
|
|
|
print("\n=== 2) 造一笔 120 条 SN 的精密件入库 ===")
|
|
sn_list = ["SN%05d" % i for i in range(1, 121)]
|
|
_, cin = req("POST", "/api/inbound/create", TOKEN, {
|
|
"inboundType": "purchase", "materialCode": MAT,
|
|
"snList": sn_list, "zoneCode": "AREA_TEST", "operator": "e2e"
|
|
})
|
|
INBOUND_NO = cin.get("data", {}).get("inboundNo")
|
|
print("inboundNo:", INBOUND_NO, "qty:", cin.get("data", {}).get("quantity"))
|
|
|
|
print("\n=== 3) 列表查询：应只带 detailCount，不带 details 全量 ===")
|
|
_, q = req("GET", "/api/inbound/query?materialCode=%s&pageSize=10" % MAT, TOKEN)
|
|
row = next((r for r in q["data"]["list"] if r.get("inboundNo") == INBOUND_NO), None)
|
|
assert row, "列表未找到该单"
|
|
print("detailCount =", row.get("detailCount"), "| 'details' 字段是否存在:", "details" in row)
|
|
assert row.get("detailCount") == 120, "detailCount 应为 120"
|
|
assert "details" not in row, "列表不应内嵌全量明细(数据量炸弹!)"
|
|
print("✅ 列表未加载全量明细，仅带总数 120")
|
|
|
|
print("\n=== 4) 明细分页接口 /inbound/details ===")
|
|
for pg in (1, 2, 3):
|
|
_, d = req("GET", "/api/inbound/details?inboundNo=%s&page=%d&pageSize=50" % (INBOUND_NO, pg), TOKEN)
|
|
dd = d.get("data", {})
|
|
print(" page %d -> total=%s 返回行数=%d" % (pg, dd.get("total"), len(dd.get("list", []))))
|
|
assert dd.get("total") == 120
|
|
if pg == 1:
|
|
assert len(dd.get("list", [])) == 50
|
|
elif pg == 2:
|
|
assert len(dd.get("list", [])) == 50
|
|
elif pg == 3:
|
|
assert len(dd.get("list", [])) == 20
|
|
print("✅ 明细分页正确: 50 / 50 / 20，服务端未一次性加载 120 条")
|
|
|
|
print("\n=== 5) 清理测试数据 ===")
|
|
import subprocess
|
|
sql = (
|
|
"DELETE FROM inbound_details WHERE inbound_no='%s';"
|
|
"DELETE FROM inbound_orders WHERE inbound_no='%s';"
|
|
"DELETE FROM inventory WHERE inbound_no='%s';"
|
|
"DELETE FROM materials WHERE code='%s';"
|
|
) % (INBOUND_NO, INBOUND_NO, INBOUND_NO, MAT)
|
|
subprocess.run(["psql", "-h", "127.0.0.1", "-U", "postgres", "-d", "bj_power_wms",
|
|
"-c", sql], env={"PGPASSWORD": "postgres", **__import__("os").environ},
|
|
check=False, capture_output=True)
|
|
print("已清理测试物料/入库单/库存")
|
|
|
|
print("\nALL PASS ✅")
|