219 lines
7.5 KiB
Python
219 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""2026-09-07 预算版本管理UI修复 - 浏览器页面实测
|
|
流程: 登录博海(entity2) → /budget 版本管理 → 提取行DOM断言 → switch-entity 到酣客(entity1) → 再提取断言
|
|
验收: 博海 v2026H2 行: 状态草稿/条目数25/总额462/最后更新非空/提交审批按钮可见
|
|
酣客 active版本有提交审批; 2026H2 approved 显示已批准 无提交按钮
|
|
"""
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import websocket
|
|
|
|
CDP_HTTP = "http://127.0.0.1:9224"
|
|
BASE = "https://cma.sxbh.ltd/api/cma"
|
|
|
|
|
|
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=40)
|
|
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", {})
|
|
|
|
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)[:300]
|
|
return r.get("result", {}).get("value")
|
|
|
|
def close(self):
|
|
try:
|
|
self.ws.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def open_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.5)
|
|
for p in http_get("/json"):
|
|
if p["id"] == tid:
|
|
return tid, p["webSocketDebuggerUrl"]
|
|
return tid, None
|
|
|
|
|
|
def wait_for(c, expr, timeout=25, interval=1.0):
|
|
"""轮询直到 expr 为 truthy，返回其值"""
|
|
for _ in range(int(timeout / interval)):
|
|
v = c.eval(expr)
|
|
if v:
|
|
return v
|
|
time.sleep(interval)
|
|
return None
|
|
|
|
|
|
def login_via_fetch(c, entity_id):
|
|
"""同源 fetch 登录并写 localStorage，返回状态文本"""
|
|
expr = f"""
|
|
(async () => {{
|
|
const r = await fetch('{BASE}/auth/login', {{
|
|
method: 'POST', headers: {{'Content-Type': 'application/json'}},
|
|
body: JSON.stringify({{username:'admin', password:'admin123', entity_id:{entity_id}}})
|
|
}});
|
|
const j = await r.json();
|
|
if (!r.ok) return 'LOGIN_FAIL:' + JSON.stringify(j);
|
|
localStorage.setItem('cma_token', j.token);
|
|
localStorage.setItem('cma_entity_id', String(j.entity_id));
|
|
localStorage.setItem('cma_entity_name', j.entity_name || '');
|
|
localStorage.setItem('cma_user', JSON.stringify(j.user || {{}}));
|
|
return 'LOGIN_OK:' + j.entity_name + ':' + j.token.slice(0, 8);
|
|
}})()
|
|
"""
|
|
return c.eval(expr)
|
|
|
|
|
|
JS_GO_BUDGET = """
|
|
(() => {
|
|
location.href = '/budget';
|
|
return 'NAV';
|
|
})()
|
|
"""
|
|
|
|
JS_EXTRACT = r"""
|
|
(() => {
|
|
// 等表格渲染完
|
|
const rows = Array.from(document.querySelectorAll('.el-table__body-wrapper tbody tr'));
|
|
if (!rows.length) return 'NO_ROWS';
|
|
const out = [];
|
|
for (const tr of rows) {
|
|
const tds = Array.from(tr.querySelectorAll('td'));
|
|
if (tds.length < 6) continue;
|
|
const getTxt = (td) => (td.innerText || '').trim().replace(/\s+/g, ' ');
|
|
const version = getTxt(tds[0]);
|
|
const status = getTxt(tds[1]);
|
|
const planCount = getTxt(tds[2]);
|
|
const total = getTxt(tds[3]);
|
|
const updated = getTxt(tds[4]);
|
|
const opTxt = (tds[5] ? tds[5].innerText || '' : '').trim().replace(/\s+/g, ' ');
|
|
const btns = Array.from(tds[5].querySelectorAll('button')).map(b => (b.innerText || '').trim().replace(/\s+/g, ''));
|
|
const checkboxCount = tds[5].querySelectorAll('.el-checkbox').length;
|
|
out.push({ version, status, planCount, total, updated, opTxt, btns, checkboxCount });
|
|
}
|
|
return JSON.stringify(out);
|
|
})()
|
|
"""
|
|
|
|
JS_CLICK_VERSION_TAB = """
|
|
(() => {
|
|
const tabs = Array.from(document.querySelectorAll('[role="tab"]'));
|
|
const t = tabs.find(e => (e.innerText || '').replace(/\\s/g, '') === '版本管理');
|
|
if (!t) return 'NO_TAB:' + tabs.map(x => x.innerText).join('|');
|
|
t.click();
|
|
return 'TAB_CLICKED';
|
|
})()
|
|
"""
|
|
|
|
|
|
def main():
|
|
tid, ws = open_page("https://cma.sxbh.ltd/login")
|
|
if not ws:
|
|
print("FAIL: no ws for new page")
|
|
sys.exit(1)
|
|
c = CDP(ws)
|
|
c.call("Page.enable")
|
|
c.call("Runtime.enable")
|
|
time.sleep(4)
|
|
|
|
# 1) 登录博海 entity=2
|
|
r = login_via_fetch(c, 2)
|
|
print("LOGIN_E2:", r)
|
|
if not (isinstance(r, str) and r.startswith("LOGIN_OK")):
|
|
print("FAIL login entity2")
|
|
sys.exit(1)
|
|
c.eval(JS_GO_BUDGET)
|
|
time.sleep(5)
|
|
print("URL1:", c.eval("location.href"))
|
|
# 确认当前账套
|
|
print("ENTITY1:", c.eval("localStorage.getItem('cma_entity_id')"))
|
|
tab = wait_for(c, JS_CLICK_VERSION_TAB, timeout=15)
|
|
print("TAB1:", tab)
|
|
time.sleep(4)
|
|
rows_json = wait_for(c, JS_EXTRACT, timeout=20)
|
|
print("ROWS_E2_RAW:", rows_json)
|
|
rows_e2 = json.loads(rows_json) if isinstance(rows_json, str) and rows_json.startswith("[") else []
|
|
for row in rows_e2:
|
|
print(" E2", json.dumps(row, ensure_ascii=False))
|
|
|
|
# 断言博海
|
|
ok = True
|
|
target = [r for r in rows_e2 if r.get("version", "").strip().startswith("v2026H2")]
|
|
if not target:
|
|
print("FAIL: 博海未见 v2026H2")
|
|
ok = False
|
|
else:
|
|
r = target[0]
|
|
assert "草稿" in r["status"], f"状态应含草稿, got {r['status']}"
|
|
assert r["planCount"].replace(" ", "") == "25", f"条目数应25, got {r['planCount']}"
|
|
assert "462" in r["total"].replace(",", ""), f"总额应462, got {r['total']}"
|
|
assert r["updated"], "最后更新应非空"
|
|
assert "提交审批" in r["opTxt"], f"应有提交审批按钮, got {r['btns']}"
|
|
print("E2_ASSERT_OK:", json.dumps(r, ensure_ascii=False))
|
|
|
|
# 2) 切换酣客 entity=1（同页面 fetch switch-entity 或直接重新登录）
|
|
r = login_via_fetch(c, 1)
|
|
print("LOGIN_E1:", r)
|
|
c.eval(JS_GO_BUDGET)
|
|
time.sleep(5)
|
|
print("ENTITY2:", c.eval("localStorage.getItem('cma_entity_id')"))
|
|
tab = wait_for(c, JS_CLICK_VERSION_TAB, timeout=15)
|
|
print("TAB2:", tab)
|
|
time.sleep(4)
|
|
rows_json = wait_for(c, JS_EXTRACT, timeout=20)
|
|
print("ROWS_E1_RAW:", rows_json)
|
|
rows_e1 = json.loads(rows_json) if isinstance(rows_json, str) and rows_json.startswith("[") else []
|
|
for row in rows_e1:
|
|
print(" E1", json.dumps(row, ensure_ascii=False))
|
|
|
|
ok2 = True
|
|
for row in rows_e1:
|
|
ver = row.get("version", "").strip()
|
|
if ver == "2026H2":
|
|
if "已批准" not in row["status"] or "提交审批" in row["opTxt"]:
|
|
print(f"FAIL: 酣客2026H2 approved 应只读已批准无提交, got {json.dumps(row, ensure_ascii=False)}")
|
|
ok2 = False
|
|
elif row["status"].replace(" ", "").startswith("草稿"):
|
|
if "提交审批" not in row["opTxt"]:
|
|
print(f"FAIL: 酣客{ver} active/草稿 应有提交审批, got {json.dumps(row, ensure_ascii=False)}")
|
|
ok2 = False
|
|
if not rows_e1:
|
|
ok2 = False
|
|
print("FAIL: 酣客无行")
|
|
|
|
print("\n" + ("ALL_PASS" if (ok and ok2) else "HAS_FAIL"))
|
|
c.close()
|
|
sys.exit(0 if (ok and ok2) else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|