bj_power/e2e_inspection.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("\n=== 2) 从库存取一个可检验的批次号和 SN ===")
_, stq = req("GET", "/api/stock/query?manageMode=1&page=1&pageSize=5", TOKEN)
st_rows = stq.get("data", {}).get("list", [])
_, sq = req("GET", "/api/stock/query?manageMode=2&page=1&pageSize=5", TOKEN)
sn_rows = sq.get("data", {}).get("list", [])
print("  结构件聚合数:", len(st_rows), "| 精密件聚合数:", len(sn_rows))
# 从明细取真实批次号 / SN
batch_id = None
if st_rows:
    mc = st_rows[0]["materialCode"]
    _, dd = req("GET", f"/api/stock/details?materialCode={mc}&manageMode=1&page=1&pageSize=5", TOKEN)
    dd_list = dd.get("data", {}).get("list", [])
    if dd_list:
        batch_id = dd_list[0].get("batchNo")
print("  测试批次号:", batch_id)

sn_id = None
if sn_rows:
    mc2 = sn_rows[0]["materialCode"]
    _, dd2 = req("GET", f"/api/stock/details?materialCode={mc2}&manageMode=2&page=1&pageSize=5", TOKEN)
    dd2_list = dd2.get("data", {}).get("list", [])
    if dd2_list:
        sn_id = dd2_list[0].get("snCode")
print("  测试 SN:", sn_id)

print("\n=== 3) 批量检验（含新字段）===")
if batch_id:
    s, r = req("POST", "/api/inspection/batch-flip", TOKEN, {
        "targetType": "BATCH", "targetIds": [batch_id], "status": "合格",
        "inspectQty": 5, "passQty": 5, "checkDesc": "外观+尺寸抽检", "inspector": "核验员"
    })
    print("  批次检验 -> HTTP", s, r.get("data"))
if sn_id:
    s2, r2 = req("POST", "/api/inspection/batch-flip", TOKEN, {
        "targetType": "SN", "targetIds": [sn_id], "status": "不合格",
        "inspectQty": 1, "passQty": 0, "checkDesc": "通断测试", "failReason": "导通电阻超标", "inspector": "核验员"
    })
    print("  SN检验 -> HTTP", s2, r2.get("data"))

print("\n=== 4) 检验记录查询 /inspection/query ===")
s, q = req("GET", "/api/inspection/query?page=1&pageSize=20", TOKEN)
print("HTTP", s, "| total:", q.get("data", {}).get("total"))
rows = q.get("data", {}).get("list", [])
for r in rows[:3]:
    print(f"  id={r['id']} 对象={r['targetType']} 编号={r['targetId']} 结论={r['status']} "
          f"检验数={r.get('inspectQty')} 合格数={r.get('passQty')} 说明={r.get('checkDesc')} 不合格原因={r.get('failReason')}")

# 校验新字段落库：找一个我们刚写的记录
found = any(r.get("checkDesc") for r in rows)
assert found, "检验记录未保存新字段(checkDesc)!"
print("  ✓ 新字段(检验数/合格数/说明/不合格原因)已落库")

print("\n=== 5) 筛选联动（结论=不合格）===")
s, fq = req("GET", "/api/inspection/query?status=" + urllib.parse.quote("不合格") + "&page=1&pageSize=20", TOKEN)
fq_list = fq.get("data", {}).get("list", [])
print("  结论=不合格 total:", fq.get("data", {}).get("total"), "| 全部不合格:", all(r["status"]=="不合格" for r in fq_list))

print("\n=== 6) 全量导出 /inspection/export ===")
s, ex = req("GET", "/api/inspection/export?page=1&pageSize=99999", TOKEN)
ex_list = ex.get("data", {}).get("list", [])
print("HTTP", s, "| 导出条数:", len(ex_list))
first = ex_list[0] if ex_list else {}
print("  导出首行 keys:", sorted(first.keys()))
assert "inspectQty" in first or "checkDesc" in first, "导出缺少新字段"
print("  ✓ 导出接口正常返回全部记录（含新字段）")

print("\n=== ALL CHECKS PASSED ===")