star
7d34a481b3
- Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留)
90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
"""验证打包后的 EXE 包含批量导入窗口功能并端到端可用。"""
|
|
import sqlite3, time, subprocess, sys, urllib.request, urllib.parse, json, os
|
|
import atexit
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from _exe_util import start_exe, stop_exe
|
|
|
|
EXE = r"D:\2026-09-10-task-20\material_wms\dist\MaterialWMS.exe"
|
|
BASE = "http://127.0.0.1:8080"
|
|
|
|
# 取一条真实物料编号
|
|
conn = sqlite3.connect(r"D:\2026-09-10-task-20\material_wms\dist\app.db")
|
|
pn = conn.execute("select part_number from material limit 1").fetchone()[0]
|
|
conn.close()
|
|
print("使用物料编号:", pn)
|
|
|
|
print("启动 EXE ...")
|
|
p = start_exe(EXE)
|
|
atexit.register(stop_exe, p)
|
|
for _ in range(40):
|
|
try:
|
|
urllib.request.urlopen(BASE + "/login", timeout=2)
|
|
break
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
else:
|
|
print("ERROR: 服务未启动"); sys.exit(1)
|
|
print("服务已就绪")
|
|
|
|
def post(path, data=None, raw=None, headers=None):
|
|
if raw is not None:
|
|
req = urllib.request.Request(BASE + path, data=raw, headers=headers or {}, method="POST")
|
|
else:
|
|
req = urllib.request.Request(BASE + path, data=urllib.parse.urlencode(data).encode(), method="POST")
|
|
try:
|
|
return urllib.request.urlopen(req, timeout=5)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
|
|
# 登录(保持 cookie)
|
|
jar = {}
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor())
|
|
def get(path):
|
|
return opener.open(BASE + path, timeout=5)
|
|
def postj(path, obj):
|
|
req = urllib.request.Request(BASE + path, data=json.dumps(obj).encode(),
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
return opener.open(req, timeout=5)
|
|
|
|
# 登录
|
|
login = urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode()
|
|
opener.open(urllib.request.Request(BASE + "/login", data=login, method="POST"))
|
|
|
|
# 1. 入库页含批量导入窗口按钮
|
|
html = get("/inbound").read().decode("utf-8", "ignore")
|
|
assert "批量导入窗口" in html, "入库页缺少批量导入按钮"
|
|
print("[OK] 入库页含「批量导入窗口」按钮")
|
|
|
|
# 2. 下载批量模板
|
|
t = get("/template/transactions")
|
|
assert t.status == 200, "模板下载失败"
|
|
print("[OK] 批量导入模板下载 (HTTP %d)" % t.status)
|
|
|
|
# 3. 预览: 1 条有效(真实物料) + 1 条异常(物料不存在)
|
|
prev = json.loads(postj("/api/import/transactions/preview", {
|
|
"kind": "inbound",
|
|
"rows": [
|
|
{"part_number": pn, "qty": 25, "batch_no": "EXEBATCH", "date": "2026-09-11", "remark": "exe验证"},
|
|
{"part_number": "NO_SUCH", "qty": 1},
|
|
],
|
|
}).read())
|
|
assert prev["valid"] == 1 and prev["invalid"] == 1, prev
|
|
print("[OK] 预览接口: 有效 %d / 异常 %d" % (prev["valid"], prev["invalid"]))
|
|
|
|
# 4. 提交: 应写入 1 条入库(另一条跳过)
|
|
res = json.loads(postj("/api/import/transactions/commit", {
|
|
"kind": "inbound", "rows": prev["rows"]}).read())
|
|
assert res["added"] == 1 and res["skipped"] == 1, res
|
|
print("[OK] 提交接口: 入库 %d 条 / 跳过 %d 条" % (res["added"], res["skipped"]))
|
|
|
|
# 5. 确认库里确有这条 EXEBATCH 入库
|
|
conn = sqlite3.connect(r"D:\2026-09-10-task-20\material_wms\dist\app.db")
|
|
cnt = conn.execute("select count(*) from inbound where batch_no='EXEBATCH'").fetchone()[0]
|
|
conn.close()
|
|
assert cnt == 1, "入库流水未写入"
|
|
print("[OK] 数据库确认: EXEBATCH 入库流水已生成")
|
|
|
|
stop_exe(p)
|
|
print("\nEXE 批量导入窗口 端到端验证通过 ✅")
|