_verify_s1_s4_extra_smoke.py 3.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  1. #!/usr/bin/env python3
  2. import json
  3. import urllib.request
  4. from gmssl import sm2
  5. BASE = "http://127.0.0.1:5005"
  6. TENANT = 797403760988229
  7. pk = "84C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312"
  8. crypt = sm2.CryptSM2(public_key=pk, private_key=None, mode=1)
  9. pwd = crypt.encrypt(b"1234567890dop").hex()
  10. def http(method, path, body=None, token=None):
  11. headers = {"Content-Type": "application/json", "Accept": "application/json"}
  12. if token:
  13. headers["Authorization"] = f"Bearer {token}"
  14. data = None if body is None else json.dumps(body).encode()
  15. req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
  16. try:
  17. with urllib.request.urlopen(req, timeout=120) as r:
  18. raw = r.read().decode()
  19. return r.status, json.loads(raw) if raw else {}
  20. except Exception as e:
  21. if hasattr(e, "read"):
  22. raw = e.read().decode("utf-8", "replace")
  23. try:
  24. return e.code, json.loads(raw)
  25. except Exception:
  26. return getattr(e, "code", 0), {"raw": raw[:500]}
  27. return 0, {"error": str(e)}
  28. st, body = http(
  29. "POST",
  30. "/api/sysAuth/login",
  31. {"account": "AIDOPDemo", "password": pwd, "tenantId": str(TENANT)},
  32. )
  33. token = (body.get("result") or {}).get("accessToken")
  34. assert token, body
  35. domain = str(TENANT)
  36. apis = [
  37. ("GET", "/api/Order/seorder/list?billNo=SO202607120001&page=1&pageSize=5", None),
  38. ("GET", "/api/Order/delivery/list?billNo=SO202607120001&page=1&pageSize=10", None),
  39. ("GET", f"/api/Production/scheduling/list?page=1&pageSize=5&domain={domain}", None),
  40. ("POST", "/api/Production/scheduling/sync-routing", {"workOrd": "M000000005", "domain": domain}),
  41. ("GET", f"/api/Production/scheduling/materials?workOrd=M000000005&domain={domain}", None),
  42. ("GET", f"/api/Production/scheduling/routings?workOrd=M000000005&domain={domain}", None),
  43. ("GET", "/api/Supply/demand-schedule/list?page=1&pageSize=5", None),
  44. ("GET", "/api/Supply/purchase-request/list?page=1&pageSize=5", None),
  45. ("GET", "/api/Supply/purchase-order/list?page=1&pageSize=5", None),
  46. ("POST", "/api/WorkOrder/dispatch/kitting-check", {"workOrd": "M000000005", "tenantId": TENANT}),
  47. ("GET", "/api/ProcurementExecution/supplier-shipment/list?page=1&pageSize=5", None),
  48. ("POST", f"/api/Production/scheduling/generate?domain={domain}", None),
  49. ]
  50. out = []
  51. for method, path, body in apis:
  52. st, b = http(method, path, body, token)
  53. res = b.get("result", b) if isinstance(b, dict) else b
  54. brief = {}
  55. if isinstance(res, dict):
  56. for k in ("total", "message", "Message", "list", "workOrd", "count", "enabled", "scheduleRowCount", "workOrderCount"):
  57. if k in res:
  58. brief[k] = len(res[k]) if k == "list" and isinstance(res[k], list) else res[k]
  59. if not brief:
  60. brief = {"keys": list(res.keys())[:8], "code": b.get("code"), "msg": b.get("message")}
  61. else:
  62. brief = {"value": str(res)[:200]}
  63. out.append(
  64. {
  65. "api": f"{method} {path.split('?')[0]}",
  66. "http": st,
  67. "ok": st == 200 and b.get("code", 200) in (200, None),
  68. "brief": brief,
  69. "err": b.get("message") or b.get("raw") or b.get("error"),
  70. }
  71. )
  72. path = r"d:\Projects\Ai-DOP\SourceCode\ZZYDOP\doc\_verify_s1_s4_extra_smoke_result.json"
  73. with open(path, "w", encoding="utf-8") as f:
  74. json.dump(out, f, ensure_ascii=False, indent=2, default=str)
  75. print(json.dumps(out, ensure_ascii=False, indent=2, default=str))