209 lines
8.7 KiB
Python
209 lines
8.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""批量导入 + 需方预置 + PO 号规则 E2E 测试。
|
||
|
||
用法：先启动服务（python app.py 8099），再运行本脚本。
|
||
覆盖：物料/供应商导入的 新增 / 更新 / 跳过 / 失败 四条路径，
|
||
模板下载，PO 号 4 位规则，以及旧格式单号的兼容解析。
|
||
"""
|
||
import datetime
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
|
||
import openpyxl
|
||
import requests
|
||
|
||
PORT = sys.argv[1] if len(sys.argv) > 1 else "8099"
|
||
BASE = "http://127.0.0.1:%s" % PORT
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
OUT = os.path.join(HERE, "out")
|
||
os.makedirs(OUT, exist_ok=True)
|
||
DB = os.path.join(os.path.dirname(HERE), "po_app", "instance", "po.db")
|
||
|
||
s = requests.Session()
|
||
FAILS = []
|
||
|
||
|
||
def check(label, cond, extra=""):
|
||
print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else ""))
|
||
if not cond:
|
||
FAILS.append(label)
|
||
|
||
|
||
def login():
|
||
r = s.post(BASE + "/login",
|
||
data={"username": "admin", "password": "admin123"},
|
||
allow_redirects=False)
|
||
return r.status_code in (200, 302)
|
||
|
||
|
||
def make_material_xlsx(path):
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.append([u"物料代码", u"物料规格", u"单位", u"默认含税单价"])
|
||
ws.append(["M-9001", u"测试新增物料A", u"件", 12.5]) # 新增
|
||
ws.append(["M-9002", u"测试新增物料B", u"台", 88.0]) # 新增
|
||
ws.append(["M-1001", u"铝合金支架(已存在-应被更新)", u"件", 99.9]) # 已存在
|
||
ws.append(["", u"缺代码应失败", u"件", 1.0]) # 失败
|
||
wb.save(path)
|
||
|
||
|
||
def make_supplier_xlsx(path):
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.append([u"编号", u"名称", u"联系人", u"电话", u"付款方式", u"税号",
|
||
u"开户银行", u"银行帐号"])
|
||
ws.append(["S901", u"测试新增供应商A", u"王经理", "13700000001",
|
||
u"月结30天", "91320500TEST001", u"工商银行苏州分行", "1111 2222 3333"])
|
||
ws.append(["S001", u"苏州精锐五金制品有限公司", u"周经理(已更新)", "13800000001",
|
||
u"月结60天", "91320500MA1XXXXXXX", u"中国银行苏州工业园区支行",
|
||
"4562 1088 0012 3456"]) # 已存在
|
||
ws.append(["S902", u"", u"缺名称", "13900000009", u"", u"", u"", u""]) # 失败
|
||
wb.save(path)
|
||
|
||
|
||
def post_import(url, path, mode):
|
||
with open(path, "rb") as f:
|
||
return s.post(url,
|
||
files={"file": (os.path.basename(path), f,
|
||
"application/vnd.openxmlformats-officedocument."
|
||
"spreadsheetml.sheet")},
|
||
data={"mode": mode})
|
||
|
||
|
||
def flash_texts(html):
|
||
out = []
|
||
for m in re.finditer(r'class="flash[^"]*"[^>]*>(.*?)</div>', html, re.S):
|
||
out.append(re.sub(r"<[^>]+>", "", m.group(1)).strip())
|
||
return out
|
||
|
||
|
||
def main():
|
||
print("登录:", "OK" if login() else "FAIL")
|
||
# 预清理：让脚本可重复运行（上次残留的测试数据不影响本次断言）
|
||
try:
|
||
today0 = datetime.datetime.now().strftime("%Y%m%d")
|
||
con = sqlite3.connect(DB)
|
||
cur = con.cursor()
|
||
cur.execute("UPDATE materials SET spec=?, unit=?, price=? WHERE code=?",
|
||
(u"铝合金支架 6061-T6 200×80×10mm", u"件", 38.5, "M-1001"))
|
||
cur.execute("UPDATE suppliers SET contact=?, payment_terms=? WHERE code=?",
|
||
(u"周经理", u"月结30天", "S001"))
|
||
cur.execute("DELETE FROM materials WHERE code IN ('M-9001','M-9002')")
|
||
cur.execute("DELETE FROM suppliers WHERE code IN ('S901','S902')")
|
||
for p in ("PO", "WO"):
|
||
cur.execute("DELETE FROM order_items WHERE order_id IN "
|
||
"(SELECT id FROM orders WHERE order_no LIKE ?)",
|
||
(p + today0 + "%",))
|
||
cur.execute("DELETE FROM orders WHERE order_no LIKE ?", (p + today0 + "%",))
|
||
con.commit()
|
||
con.close()
|
||
except Exception as e:
|
||
print("预清理失败:", e)
|
||
|
||
mp = os.path.join(OUT, "materials_test.xlsx")
|
||
sp = os.path.join(OUT, "suppliers_test.xlsx")
|
||
make_material_xlsx(mp)
|
||
make_supplier_xlsx(sp)
|
||
|
||
print("\n[1] 物料导入 update 模式")
|
||
r = post_import(BASE + "/materials/import_excel", mp, "update")
|
||
t = " | ".join(flash_texts(r.text))
|
||
print(" flash:", t)
|
||
check("新增2 / 更新1 / 失败1", u"新增 2" in t and u"更新 1" in t and u"失败 1" in t)
|
||
|
||
print("\n[2] 供应商导入 update 模式")
|
||
r = post_import(BASE + "/suppliers/import_excel", sp, "update")
|
||
t = " | ".join(flash_texts(r.text))
|
||
print(" flash:", t)
|
||
check("新增1 / 更新1 / 失败1", u"新增 1" in t and u"更新 1" in t and u"失败 1" in t)
|
||
|
||
print("\n[3] 再导一次，skip 模式（应全部跳过）")
|
||
r = post_import(BASE + "/materials/import_excel", mp, "skip")
|
||
t = " | ".join(flash_texts(r.text))
|
||
print(" 物料 flash:", t)
|
||
check("物料跳过3（M-9001/M-9002/M-1001 此时都已存在）", u"跳过 3" in t)
|
||
r = post_import(BASE + "/suppliers/import_excel", sp, "skip")
|
||
t = " | ".join(flash_texts(r.text))
|
||
print(" 供应商 flash:", t)
|
||
check("供应商跳过2（S901/S001 此时都已存在）", u"跳过 2" in t)
|
||
|
||
print("\n[4] 列表核对")
|
||
html = s.get(BASE + "/materials").text
|
||
check("M-9001 已新增", "M-9001" in html)
|
||
check("M-1001 单价已更新为 99.90", "99.90" in html)
|
||
html = s.get(BASE + "/suppliers").text
|
||
check("S901 已新增", "S901" in html)
|
||
check("联系人已更新为 周经理(已更新)", u"周经理(已更新)" in html)
|
||
|
||
print("\n[5] 模板下载")
|
||
for u in ("/materials/import_template.xlsx", "/suppliers/import_template.xlsx"):
|
||
rr = s.get(BASE + u)
|
||
check(u, rr.status_code == 200 and len(rr.content) > 3000,
|
||
"%s bytes" % len(rr.content))
|
||
|
||
print("\n[6] 新建采购单：PO 号规则 + 需方预置")
|
||
created = []
|
||
r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"})
|
||
m = re.search(r"PO\d{12}", r.text)
|
||
no = m.group(0) if m else ""
|
||
created.append(no)
|
||
print(" 单号:", no)
|
||
today = datetime.datetime.now().strftime("%Y%m%d")
|
||
check("格式为 PO+年月日+4位", bool(re.match(r"^PO\d{8}\d{4}$", no)))
|
||
check("日期为今天 " + today, no.startswith("PO" + today))
|
||
for k, v in (("our_contact", u"胡乃琛"), ("our_phone", u"13913172516"),
|
||
("our_tax_no", u"91320507MA1WW8D15D"), ("tax_rate", u"13%")):
|
||
check("预置 %s = %s" % (k, v), v in r.text)
|
||
|
||
print("\n[7] 旧格式单号兼容（插入带横杠的 PO%s-0007）" % today)
|
||
legacy = "PO%s-0007" % today
|
||
try:
|
||
con = sqlite3.connect(DB)
|
||
con.execute("INSERT INTO orders (order_no, doc_type, status, order_date) "
|
||
"VALUES (?,?,?,?)",
|
||
(legacy, "purchase", "draft",
|
||
datetime.datetime.now().strftime("%Y-%m-%d")))
|
||
con.commit()
|
||
con.close()
|
||
r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"})
|
||
m2 = re.search(r"PO\d{12}", r.text)
|
||
got = m2.group(0) if m2 else ""
|
||
print(" 新单号:", got, " 期望: PO%s0008" % today)
|
||
check("旧格式未导致序号重置", got == "PO%s0008" % today)
|
||
created.extend([got, legacy])
|
||
except Exception as e:
|
||
print(" 跳过，原因:", e)
|
||
|
||
print("\n[8] 清理测试数据（还原被更新的既有记录）")
|
||
try:
|
||
con = sqlite3.connect(DB)
|
||
cur = con.cursor()
|
||
cur.execute("UPDATE materials SET spec=?, unit=?, price=? WHERE code=?",
|
||
(u"铝合金支架 6061-T6 200×80×10mm", u"件", 38.5, "M-1001"))
|
||
cur.execute("UPDATE suppliers SET contact=?, payment_terms=? WHERE code=?",
|
||
(u"周经理", u"月结30天", "S001"))
|
||
cur.execute("DELETE FROM materials WHERE code IN ('M-9001','M-9002')")
|
||
cur.execute("DELETE FROM suppliers WHERE code IN ('S901','S902')")
|
||
for no_ in [x for x in created if x]:
|
||
cur.execute("DELETE FROM order_items WHERE order_id IN "
|
||
"(SELECT id FROM orders WHERE order_no=?)", (no_,))
|
||
cur.execute("DELETE FROM orders WHERE order_no=?", (no_,))
|
||
con.commit()
|
||
cur.execute("SELECT COUNT(*) FROM materials WHERE code LIKE 'M-900%'")
|
||
n1 = cur.fetchone()[0]
|
||
cur.execute("SELECT COUNT(*) FROM suppliers WHERE code IN ('S901','S902')")
|
||
n2 = cur.fetchone()[0]
|
||
con.close()
|
||
check("测试物料已清除", n1 == 0)
|
||
check("测试供应商已清除", n2 == 0)
|
||
except Exception as e:
|
||
print(" 清理失败:", e)
|
||
|
||
print("\n" + ("全部通过" if not FAILS else "失败项: " + "; ".join(FAILS)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|