star
7d34a481b3
- Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留)
123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""验证新增功能: 物料明细(入库/消耗总量 + 入库流水时间) 与 BOM 明细(规格型号)。
|
|
|
|
正确收尾: 用 _exe_util.stop_exe 杀整棵进程树, 避免残留进程霸占 8080。
|
|
"""
|
|
import http.client
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from _exe_util import EXE, start_exe, stop_exe
|
|
|
|
HOST, PORT = "127.0.0.1", 8080
|
|
DB = r"D:\2026-09-10-task-20\material_wms\dist\app.db"
|
|
SEED = r"D:\2026-09-10-task-20\material_wms\instance\app.db"
|
|
|
|
|
|
def seed_test_bom():
|
|
"""往 dist 库插入一个测试 BOM(幂等), 引用已存在的物料 1/2/3。"""
|
|
c = sqlite3.connect(DB)
|
|
c.execute("DELETE FROM bom_item WHERE bom_id IN (SELECT id FROM bom WHERE code='TESTBOM01')")
|
|
c.execute("DELETE FROM bom WHERE code='TESTBOM01'")
|
|
c.execute("INSERT INTO bom(code,name,note) VALUES('TESTBOM01','测试产品','verify')")
|
|
bid = c.execute("SELECT id FROM bom WHERE code='TESTBOM01'").fetchone()[0]
|
|
for mid, q in [(1, 2.0), (2, 3.0), (3, 1.0)]:
|
|
c.execute("INSERT INTO bom_item(bom_id,material_id,qty_per_unit) VALUES(?,?,?)", (bid, mid, q))
|
|
c.commit()
|
|
c.close()
|
|
return bid
|
|
|
|
|
|
def req(c, method, path, cookie="", body=None, headers=None):
|
|
h = {"Cookie": cookie}
|
|
if headers:
|
|
h.update(headers)
|
|
if body is not None and "Content-Type" not in h:
|
|
h["Content-Type"] = "application/x-www-form-urlencoded"
|
|
c.request(method, path, body=body, headers=h)
|
|
r = c.getresponse()
|
|
data = r.read().decode("utf-8", "ignore")
|
|
return r.status, data, r.getheader("Set-Cookie") or ""
|
|
|
|
|
|
def wait_port():
|
|
for _ in range(60):
|
|
try:
|
|
http.client.HTTPConnection(HOST, PORT, timeout=2).connect()
|
|
return True
|
|
except Exception:
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def main():
|
|
if os.path.exists(SEED):
|
|
shutil.copyfile(SEED, DB)
|
|
bid = seed_test_bom()
|
|
|
|
p = start_exe()
|
|
try:
|
|
assert wait_port(), "端口未就绪"
|
|
c = http.client.HTTPConnection(HOST, PORT, timeout=15)
|
|
_, _, ck = req(c, "GET", "/login")
|
|
_, _, ck2 = req(c, "POST", "/login",
|
|
body=b"username=admin&password=admin123",
|
|
cookie=ck.split(";")[0])
|
|
cookie = (ck2.split(";")[0]) if ck2 else ck.split(";")[0]
|
|
|
|
ok = True
|
|
|
|
# 1) 物料列表含 入库总数量 / 消耗总数量 列
|
|
s, html, _ = req(c, "GET", "/materials", cookie=cookie)
|
|
print(f"[materials] status={s}")
|
|
assert s == 200, f"materials status={s}"
|
|
for kw in ["入库总数量", "消耗总数量"]:
|
|
has = kw in html
|
|
print(f" [{'OK' if has else 'FAIL'}] 列表含列: {kw}")
|
|
ok = ok and has
|
|
|
|
# 2) 物料明细 material/1: 入库=100, 消耗=30, 含入库流水日期
|
|
s, html, _ = req(c, "GET", "/material/1", cookie=cookie)
|
|
print(f"[material/1] status={s} len={len(html)}")
|
|
assert s == 200, f"material_detail status={s}"
|
|
for kw in ["入库总数量", "消耗总数量", "入库流水", "当前在库"]:
|
|
has = kw in html
|
|
print(f" [{'OK' if has else 'FAIL'}] 明细含: {kw}")
|
|
ok = ok and has
|
|
has_date = "2026-09-11" in html
|
|
print(f" [{'OK' if has_date else 'FAIL'}] 入库流水含入库日期 2026-09-11")
|
|
ok = ok and has_date
|
|
has_in = "100.00" in html
|
|
print(f" [{'OK' if has_in else 'FAIL'}] 入库总数量显示 100.00")
|
|
ok = ok and has_in
|
|
has_out = "30.00" in html
|
|
print(f" [{'OK' if has_out else 'FAIL'}] 消耗总数量显示 30.00")
|
|
ok = ok and has_out
|
|
|
|
# 3) BOM 明细含规格型号
|
|
s, html, _ = req(c, "GET", f"/bom/{bid}", cookie=cookie)
|
|
print(f"[bom/{bid}] status={s} len={len(html)}")
|
|
assert s == 200, f"bom_detail status={s}"
|
|
for kw in ["型号 / 规格", "单套用量", "组成物料"]:
|
|
has = kw in html
|
|
print(f" [{'OK' if has else 'FAIL'}] BOM明细含: {kw}")
|
|
ok = ok and has
|
|
for spec in ["X1500", "FH8626V100", "CFW32C7UL_BUA"]:
|
|
has = spec in html
|
|
print(f" [{'OK' if has else 'FAIL'}] BOM明细含物料规格型号 {spec}")
|
|
ok = ok and has
|
|
|
|
c.close()
|
|
print("\n" + ("✅ 新功能端到端验证通过" if ok else "❌ 存在未通过项"))
|
|
return 0 if ok else 1
|
|
finally:
|
|
stop_exe(p)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|