chore(scripts): 提交分解弹窗E2E验证辅助脚本（cdp_driver + review）

2026-08-31 17:05:21 +08:00
parent c45dc554ed
commit 2241a89fd1
2 changed files with 172 additions and 0 deletions
@@ -0,0 +1,93 @@
+"""CDP e2e 续：预算管理页 → 年度分解弹窗实测"""
+import json
+import sys
+import time
+import urllib.request
+import websocket
+
+CDP_HTTP = "http://127.0.0.1:9222"
+
+
+def http_get(path):
+    with urllib.request.urlopen(CDP_HTTP + path, timeout=5) as r:
+        return json.loads(r.read().decode())
+
+
+class CDP:
+    def __init__(self, ws_url):
+        self.ws = websocket.create_connection(ws_url, timeout=30)
+        self.msg_id = 0
+
+    def call(self, method, params=None):
+        self.msg_id += 1
+        mid = self.msg_id
+        self.ws.send(json.dumps({"id": mid, "method": method, "params": params or {}}))
+        while True:
+            resp = json.loads(self.ws.recv())
+            if resp.get("id") == mid:
+                if "error" in resp:
+                    raise RuntimeError(f"{method}: {resp['error']}")
+                return resp.get("result", {})
+            if resp.get("method") in ("Page.loadEventFired", "Page.frameStoppedLoading"):
+                pass
+
+    def eval(self, expr):
+        r = self.call("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True})
+        if r.get("exceptionDetails"):
+            return "EXC: " + json.dumps(r["exceptionDetails"], ensure_ascii=False)[:200]
+        return r.get("result", {}).get("value")
+
+    def close(self):
+        try:
+            self.ws.close()
+        except Exception:
+            pass
+
+
+def new_page(url):
+    ver = http_get("/json/version")
+    bws = CDP(ver["webSocketDebuggerUrl"])
+    t = bws.call("Target.createTarget", {"url": url})
+    tid = t["targetId"]
+    bws.close()
+    time.sleep(2)
+    for p in http_get("/json"):
+        if p["id"] == tid:
+            return tid, p["webSocketDebuggerUrl"]
+    return tid, None
+
+
+def main():
+    tid, ws = new_page("https://cma.sxbh.ltd/")
+    if not ws:
+        print("FAIL: no ws")
+        sys.exit(1)
+    c = CDP(ws)
+    c.call("Page.enable")
+    c.call("Runtime.enable")
+    time.sleep(6)
+
+    # 点击侧边栏"预算管理"
+    r = c.eval("""(() => {
+      const els = Array.from(document.querySelectorAll('a, li, span, div'));
+      const target = els.find(e => e.innerText && e.innerText.trim() === '预算管理' && e.offsetParent !== null);
+      if (!target) return 'NO_MENU';
+      target.click();
+      return 'CLICKED';
+    })()""")
+    print("MENU_CLICK:", r)
+    time.sleep(4)
+    print("URL_NOW:", c.eval("location.href"))
+    body = c.eval("document.body ? document.body.innerText.slice(0,400) : ''")
+    print("BODY:", body.replace("\n", " | ")[:400])
+
+    # 检查是否有 预算管理 页面关键元素：年份选择 + 年度分解按钮
+    btns = c.eval("""Array.from(document.querySelectorAll('button')).map((b,idx)=>({idx, text:(b.innerText||'').trim().slice(0,20)})).filter(x=>x.text)""")
+    print("BUTTONS:", json.dumps(btns, ensure_ascii=False))
+
+    c.close()
+    print("TARGET:", tid)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,79 @@
+"""budget-decompose-dialog-fix 独立复核：API 级实测 auto-decompose 全链路"""
+import json
+import sys
+import urllib.request
+
+BASE = "http://127.0.0.1:8010"
+
+
+def post(path, body, token=None, method="POST"):
+    req = urllib.request.Request(
+        BASE + path,
+        data=json.dumps(body).encode("utf-8"),
+        method=method,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}" if token else "",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return resp.status, json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        try:
+            return e.code, json.loads(e.read().decode("utf-8"))
+        except Exception:
+            return e.code, {"detail": e.read().decode("utf-8", "ignore")}
+
+
+def main():
+    # 1. 登录（账套模式 entity_id=1）
+    status, login = post("/api/cma/auth/login", {"username": "admin", "password": "admin123", "entity_id": 1})
+    token = login.get("token") or login.get("access_token")
+    if status != 200 or not token:
+        print("FAIL login:", status, login)
+        sys.exit(1)
+    print("PASS 登录成功, token 前缀:", token[:12], "...")
+
+    # 2. 调用 auto-decompose（equal 均分）
+    status, r = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token)
+    print("auto-decompose status:", status)
+    if status != 200:
+        print("  detail:", r.get("detail", r))
+        print("FAIL auto-decompose 非200（可能该年无年度预算数据）")
+        sys.exit(2)
+
+    print("  message:", r.get("message"))
+    results = r.get("results") or []
+    print("  created:", r.get("created"), " results数:", len(results))
+    for res in results[:5]:
+        print("   -", res.get("kpi_code"), res.get("kpi_name"),
+              "annual=", res.get("annual_budget"), "method=", res.get("method"),
+              "monthly_count=", len(res.get("monthly") or []))
+    if not results:
+        print("FAIL results 为空")
+        sys.exit(3)
+
+    # 3. 验证每条结果字段完整（前端表格依赖）
+    required = ["kpi_code", "kpi_name", "annual_budget", "method", "monthly"]
+    for res in results:
+        missing = [k for k in required if k not in res]
+        if missing:
+            print("FAIL 结果缺字段:", missing, res)
+            sys.exit(4)
+        if not res.get("monthly"):
+            print("FAIL monthly 为空:", res.get("kpi_code"))
+            sys.exit(5)
+    print("PASS 所有结果字段完整（kpi_code/kpi_name/annual_budget/method/monthly）")
+
+    # 4. 幂等抽查：再调一次，结果一致
+    status2, r2 = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token)
+    snap1 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in results}
+    snap2 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in (r2.get("results") or [])}
+    print("PASS 二次调用幂等一致" if snap1 == snap2 else "WARN 二次调用结果不同（非幂等）")
+
+    print("\nRESULT: API 全链路通过")
+
+
+if __name__ == "__main__":
+    main()