star
7d34a481b3
- Flask + SQLAlchemy 物料/入库/出库/BOM/生产订单全模块管理 - PyInstaller 打包为单文件 EXE (双击运行, 自动建库) - 物料维度: 入库总数量 / 消耗总数量 / 逐次入库流水(含时间) - BOM 详情展示组成物料的规格型号与单套用量 - 支持 Altium / 嘉立创 BOM 批量导入(文本与 Excel 双通道) - 内置端到端验证脚本(tools/_exe_util.py 等, 清理进程树防残留)
229 lines
11 KiB
Python
229 lines
11 KiB
Python
"""M1+M2+M3 冒烟测试: 登录 -> 看板 -> 入库出库 -> BOM -> 订单展开 -> 领料 -> 日志 -> 导出 -> 导入 -> 用户
|
||
|
||
用 Flask test_client; 通过 WMS_DB 指向临时库, 不污染真实数据。
|
||
"""
|
||
import sys, os, tempfile
|
||
tmp = os.path.join(tempfile.gettempdir(), "wms_smoke.db")
|
||
if os.path.exists(tmp):
|
||
os.remove(tmp)
|
||
os.environ["WMS_DB"] = tmp
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from app import app, db, stock_map, seed_admin
|
||
from models import (Material, Inbound, Outbound, Bom, BomItem, ProdOrder, OperationLog, User)
|
||
|
||
with app.app_context():
|
||
db.create_all()
|
||
seed_admin()
|
||
|
||
c = app.test_client()
|
||
|
||
# 1. 未登录跳登录
|
||
assert c.get("/dashboard").status_code == 302
|
||
print("[OK] 未登录跳登录")
|
||
|
||
# 2. 登录
|
||
r = c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True)
|
||
assert r.status_code == 200 and "库存看板" in r.text
|
||
print("[OK] 登录成功")
|
||
|
||
# 3. 物料收发核算
|
||
with app.app_context():
|
||
m = db.session.scalars(db.select(Material).limit(1)).first()
|
||
if not m: # 临时库无数据, 造一条
|
||
m = Material(part_number="T001", manufacturer="测试料", unit="个")
|
||
db.session.add(m); db.session.flush()
|
||
mid = m.id
|
||
before = stock_map().get(mid, 0)
|
||
c.post("/inbound", data={"material_id": mid, "qty": 100}, follow_redirects=True)
|
||
assert abs(stock_map().get(mid, 0) - (before + 100)) < 1e-6
|
||
c.post("/outbound", data={"material_id": mid, "qty": 30}, follow_redirects=True)
|
||
assert abs(stock_map().get(mid, 0) - (before + 70)) < 1e-6
|
||
rr = c.post("/outbound", data={"material_id": mid, "qty": 999999}, follow_redirects=True)
|
||
assert "库存不足" in rr.text
|
||
print(f"[OK] 库存核算: 期初 {before} -> +100 -> -30 -> {stock_map().get(mid,0)}; 超额拦截")
|
||
|
||
# 4. BOM 新增
|
||
r = c.post("/bom/add", data={"code": "BOMTEST", "name": "测试产品",
|
||
"pn": ["T001"], "qty": ["2"]}, follow_redirects=True)
|
||
assert r.status_code == 200
|
||
with app.app_context():
|
||
b = db.session.scalar(db.select(Bom).where(Bom.code == "BOMTEST"))
|
||
assert b and db.session.scalar(db.select(BomItem).where(BomItem.bom_id == b.id))
|
||
print("[OK] BOM 新增 + 组成项")
|
||
|
||
# 5. 生产订单 + 展开
|
||
r = c.post("/order/new", data={"bom_id": b.id, "qty": 10}, follow_redirects=True)
|
||
assert r.status_code == 200 and "领料需求" in r.text
|
||
with app.app_context():
|
||
o = db.session.scalar(db.select(ProdOrder).where(ProdOrder.bom_id == b.id))
|
||
assert o
|
||
print("[OK] 生产订单生成 + 需求展开 (单套2 × 10 = 20)")
|
||
|
||
# 6. 确认领料
|
||
c.post(f"/order/{o.id}/pick", data={}, follow_redirects=True)
|
||
with app.app_context():
|
||
assert db.session.scalar(db.select(ProdOrder).where(ProdOrder.id == o.id)).status == "picked"
|
||
pick = db.session.scalar(db.select(Outbound).where(Outbound.order_ref == o.order_no))
|
||
assert pick and abs(pick.qty - 20) < 1e-6
|
||
print("[OK] 领料出库生成 (20)")
|
||
|
||
# 7. 操作日志
|
||
r = c.get("/logs")
|
||
assert "操作日志" in r.text and "领料" in r.text
|
||
print("[OK] 操作日志已记录")
|
||
|
||
# 8. 报表 + 导出 Excel
|
||
assert "库存报表" in c.get("/report").text
|
||
r = c.get("/export/inventory")
|
||
assert r.status_code == 200 and r.mimetype in ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/octet-stream")
|
||
print("[OK] 报表页 + Excel 导出")
|
||
|
||
# 9. 模板下载
|
||
assert c.get("/template/materials").status_code == 200
|
||
assert c.get("/template/bom").status_code == 200
|
||
print("[OK] 导入模板下载")
|
||
|
||
# 10. Excel 导入物料
|
||
import openpyxl
|
||
from excel_io import read_materials, import_materials
|
||
wb = openpyxl.Workbook(); ws = wb.active
|
||
ws.append(["物料编号", "型号/规格", "单位", "安全库存", "期初库存"])
|
||
ws.append(["IMP001", "导入料A", "个", "10", "100"])
|
||
tp = os.path.join(tempfile.gettempdir(), "imp_m.xlsx"); wb.save(tp)
|
||
rows = read_materials(tp)
|
||
assert rows and rows[0]["part_number"] == "IMP001"
|
||
with app.app_context():
|
||
a, u = import_materials(rows, operator="tester")
|
||
assert a == 1
|
||
print("[OK] Excel 物料导入 (新增 IMP001)")
|
||
|
||
# 10.5 批量导入窗口(入库) 预览 + 提交
|
||
assert "批量导入窗口" in c.get("/inbound").text
|
||
preview = c.post("/api/import/transactions/preview",
|
||
json={"kind": "inbound", "rows": [
|
||
{"part_number": "IMP001", "qty": 50, "batch_no": "BA", "date": "2026-09-11", "remark": "批导"},
|
||
{"part_number": "NOPE", "qty": 5},
|
||
]})
|
||
pv = preview.get_json()
|
||
assert pv["valid"] == 1 and pv["invalid"] == 1, pv
|
||
commit = c.post("/api/import/transactions/commit", json={"kind": "inbound", "rows": pv["rows"]})
|
||
cj = commit.get_json()
|
||
assert cj["added"] == 1 and cj["skipped"] == 1, cj
|
||
with app.app_context():
|
||
rec = db.session.scalars(db.select(Inbound).where(Inbound.batch_no == "BA")).first()
|
||
assert rec and abs(rec.qty - 50) < 1e-6
|
||
# 出库预览: 物料不存在应标记异常
|
||
opv = c.post("/api/import/transactions/preview",
|
||
json={"kind": "outbound", "rows": [{"part_number": "GHOST", "qty": 1}]}).get_json()
|
||
assert opv["rows"][0]["ok"] is False
|
||
print("[OK] 批量导入窗口: 预览校验 + 提交入库 1 条 / 跳过 1 条(物料不存在)")
|
||
|
||
# 11. 用户管理 (admin)
|
||
assert "用户列表" in c.get("/users").text
|
||
c.post("/user/add", data={"username": "op1", "password": "pw", "role": "operator"}, follow_redirects=True)
|
||
with app.app_context():
|
||
assert db.session.scalar(db.select(User).where(User.username == "op1"))
|
||
print("[OK] 用户新增 (operator)")
|
||
|
||
# 12. 角色拦截: operator 不能访问 /users
|
||
c.post("/logout")
|
||
c.post("/login", data={"username": "op1", "password": "pw"}, follow_redirects=True)
|
||
assert c.get("/users").status_code == 302
|
||
assert c.get("/boms").status_code == 302
|
||
print("[OK] 角色权限拦截 (operator 无权访问用户/BOM)")
|
||
|
||
# 13. 物料批量导入窗口(粘贴预览 + 提交)
|
||
c.post("/logout")
|
||
c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True)
|
||
assert "批量导入窗口" in c.get("/materials").text
|
||
mpv = c.post("/api/import/materials/preview", json={"text": "PK001 料一 个 10 200\nPK002 料二 卷 5 50\nPK001 重复料 个 1 1"})
|
||
mp = mpv.get_json()
|
||
assert mp["valid"] == 2 and mp["invalid"] == 1, mp # 第三行本批重复编号
|
||
# 提交有效行
|
||
mc = c.post("/api/import/materials/commit", json={"rows": mp["rows"]}).get_json()
|
||
assert mc["added"] == 2 and mc["updated"] == 0, mc
|
||
with app.app_context():
|
||
assert db.session.scalar(db.select(Material).where(Material.part_number == "PK001"))
|
||
# operator 越权应被拦截
|
||
c.post("/logout")
|
||
c.post("/login", data={"username": "op1", "password": "pw"}, follow_redirects=True)
|
||
assert c.post("/api/import/materials/preview", json={"text": "X1 a 个 1 1"}).status_code == 302
|
||
c.post("/logout")
|
||
c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True)
|
||
print("[OK] 物料批量导入窗口: 预览校验 + 提交 2 条 / 重复拦截 + 角色拦截")
|
||
|
||
# 14. BOM 批量导入窗口(多 BOM 粘贴预览 + 提交 + 自动建物料)
|
||
assert "批量导入窗口" in c.get("/boms").text
|
||
bpv = c.post("/api/import/boms/preview", json={"text":
|
||
"#BOM BM1 产品一\nPK001 2\nPK002 0.5\nNOPE 3\n#BOM BM2 产品二\nPK001 4"})
|
||
bp = bpv.get_json()
|
||
assert bp["valid"] == 2 and bp["invalid"] == 0, bp
|
||
bm1 = next(b for b in bp["boms"] if b["code"] == "BM1")
|
||
# NOPE 物料不存在 -> 预览标 auto_create, 仍算可导入项(导入时自动新建)
|
||
assert bm1["valid_items"] == 3 and bm1["skip"] == 0, bm1
|
||
nope_item = next(it for it in bm1["items"] if it["part_number"] == "NOPE")
|
||
assert nope_item["auto_create"] is True, nope_item
|
||
bc = c.post("/api/import/boms/commit", json={"boms": bp["boms"]}).get_json()
|
||
# NOPE 被自动新建物料; BM2 的 PK001 已存在 -> 共新建 1 个物料, 成功项 4
|
||
assert bc["boms"] == 2 and bc["added_items"] == 4 and bc["created_materials"] == 1, bc
|
||
with app.app_context():
|
||
assert db.session.scalar(db.select(Bom).where(Bom.code == "BM1"))
|
||
assert db.session.scalar(db.select(Bom).where(Bom.code == "BM2"))
|
||
assert db.session.scalar(db.select(Material).where(Material.part_number == "NOPE"))
|
||
print("[OK] BOM 批量导入窗口: 多 BOM 预览 + 物料不存在自动新建 + 提交 2 BOM / 成功项 4 / 新建物料 1")
|
||
|
||
# 15. 真实 Altium/嘉立创 格式: 多 sheet, Part Number + Quantity 列
|
||
wb = openpyxl.Workbook()
|
||
ws1 = wb.active
|
||
ws1.title = "XM-A"
|
||
ws1.append(["Item", "Part Number", "Manufacture Part Number", "Quantity", "Reference", "Part", "Package"])
|
||
ws1.append([1, "P100", "MF100", 2, "U1", "MF100", "PKG1"])
|
||
ws1.append([2, "P101", "MF101", 1, "U2", "MF101", "PKG2"])
|
||
ws2 = wb.create_sheet("组装")
|
||
ws2.append(["序号", "层级", "物料代码", "描述", "数量", "材质"])
|
||
ws2.append([1, "0.1", "P200", "外壳", 3, "ABS"])
|
||
ws2.append([2, "0.1", "P201", "螺丝", 4, "铁"])
|
||
tp2 = os.path.join(tempfile.gettempdir(), "bom_real.xlsx")
|
||
wb.save(tp2)
|
||
with open(tp2, "rb") as f:
|
||
r = c.post("/api/import/boms/preview", data={"file": (f, "bom_real.xlsx")},
|
||
content_type="multipart/form-data")
|
||
rpv = r.get_json()
|
||
codes = [b["code"] for b in rpv["boms"]]
|
||
assert "XM-A" in codes and "组装" in codes, rpv
|
||
assert rpv["valid"] == 2, rpv
|
||
# 提交(自动建 4 个物料)
|
||
rbc = c.post("/api/import/boms/commit", json={"boms": rpv["boms"]}).get_json()
|
||
assert rbc["boms"] == 2 and rbc["created_materials"] == 4, rbc
|
||
with app.app_context():
|
||
from models import Material as _M
|
||
xm = db.session.scalar(db.select(Bom).where(Bom.code == "XM-A"))
|
||
assert xm
|
||
its = db.session.scalars(db.select(BomItem).where(BomItem.bom_id == xm.id)).all()
|
||
assert len(its) == 2
|
||
p100 = db.session.scalar(db.select(_M).where(_M.part_number == "P100"))
|
||
assert p100
|
||
it = next(i for i in its if i.material_id == p100.id)
|
||
assert abs(it.qty_per_unit - 2) < 1e-6
|
||
print("[OK] 真实格式 BOM: 多 sheet 解析 + 自动建 4 物料 + 提交 2 个 BOM")
|
||
|
||
# 15b. 用户提供的真实 BOM 模板文件直接解析验证(双 Quantity 取首列)
|
||
REAL = r"D:\办公\Q其他\20190316\2301045(码灵枪模组)\原理图\2301045T03\2301045T03\Output\2301045T03(生产)\2301045T03BOM(032A)-20260811.xlsx"
|
||
if os.path.exists(REAL):
|
||
from excel_io import read_boms_multi
|
||
real = read_boms_multi(REAL)
|
||
assert len(real) >= 2, real
|
||
assert "2301045T03" in [b["code"] for b in real], [b["code"] for b in real]
|
||
pcb = next(b for b in real if b["code"] == "2301045T03")
|
||
it_1501001 = next((it for it in pcb["items"] if it["part_number"] == "1501001"), None)
|
||
# 双 Quantity 列: 取第一个(D 列)=2, 而非 E 列=1
|
||
assert it_1501001 and it_1501001["qty_per_unit"] == 2, it_1501001
|
||
it_1001003 = next(it for it in pcb["items"] if it["part_number"] == "1001003")
|
||
assert it_1001003["manufacture"] == "CFW32C7UL_BUA", it_1001003
|
||
print(f"[OK] 真实模板文件解析: {len(real)} 个 BOM, 主 BOM {len(pcb['items'])} 项, 双Quantity取首列正确")
|
||
else:
|
||
print("[SKIP] 真实模板文件不存在, 跳过 15b")
|
||
|
||
print("\n全部冒烟测试通过 ✅")
|