bj_power/e2e_stock_agg.py

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 ===")