501 lines
21 KiB
Python
501 lines
21 KiB
Python
#!/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())
|