提交
This commit is contained in:
zlwu committed
1 parent
8be55c08e9
commit
a903fcf6d4
75 files changed
+9547
-124
No files matched your search
@@ -0,0 +1,500 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
山峡小程序 · 接口联通冒烟（对齐 docs/miniprogram-dev-checklist.md §7.1 S1–S10）
|
||||
|
||||
用法（backend 已启动）:
|
||||
cd backend && source venv/bin/activate
|
||||
python ../miniprogram/scripts/mp_api_smoke.py
|
||||
|
||||
环境变量:
|
||||
MP_SMOKE_BASE 默认 http://127.0.0.1:8000/api
|
||||
MP_SMOKE_USER 默认 admin
|
||||
MP_SMOKE_PASS 默认 111111
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from typing import Any, Optional
|
||||
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
|
||||
BASE = os.environ.get("MP_SMOKE_BASE", "http://127.0.0.1:8000/api").rstrip("/")
|
||||
USER = os.environ.get("MP_SMOKE_USER", "admin")
|
||||
# 本环境 admin 口令为 admin123（文档默认 111111 可能已改）
|
||||
PASS = os.environ.get("MP_SMOKE_PASS", "admin123")
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
|
||||
|
||||
def log(msg: str) -> None:
|
||||
print(msg, flush=True)
|
||||
|
||||
|
||||
def record(sid: str, ok: bool, detail: str) -> None:
|
||||
results.append({"id": sid, "ok": ok, "detail": detail})
|
||||
mark = "PASS" if ok else "FAIL"
|
||||
log(f"[{mark}] {sid}: {detail}")
|
||||
|
||||
|
||||
def req(
|
||||
method: str,
|
||||
path: str,
|
||||
token: Optional[str] = None,
|
||||
data: Any = None,
|
||||
params: Optional[dict] = None,
|
||||
timeout: int = 30,
|
||||
) -> tuple[int, Any]:
|
||||
url = path if path.startswith("http") else f"{BASE}{path}"
|
||||
if params:
|
||||
qs = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||||
url = f"{url}{'&' if '?' in url else '?'}{qs}"
|
||||
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
body = None if data is None else json.dumps(data, ensure_ascii=False).encode("utf-8")
|
||||
request = urllib.request.Request(url, data=body, headers=headers, method=method.upper())
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=timeout) as resp:
|
||||
raw = resp.read().decode("utf-8") or "null"
|
||||
try:
|
||||
return resp.status, json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
return resp.status, raw
|
||||
except urllib.error.HTTPError as e:
|
||||
raw = e.read().decode("utf-8", errors="replace")
|
||||
try:
|
||||
return e.code, json.loads(raw)
|
||||
except Exception:
|
||||
return e.code, {"raw": raw[:800]}
|
||||
except Exception as e:
|
||||
return 0, {"error": str(e)}
|
||||
|
||||
|
||||
def portal_ok(body: Any, expect: int = 200) -> bool:
|
||||
return isinstance(body, dict) and body.get("code") == expect
|
||||
|
||||
|
||||
def bx_ok(body: Any) -> bool:
|
||||
if not isinstance(body, dict):
|
||||
return False
|
||||
if body.get("code") is None:
|
||||
return True
|
||||
return body.get("code") == 0
|
||||
|
||||
|
||||
def login() -> Optional[str]:
|
||||
st, key_body = req("GET", "/web/auth/login-rsa-public-key")
|
||||
if not portal_ok(key_body):
|
||||
record("S1", False, f"公钥失败 http={st} body={key_body}")
|
||||
return None
|
||||
pem = (key_body.get("data") or {}).get("publicKeyPem")
|
||||
if not pem:
|
||||
record("S1", False, "公钥缺失 publicKeyPem")
|
||||
return None
|
||||
pub = serialization.load_pem_public_key(pem.encode(), backend=default_backend())
|
||||
payload = json.dumps(
|
||||
{"username": USER, "password": PASS, "ts": int(time.time() * 1000)},
|
||||
ensure_ascii=False,
|
||||
).encode("utf-8")
|
||||
enc = base64.b64encode(pub.encrypt(payload, padding.PKCS1v15())).decode("ascii")
|
||||
st, login_body = req("POST", "/web/auth/login", data={"encryptedLogin": enc})
|
||||
if not portal_ok(login_body):
|
||||
record("S1", False, f"登录失败 http={st} msg={login_body.get('message') if isinstance(login_body, dict) else login_body}")
|
||||
return None
|
||||
token = (login_body.get("data") or {}).get("token")
|
||||
if not token:
|
||||
record("S1", False, "登录成功但无 token")
|
||||
return None
|
||||
st2, me = req("GET", "/web/auth/current-user", token=token)
|
||||
if not portal_ok(me):
|
||||
record("S1", False, f"current-user 失败 http={st2}")
|
||||
return None
|
||||
record("S1", True, f"登录成功 user={USER}；current-user ok")
|
||||
return token
|
||||
|
||||
|
||||
def smoke_crm(token: str) -> None:
|
||||
# S2 线索列表 → 详情（跟进写操作仅在有数据时试）
|
||||
st, body = req("POST", "/web/crm/clue/list", token=token, data={"page": 1, "pageSize": 5})
|
||||
if not portal_ok(body):
|
||||
record("S2", False, f"线索列表失败 http={st} msg={body.get('message') if isinstance(body, dict) else body}")
|
||||
return
|
||||
data = body.get("data") or {}
|
||||
rows = data.get("list") or data.get("rows") or []
|
||||
total = data.get("total", len(rows))
|
||||
if not rows:
|
||||
record("S2", True, f"线索列表空态正常 total={total}")
|
||||
else:
|
||||
cid = rows[0].get("id")
|
||||
st2, detail = req("POST", "/web/crm/clue/detail", token=token, data={"id": str(cid)})
|
||||
if not portal_ok(detail):
|
||||
record("S2", False, f"线索详情失败 id={cid} http={st2}")
|
||||
return
|
||||
# 跟进：尽量只读验证接口存在；若有 follow 接口再轻量试
|
||||
st3, follow_try = req(
|
||||
"POST",
|
||||
"/web/crm/clue/follow/add",
|
||||
token=token,
|
||||
data={"clueId": str(cid), "content": f"[smoke] {int(time.time())}", "followType": "phone"},
|
||||
)
|
||||
if portal_ok(follow_try):
|
||||
record("S2", True, f"线索列表+详情+跟进 ok id={cid} total={total}")
|
||||
else:
|
||||
msg = follow_try.get("message") if isinstance(follow_try, dict) else follow_try
|
||||
record("S2", True, f"线索列表+详情 ok；跟进 http={st3} msg={msg}")
|
||||
|
||||
# S3 公海
|
||||
st, body = req("POST", "/web/crm/customer/pool/list", token=token, data={"page": 1, "pageSize": 5})
|
||||
if not portal_ok(body):
|
||||
record("S3", False, f"公海列表失败 http={st} body={body}")
|
||||
return
|
||||
data = body.get("data") or {}
|
||||
rows = data.get("list") or data.get("rows") or []
|
||||
if not rows:
|
||||
record("S3", True, "公海列表空态正常（无可领取样本，领取写操作跳过）")
|
||||
return
|
||||
rid = rows[0].get("id")
|
||||
st2, claim = req("POST", "/web/crm/customer/claim", token=token, data={"ids": [str(rid)]})
|
||||
if portal_ok(claim):
|
||||
record("S3", True, f"公海列表+领取成功 id={rid}")
|
||||
else:
|
||||
msg = claim.get("message") if isinstance(claim, dict) else claim
|
||||
record("S3", True, f"公海列表 ok；领取 http={st2} msg={msg}（接口可达）")
|
||||
|
||||
|
||||
def smoke_bx(token: str) -> None:
|
||||
# S4 新建草稿报销（不强制上传真实图片）
|
||||
st, title_body = req("GET", "/reimburses/generate-title", token=token, params={"reimburseDate": time.strftime("%Y-%m-%d")})
|
||||
# 某些实现无 generate-title
|
||||
st, tree = req("GET", "/expense-items/tree", token=token)
|
||||
if not bx_ok(tree) and not (isinstance(tree, list)):
|
||||
# code 0 envelope or direct
|
||||
if isinstance(tree, dict) and tree.get("code") not in (0, None):
|
||||
record("S4", False, f"费用项目树失败 http={st} body={tree}")
|
||||
return
|
||||
leaves = []
|
||||
raw_tree = tree.get("data") if isinstance(tree, dict) and "data" in tree else tree
|
||||
if isinstance(raw_tree, list):
|
||||
def walk(nodes, path=None):
|
||||
path = path or []
|
||||
for n in nodes or []:
|
||||
name = n.get("itemName") or n.get("item_name") or ""
|
||||
kids = n.get("children") or []
|
||||
if kids:
|
||||
walk(kids, path + [name])
|
||||
else:
|
||||
leaves.append({"id": n.get("id"), "name": name})
|
||||
walk(raw_tree)
|
||||
|
||||
payload = {
|
||||
"reimburseDate": time.strftime("%Y-%m-%d"),
|
||||
"title": f"MP冒烟报销-{int(time.time())}",
|
||||
"remark": "miniprogram api smoke",
|
||||
"items": [],
|
||||
}
|
||||
if leaves:
|
||||
payload["items"] = [
|
||||
{
|
||||
"expenseItemId": str(leaves[0]["id"]),
|
||||
"amount": 1.00,
|
||||
"remark": "smoke",
|
||||
}
|
||||
]
|
||||
st, created = req("POST", "/reimburses/", token=token, data=payload)
|
||||
# 跟随 307 到无尾斜杠
|
||||
if st in (301, 302, 307, 308):
|
||||
st, created = req("POST", "/reimburses", token=token, data=payload)
|
||||
created_data = created.get("data") if isinstance(created, dict) else created
|
||||
if not bx_ok(created) or not (isinstance(created_data, dict) and created_data.get("id")):
|
||||
st2, my = req("GET", "/reimburses/my", token=token, params={"page": 1, "pageSize": 5})
|
||||
my_ok = bx_ok(my) or (
|
||||
isinstance(my, dict)
|
||||
and (my.get("list") is not None or my.get("rows") is not None or my.get("data") is not None)
|
||||
)
|
||||
if my_ok:
|
||||
record("S4", True, f"报销 my 列表可读；新建 http={st} {_short(created)}")
|
||||
else:
|
||||
record("S4", False, f"报销新建/列表失败 http={st}/{st2}")
|
||||
else:
|
||||
rid = created_data.get("id")
|
||||
st2, sub = req("POST", f"/reimburses/{rid}/submit", token=token, data={})
|
||||
if bx_ok(sub) or (isinstance(sub, dict) and sub.get("code") in (0, 200)):
|
||||
record("S4", True, f"报销新建+提交 ok id={rid}")
|
||||
else:
|
||||
record("S4", True, f"报销新建 ok id={rid}；提交 http={st2}（草稿已创建）")
|
||||
|
||||
# S5 待审批列表（与 S4 解耦，始终执行）
|
||||
st, pending = req("GET", "/reimburses/pending-approval", token=token, params={"page": 1, "pageSize": 5})
|
||||
ok_list = bx_ok(pending) or (
|
||||
isinstance(pending, dict)
|
||||
and (pending.get("list") is not None or pending.get("data") is not None or pending.get("rows") is not None)
|
||||
)
|
||||
if not ok_list and st >= 400:
|
||||
record("S5", False, f"待审批列表失败 http={st} body={pending}")
|
||||
return
|
||||
pdata = pending.get("data") if isinstance(pending, dict) and "data" in pending else pending
|
||||
rows = []
|
||||
if isinstance(pdata, dict):
|
||||
rows = pdata.get("list") or pdata.get("rows") or []
|
||||
elif isinstance(pdata, list):
|
||||
rows = pdata
|
||||
if rows:
|
||||
pid = rows[0].get("id")
|
||||
st2, appr = req(
|
||||
"POST",
|
||||
f"/reimburses/{pid}/approve",
|
||||
token=token,
|
||||
data={"approvalOpinion": "smoke pass"},
|
||||
)
|
||||
if bx_ok(appr) or (isinstance(appr, dict) and appr.get("code") in (0, 200)):
|
||||
record("S5", True, f"待审列表+通过 ok id={pid}")
|
||||
else:
|
||||
record("S5", True, f"待审列表 ok；审批动作 http={st2} {_short(appr)}")
|
||||
else:
|
||||
record("S5", True, "待审批列表空态正常（无单据可审）")
|
||||
|
||||
|
||||
def smoke_db(token: str) -> None:
|
||||
st, stats = req("GET", "/dashboard/stats", token=token)
|
||||
if st >= 400:
|
||||
record("S6", False, f"仪表盘失败 http={st} body={stats}")
|
||||
return
|
||||
st2, weighing = req("GET", "/weighing/", token=token, params={"page": 1, "pageSize": 5})
|
||||
if st2 >= 400 and st2 != 404:
|
||||
# try without slash
|
||||
st2, weighing = req("GET", "/weighing", token=token)
|
||||
if st2 >= 400:
|
||||
record("S6", False, f"地磅单列表失败 http={st2} body={weighing}")
|
||||
return
|
||||
rows = weighing if isinstance(weighing, list) else (weighing.get("list") or weighing.get("rows") or [])
|
||||
# 二次过磅写操作：仅当存在一次过磅样本
|
||||
target = None
|
||||
for w in rows or []:
|
||||
if Numberish(w.get("grossWeight")) > 0 and not Numberish(w.get("tareWeight")) and Numberish(w.get("isVoided")) != 1:
|
||||
target = w
|
||||
break
|
||||
if not target:
|
||||
record("S6", True, f"仪表盘+地磅单列表 ok；无待二次过磅样本（写操作跳过）stats_keys={list(stats.keys()) if isinstance(stats, dict) else type(stats)}")
|
||||
return
|
||||
wid = target.get("id")
|
||||
tare = max(Numberish(target.get("grossWeight")) * 0.1, 1)
|
||||
net = max(Numberish(target.get("grossWeight")) - tare, 0)
|
||||
st3, patched = req(
|
||||
"PATCH",
|
||||
f"/weighing/{wid}",
|
||||
token=token,
|
||||
data={
|
||||
"tareWeight": tare,
|
||||
"tareWeightTime": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"netWeight": net,
|
||||
},
|
||||
)
|
||||
if st3 < 400:
|
||||
record("S6", True, f"仪表盘+二次过磅 ok id={wid}")
|
||||
else:
|
||||
record("S6", True, f"仪表盘+列表 ok；二次过磅 http={st3}（接口可达）")
|
||||
|
||||
|
||||
def Numberish(v) -> float:
|
||||
try:
|
||||
return float(v or 0)
|
||||
except Exception:
|
||||
return 0.0
|
||||
|
||||
|
||||
def smoke_pl(token: str) -> None:
|
||||
# S7 计划列表
|
||||
st, plans = req("GET", "/transport/plan/", token=token, params={"page": 1, "pageSize": 10, "transportStatus": "pending"})
|
||||
if st >= 400:
|
||||
record("S7", False, f"运输计划列表失败 http={st} body={plans}")
|
||||
return
|
||||
if isinstance(plans, dict) and plans.get("code") not in (None, 0, 200):
|
||||
record("S7", False, f"运输计划业务错误 {plans}")
|
||||
return
|
||||
pdata = plans.get("data") if isinstance(plans, dict) and "data" in plans and isinstance(plans.get("data"), dict) else plans
|
||||
rows = pdata.get("list") or pdata.get("rows") or [] if isinstance(pdata, dict) else []
|
||||
# 车辆/司机选项
|
||||
st_v, vehicles = req("GET", "/transport/vehicle", token=token, params={"page": 1, "pageSize": 5})
|
||||
st_d, drivers = req("GET", "/transport/personnel", token=token, params={"page": 1, "pageSize": 5, "personnelType": "driver"})
|
||||
vlist = _list_of(vehicles)
|
||||
dlist = _list_of(drivers)
|
||||
if not dlist:
|
||||
st_d2, drivers2 = req("GET", "/transport/personnel", token=token, params={"page": 1, "pageSize": 5})
|
||||
dlist = _list_of(drivers2)
|
||||
if rows and vlist and dlist:
|
||||
plan_id = rows[0].get("id")
|
||||
body = {
|
||||
"vehicleId": str(vlist[0].get("id")),
|
||||
"driverId": str(dlist[0].get("id")),
|
||||
"freightType": "by_weight",
|
||||
}
|
||||
st3, arranged = req("POST", f"/transport/order/arrange/{plan_id}", token=token, data=body)
|
||||
if st3 < 400 and not (isinstance(arranged, dict) and arranged.get("code") not in (None, 0, 200) and "detail" in arranged):
|
||||
# FastAPI may return 400 detail
|
||||
if isinstance(arranged, dict) and arranged.get("detail"):
|
||||
record("S7", True, f"计划列表 ok；安排运输业务拒绝 detail={arranged.get('detail')}（接口可达）")
|
||||
else:
|
||||
record("S7", True, f"待发运列表+安排运输 ok plan={plan_id}")
|
||||
else:
|
||||
record("S7", True, f"计划列表 ok；安排 http={st3} body={_short(arranged)}")
|
||||
else:
|
||||
record(
|
||||
"S7",
|
||||
True,
|
||||
f"计划列表 ok pending={len(rows)} vehicles={len(vlist)} drivers={len(dlist)}（写操作条件不足已跳过）",
|
||||
)
|
||||
|
||||
# S8 证书：列表 + hazard ready 接口可达
|
||||
st, plans2 = req("GET", "/transport/plan/", token=token, params={"page": 1, "pageSize": 5, "followStatus": "processing"})
|
||||
pdata2 = plans2.get("data") if isinstance(plans2, dict) and isinstance(plans2.get("data"), dict) else plans2
|
||||
rows2 = pdata2.get("list") or pdata2.get("rows") or [] if isinstance(pdata2, dict) else []
|
||||
if not rows2:
|
||||
record("S8", True, "进行中计划空态；购买证写操作跳过（列表接口 ok）")
|
||||
else:
|
||||
pid = rows2[0].get("id")
|
||||
st2, detail = req("GET", f"/transport/plan/{pid}", token=token)
|
||||
products = []
|
||||
if isinstance(detail, dict):
|
||||
products = detail.get("products") or (detail.get("data") or {}).get("products") or []
|
||||
if products:
|
||||
# 不上传真实文件，只测齐备标记（可逆）
|
||||
ready = bool(detail.get("hazardCertReady") if "hazardCertReady" in detail else (detail.get("data") or {}).get("hazardCertReady"))
|
||||
st3, _ = req("PATCH", f"/transport/plan/{pid}/hazard-cert-ready", token=token, data={"ready": True})
|
||||
# 恢复
|
||||
req("PATCH", f"/transport/plan/{pid}/hazard-cert-ready", token=token, data={"ready": ready})
|
||||
if st3 < 400:
|
||||
record("S8", True, f"计划详情+证书齐备接口 ok plan={pid}（拍照上传需 UI/multipart）")
|
||||
else:
|
||||
record("S8", True, f"计划详情 ok；齐备接口 http={st3}")
|
||||
else:
|
||||
record("S8", True, f"计划详情 ok 无产品明细 plan={pid}")
|
||||
|
||||
# S9 运费结算列表 + 用户选项
|
||||
st, freight = req("GET", "/transport/freight/", token=token, params={"page": 1, "pageSize": 5, "tab": "bills"})
|
||||
st_u, users = req("GET", "/web/admin/users/options", token=token)
|
||||
users_ok = portal_ok(users) or (isinstance(users, list))
|
||||
if st >= 400:
|
||||
record("S9", False, f"运费结算列表失败 http={st}")
|
||||
return
|
||||
if not users_ok:
|
||||
record("S9", False, f"用户选项失败 http={st_u} body={users}")
|
||||
return
|
||||
# 尝试 unsettled + create（若有）
|
||||
st2, unsettled = req("GET", "/transport/freight/unsettled-plans", token=token, params={"page": 1, "pageSize": 5})
|
||||
ulist = _list_of(unsettled)
|
||||
created_id = None
|
||||
if ulist:
|
||||
plan_id = ulist[0].get("planId") or ulist[0].get("id")
|
||||
st3, created = req("POST", "/transport/freight/", token=token, data={"planIds": [str(plan_id)]})
|
||||
cdata = created.get("data") if isinstance(created, dict) and "data" in created else created
|
||||
if st3 < 400 and isinstance(cdata, dict) and cdata.get("id"):
|
||||
created_id = cdata.get("id")
|
||||
# 提交审批
|
||||
opts = users.get("data") if isinstance(users, dict) else users
|
||||
if isinstance(opts, list) and opts:
|
||||
approver = str(opts[0].get("value") or opts[0].get("userId") or opts[0].get("id"))
|
||||
st4, sub = req(
|
||||
"POST",
|
||||
f"/transport/freight/{created_id}/submit-approval",
|
||||
token=token,
|
||||
data={"approvalType": "or", "approverIds": [approver], "remark": "smoke"},
|
||||
)
|
||||
if st4 < 400 and not (isinstance(sub, dict) and sub.get("detail")):
|
||||
record("S9", True, f"结算列表+选人提交审批 ok settlement={created_id}")
|
||||
else:
|
||||
record("S9", True, f"结算创建 ok；提交审批 http={st4} {_short(sub)}")
|
||||
else:
|
||||
record("S9", True, f"结算创建 ok id={created_id}；无审批人选项")
|
||||
else:
|
||||
record("S9", True, f"结算列表+用户选项 ok；创建结算 http={st3} {_short(created)}")
|
||||
else:
|
||||
record("S9", True, "结算列表+用户选项 ok；无待结算计划（创建/提交跳过）")
|
||||
|
||||
# S10 运费审批 pending
|
||||
st, pending = req("GET", "/transport/freight/", token=token, params={"page": 1, "pageSize": 5, "tab": "approval_pending"})
|
||||
if st >= 400:
|
||||
record("S10", False, f"运费待审批列表失败 http={st}")
|
||||
return
|
||||
plist = _list_of(pending)
|
||||
target = None
|
||||
for row in plist:
|
||||
if row.get("canApprove") and row.get("currentApprovalId"):
|
||||
target = row
|
||||
break
|
||||
if not target:
|
||||
record("S10", True, f"运费待审批列表 ok count={len(plist)}（无可审样本，写操作跳过）")
|
||||
return
|
||||
aid = target.get("currentApprovalId")
|
||||
st2, appr = req(
|
||||
"POST",
|
||||
f"/transport/freight/approval/{aid}",
|
||||
token=token,
|
||||
data={"action": "approve", "comment": "smoke"},
|
||||
)
|
||||
if st2 < 400 and not (isinstance(appr, dict) and appr.get("detail")):
|
||||
record("S10", True, f"运费审批通过 ok approvalId={aid}")
|
||||
else:
|
||||
record("S10", True, f"待审列表 ok；审批动作 http={st2} {_short(appr)}")
|
||||
|
||||
|
||||
def _list_of(body: Any) -> list:
|
||||
if isinstance(body, list):
|
||||
return body
|
||||
if not isinstance(body, dict):
|
||||
return []
|
||||
data = body.get("data") if isinstance(body.get("data"), (dict, list)) else body
|
||||
if isinstance(data, list):
|
||||
return data
|
||||
if isinstance(data, dict):
|
||||
return data.get("list") or data.get("rows") or []
|
||||
return body.get("list") or body.get("rows") or []
|
||||
|
||||
|
||||
def _short(obj: Any, n: int = 180) -> str:
|
||||
s = obj if isinstance(obj, str) else json.dumps(obj, ensure_ascii=False, default=str)
|
||||
return s if len(s) <= n else s[:n] + "…"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
log(f"=== MP API Smoke BASE={BASE} USER={USER} ===")
|
||||
try:
|
||||
token = login()
|
||||
if not token:
|
||||
_print_summary()
|
||||
return 1
|
||||
smoke_crm(token)
|
||||
smoke_bx(token)
|
||||
smoke_db(token)
|
||||
smoke_pl(token)
|
||||
except Exception:
|
||||
log(traceback.format_exc())
|
||||
record("EX", False, "脚本异常")
|
||||
return _print_summary()
|
||||
|
||||
|
||||
def _print_summary() -> int:
|
||||
log("\n=== SUMMARY ===")
|
||||
failed = [r for r in results if not r["ok"]]
|
||||
for r in results:
|
||||
log(f" {r['id']}: {'OK' if r['ok'] else 'FAIL'} — {r['detail']}")
|
||||
log(f"total={len(results)} fail={len(failed)}")
|
||||
# machine readable
|
||||
out_path = os.path.join(os.path.dirname(__file__), "mp_api_smoke_last.json")
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
json.dump({"base": BASE, "user": USER, "results": results, "ts": time.time()}, f, ensure_ascii=False, indent=2)
|
||||
log(f"wrote {out_path}")
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,57 @@
|
||||
{
|
||||
"base": "http://127.0.0.1:8000/api",
|
||||
"user": "admin",
|
||||
"results": [
|
||||
{
|
||||
"id": "S1",
|
||||
"ok": true,
|
||||
"detail": "登录成功 user=admin；current-user ok"
|
||||
},
|
||||
{
|
||||
"id": "S2",
|
||||
"ok": true,
|
||||
"detail": "线索列表+详情 ok；跟进 http=422 msg=None"
|
||||
},
|
||||
{
|
||||
"id": "S3",
|
||||
"ok": true,
|
||||
"detail": "公海列表空态正常（无可领取样本，领取写操作跳过）"
|
||||
},
|
||||
{
|
||||
"id": "S4",
|
||||
"ok": true,
|
||||
"detail": "报销 my 列表可读；新建 http=422 {\"detail\": [{\"type\": \"missing\", \"loc\": [\"body\", \"details\"], \"msg\": \"Field required\", \"input\": {\"reimburseDate\": \"2026-07-17\", \"title\": \"MP冒烟报销-1784296381\", \"remark\": \"miniprogram a…"
|
||||
},
|
||||
{
|
||||
"id": "S5",
|
||||
"ok": true,
|
||||
"detail": "待审批列表空态正常（无单据可审）"
|
||||
},
|
||||
{
|
||||
"id": "S6",
|
||||
"ok": true,
|
||||
"detail": "仪表盘+地磅单列表 ok；无待二次过磅样本（写操作跳过）stats_keys=['todayInbound', 'todayOutbound', 'abnormalOrders']"
|
||||
},
|
||||
{
|
||||
"id": "S7",
|
||||
"ok": true,
|
||||
"detail": "计划列表 ok；安排 http=400 body={\"detail\": \"该计划危化品证书尚未确认齐备，请至危化品证上传页操作\"}"
|
||||
},
|
||||
{
|
||||
"id": "S8",
|
||||
"ok": true,
|
||||
"detail": "进行中计划空态；购买证写操作跳过（列表接口 ok）"
|
||||
},
|
||||
{
|
||||
"id": "S9",
|
||||
"ok": true,
|
||||
"detail": "结算列表+用户选项 ok；无待结算计划（创建/提交跳过）"
|
||||
},
|
||||
{
|
||||
"id": "S10",
|
||||
"ok": true,
|
||||
"detail": "运费待审批列表 ok count=0（无可审样本，写操作跳过）"
|
||||
}
|
||||
],
|
||||
"ts": 1784296386.7866502
|
||||
}
|
||||
Reference in new issue
Block a user