star
7d34a481b3
- Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留)
105 lines
4.0 KiB
Python
105 lines
4.0 KiB
Python
"""EXE 端到端验证: 物料 / BOM 批量导入窗口
|
|
|
|
用 urllib + cookiejar 维持会话, 对 http://127.0.0.1:8080 实测:
|
|
1) /materials 含"批量导入窗口"按钮
|
|
2) 物料预览(粘贴) -> 校验 -> 提交
|
|
3) /boms 含"批量导入窗口"按钮
|
|
4) BOM 多 BOM 预览 -> 校验 -> 提交
|
|
5) 数据库确认新物料 / 新 BOM 已落库
|
|
"""
|
|
import sys, time, json, urllib.request, urllib.parse, http.cookiejar
|
|
|
|
BASE = "http://127.0.0.1:8080"
|
|
|
|
def wait_port(timeout=30):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
try:
|
|
urllib.request.urlopen(BASE + "/login", timeout=2)
|
|
return True
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
return False
|
|
|
|
def post(cj, url, data=None, is_json=False, headers=None):
|
|
req = urllib.request.Request(url)
|
|
if is_json:
|
|
req.add_header("Content-Type", "application/json")
|
|
body = json.dumps(data).encode("utf-8")
|
|
elif data is not None:
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
body = urllib.parse.urlencode(data).encode("utf-8")
|
|
else:
|
|
body = None
|
|
if headers:
|
|
for k, v in headers.items():
|
|
req.add_header(k, v)
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
|
|
return opener.open(req, data=body, timeout=10)
|
|
|
|
def main():
|
|
cj = http.cookiejar.CookieJar()
|
|
assert wait_port(), "服务端口未就绪"
|
|
print("[OK] 服务已启动可访问")
|
|
|
|
# 登录
|
|
r = post(cj, BASE + "/login", {"username": "admin", "password": "admin123"})
|
|
assert "库存看板" in r.read().decode("utf-8", "ignore") or r.status == 200
|
|
print("[OK] 登录成功")
|
|
|
|
# 物料页含批量按钮
|
|
r = post(cj, BASE + "/materials")
|
|
assert "批量导入窗口" in r.read().decode("utf-8", "ignore")
|
|
print("[OK] 物料页含批量导入窗口按钮")
|
|
|
|
# 物料预览(粘贴 2 条新 + 1 条重复)
|
|
text = "MTEST1 批量料A 个 10 200\nMTEST2 批量料B 卷 5 50\nMTEST1 重复料 个 1 1"
|
|
r = post(cj, BASE + "/api/import/materials/preview",
|
|
{"text": text}, is_json=True)
|
|
mp = json.loads(r.read())
|
|
assert mp["valid"] == 2 and mp["invalid"] == 1, mp
|
|
print(f"[OK] 物料预览: 可导入 {mp['valid']} / 异常 {mp['invalid']}")
|
|
|
|
r = post(cj, BASE + "/api/import/materials/commit",
|
|
{"rows": mp["rows"]}, is_json=True)
|
|
mc = json.loads(r.read())
|
|
assert mc["added"] == 2, mc
|
|
print(f"[OK] 物料提交: 新增 {mc['added']} 条")
|
|
|
|
# BOM 页含批量按钮
|
|
r = post(cj, BASE + "/boms")
|
|
assert "批量导入窗口" in r.read().decode("utf-8", "ignore")
|
|
print("[OK] BOM 页含批量导入窗口按钮")
|
|
|
|
# BOM 预览(2 个 BOM, 其中一个含不存在物料)
|
|
btext = "#BOM PROMOTE1 推广品A\nMTEST1 2\nMTEST2 0.5\nGHOST 3\n#BOM PROMOTE2 推广品B\nMTEST1 4"
|
|
r = post(cj, BASE + "/api/import/boms/preview", {"text": btext}, is_json=True)
|
|
bp = json.loads(r.read())
|
|
assert bp["valid"] == 2 and bp["invalid"] == 0, bp
|
|
print(f"[OK] BOM 预览: 可导入 {bp['valid']} 个 BOM")
|
|
|
|
r = post(cj, BASE + "/api/import/boms/commit", {"boms": bp["boms"]}, is_json=True)
|
|
bc = json.loads(r.read())
|
|
assert bc["boms"] == 2 and bc["added_items"] == 3, bc
|
|
print(f"[OK] BOM 提交: {bc['boms']} 个 BOM / 成功项 {bc['added_items']}")
|
|
|
|
# 数据库确认落库
|
|
import sqlite3
|
|
dbp = r"D:\2026-09-10-task-20\material_wms\dist\app.db"
|
|
c = sqlite3.connect(dbp)
|
|
mats = c.execute("select count(*) from material where part_number in ('MTEST1','MTEST2')").fetchone()[0]
|
|
boms = c.execute("select count(*) from bom where code in ('PROMOTE1','PROMOTE2')").fetchone()[0]
|
|
c.close()
|
|
assert mats == 2, f"物料落库数={mats}"
|
|
assert boms == 2, f"BOM落库数={boms}"
|
|
print(f"[OK] 数据库确认: 新物料 {mats} 条 / 新 BOM {boms} 个 已落库")
|
|
|
|
print("\nEXE 物料/BOM 批量导入窗口 端到端验证通过 ✅")
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as e:
|
|
print("验证失败:", e)
|
|
sys.exit(1)
|