"""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全部冒烟测试通过 ✅")