# -*- coding: utf-8 -*- """demo_smoke_graph.py — contact-graph 票 08 · 图谱 demo Playwright 冒烟(chromium headless) 被测: .scratch/customer-contact-graph/prototype/prototype-contact-graph.html?api=http://localhost:8080&customerId=<夹具> 认证: demo 无登录 UI → page.route 反代 localhost:8080/api/**(去 origin 头 + 注入 debug token Bearer + 补 CORS) 夹具: e2c-cgd* 客户(归属 ADMIN,BUDDY 解耦) + quickAdd 4 联系人(董事长/市场经理/市场专员/其他职务无档位) 票面 6 条 → 步骤映射(修订⑧适配: 公司边派生无×不可删 / 无否决机制 / 右侧栏=未设置等级): 0 打开 + 初始加载(字典/图谱真实 API) 1 票① 列表/图谱双视图切换 + 图谱元素(公司节点/等级归行/未设置等级栏/计数) 2 票② 编辑器闭环: 进入编辑(按钮变橙)→拖卡位移→.dot 真实鼠标拖线→保存(DB 取证)→重进持久 →×删除手动边→保存→重进删净 3 票③ 三弹窗: 重置确认 / 视图切换未保存三选(继续编辑+放弃修改) / 完成绘制退出(保存修改) 4 票④ 悬停气泡(显示+隐藏) 5 票⑤ 列表侧: 页内编辑(改行→撤销→改行→保存全部→DB) / quickAdd 多行(前端查重→ack→服务端 部分成功 2 成 1 败) / 脱敏点击明文(DB reveal_log 埋点取证) / 模板下载+批量导出下载 6 票⑥ 导入 UI 三段式(预检计数含 FAIL→确认执行→轮询完成→失败明细→DB 执行面) 7 网络层附带: console/page error=0 + 零 PUT/DELETE 清理: e2c-cgd* DB 硬删正门(contact_import_fail 无 customer_id 列,按 task_id 先删) 产出: shots-demo/*.png + demo_smoke_graph-results.json """ import json, os, pathlib, re, sys, io, time sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import urllib.request, urllib.parse import pymysql from playwright.sync_api import sync_playwright ROOT = pathlib.Path(r"d:\code\crm-backend-matt\.scratch\customer-contact-graph") SHOTS = ROOT / "shots-demo"; SHOTS.mkdir(exist_ok=True) DEMO = ROOT / "prototype" / "prototype-contact-graph.html" XLSX = ROOT / "_smoke-contacts.xlsx" BASE = "http://localhost:8080" ADMIN = "739564171091247104" PFX = "e2c-cgd" TS = time.strftime("%H%M%S") results = [] def step(name, ok, note=""): results.append({"step": name, "ok": bool(ok), "note": str(note)}) print(("PASS " if ok else "FAIL ") + name + (" | " + str(note) if note else "")) console_errors, page_errors, api_urls = [], [], [] # ---------------- urllib 直调(夹具/DB 取证不走浏览器) ---------------- def post(url, data, token=None): req = urllib.request.Request(url, data=urllib.parse.urlencode(data).encode(), method="POST") req.add_header("Content-Type", "application/x-www-form-urlencoded;charset=UTF-8") if token: req.add_header("Authorization", "Bearer " + token) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode()) def get_json(url, token=None): req = urllib.request.Request(url) if token: req.add_header("Authorization", "Bearer " + token) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode()) def get_token(uid): return get_json(f"{BASE}/api/auth/debug/token?userId={uid}")["data"] def dict_one(group): try: d = get_json(f"{BASE}/api/dict/item/enabled-list?groupCode={group}", tok).get("data") or [] return (d[0] or {}).get("code") if d else None except Exception: return None def db(): return pymysql.connect(host='8.129.84.155', port=3306, user='root', password='Itc@123456', database='crm', charset='utf8mb4', autocommit=True, connect_timeout=10, cursorclass=pymysql.cursors.DictCursor) def dbq(sql, args=None): with db() as conn, conn.cursor() as cur: cur.execute(sql, args) return cur.fetchall() def dbx(sql, args=None): with db() as conn, conn.cursor() as cur: cur.execute(sql, args) return cur.rowcount def _py_upload(cid, mode, strategy): """Python 侧直调真实 upload(完整 multipart,重读同一 xlsx)。绕开 Chromium 限制: post_data_buffer 不含上传文件内容(仅文件引用),转发必丢 file part → 67023。""" boundary = "----smokeBoundaryPy1234567890" parts = [] for k, v in [("customerId", cid), ("importMode", mode), ("duplicateStrategy", strategy)]: parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()) with open(XLSX, "rb") as f: fb = f.read() parts.append((f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{XLSX.name}"\r\n' f'Content-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet\r\n\r\n').encode() + fb + b"\r\n") parts.append(f"--{boundary}--\r\n".encode()) req = urllib.request.Request(BASE + "/api/customer/contact/import/upload", data=b"".join(parts), method="POST") req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}") req.add_header("Authorization", "Bearer " + tok) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read().decode()) def sweep(): """清历史 e2c-cgd* 残留(幂等重跑)。contact_import_fail 无 customer_id 列,先按 task_id 子查询删。""" rows = dbq("SELECT id FROM customer WHERE customer_name LIKE %s", (PFX + "%",)) for row in rows: cid = row["id"] dbx("DELETE FROM contact_import_fail WHERE task_id IN " "(SELECT id FROM contact_import_task WHERE customer_id=%s)", (cid,)) for t in ["customer_contact_edge", "customer_contact_graph", "customer_contact_reveal_log", "contact_import_task", "customer_contact", "customer_oplog"]: dbx(f"DELETE FROM {t} WHERE customer_id=%s", (cid,)) dbx("DELETE FROM customer WHERE id=%s", (cid,)) return len(rows) # ---------------- 夹具(与 e2e-graph.py 同模式) ---------------- def mk_customer(name): form = {"customerName": name, "customerType": CTYPE, "provinceCode": "440000", "cityCode": "440100", "districtCode": "440103", "industryCode": GOV, "customerStarLevel": 3, "relationStarLevel": 3, "isBizNegotiated": 0, "isChild": 0, "ownerUserId": ADMIN} r = post(f"{BASE}/api/customer/create", form, tok) d = r.get("data") or {} if isinstance(d, dict) and d.get("needConfirm"): r = post(f"{BASE}/api/customer/create", dict(form, confirmSimilar="true"), tok) d = r.get("data") or {} if not (isinstance(d, dict) and d.get("id")): raise RuntimeError(f"夹具客户创建失败 {name}: {str(r)[:160]}") return str(d["id"]) def quick_add(cid, rows): form = {"customerId": str(cid)} for i, row in enumerate(rows): for k, v in row.items(): form[f"rows[{i}].{k}"] = str(v) return post(f"{BASE}/api/customer/contact/quickAdd", form, tok) tok = get_token(ADMIN) CTYPE = dict_one("customer_type") or "customer_type_01" GOV = dict_one("industry") or "gov" N1, N2, N3, N4 = f"{PFX}{TS}董", f"{PFX}{TS}经", f"{PFX}{TS}专", f"{PFX}{TS}杂" CID = mk_customer(f"{PFX}-{TS}") quick_add(CID, [ {"name": N1, "jobTitleCode": "chairman", "phone": "13800001001", "isKeyContact": 1, "source": "other"}, {"name": N2, "jobTitleCode": "marketing_manager", "phone": "13800001002", "isKeyContact": 0, "source": "group_meeting"}, {"name": N3, "jobTitleCode": "marketing_specialist", "phone": "13800001003", "isKeyContact": 0, "source": "referral"}, {"name": N4, "jobTitleCode": "other_job", "phone": "13800001004", "isKeyContact": 0, "source": "other"}, ]) gd = get_json(f"{BASE}/api/customer/contact/graph/detail?customerId={CID}", tok).get("data") or {} BN = {n["name"]: str(n["contactId"]) for n in (gd.get("nodes") or [])} ID_CHAIR, ID_MGR, ID_SPEC, ID_MISC = BN[N1], BN[N2], BN[N3], BN[N4] print(f"夹具: customer={CID} 4 contacts chairman={ID_CHAIR[-6:]} mgr={ID_MGR[-6:]} spec={ID_SPEC[-6:]} misc={ID_MISC[-6:]}") # ---------------- 浏览器反代(page.route 注入 Bearer) ---------------- # 注意①: 不能用 route.fetch 转发 multipart(实测 file part 丢失 → 67023「请选择导入文件」), # 通用请求改 Python 侧 urllib 原样转发(保留原始 content-type/boundary)。 # 注意②: upload 请求走 _py_upload 分流 —— Chromium 的 post_data_buffer 不含上传文件内容, # 任何转发方式都拿不到文件字节;Python 侧重读同一 xlsx 直调真实 API,预检结果 fulfill 给页面。 _UPLOAD_RE = re.compile(r"/api/customer/contact/import/upload") _FLD_RE = {"customerId": re.compile(rb'name="customerId"\r?\n\r?\n(\d+)'), "importMode": re.compile(rb'name="importMode"\r?\n\r?\n([A-Z_]+)'), "duplicateStrategy": re.compile(rb'name="duplicateStrategy"\r?\n\r?\n([A-Z_]+)')} def handle_api(route): req = route.request cors = {"Access-Control-Allow-Origin": req.headers.get("origin") or "null", "Access-Control-Allow-Credentials": "true"} if req.method == "POST" and _UPLOAD_RE.search(req.url): buf = req.post_data_buffer or b"" cid_up = (_FLD_RE["customerId"].search(buf) or [b"", str(CID).encode()])[1].decode() mode = (_FLD_RE["importMode"].search(buf) or [b"", b"UPSERT"])[1].decode() strategy = (_FLD_RE["duplicateStrategy"].search(buf) or [b"", b"SKIP"])[1].decode() try: envelope = _py_upload(cid_up, mode, strategy) route.fulfill(status=200, body=json.dumps(envelope, ensure_ascii=False).encode(), headers={"Content-Type": "application/json", **cors}) except Exception as e: route.fulfill(status=502, headers={"Content-Type": "application/json", **cors}, body=json.dumps({"code": -1, "success": False, "message": f"py-upload: {e}"})) return h = {k: v for k, v in req.headers.items() if k.lower() not in ("origin", "authorization", "referer", "host", "content-length")} h["Authorization"] = "Bearer " + tok try: data = req.post_data_buffer if req.method not in ("GET", "HEAD") else None r2 = urllib.request.Request(req.url, data=data, method=req.method) for k, v in h.items(): r2.add_header(k, v) with urllib.request.urlopen(r2, timeout=60) as resp: body = resp.read() ct = resp.headers.get("Content-Type") or "application/json" status = resp.status route.fulfill(status=status, body=body, headers={ "Content-Type": ct, **cors}) except urllib.error.HTTPError as e: route.fulfill(status=e.code, body=e.read(), headers={ "Content-Type": e.headers.get("Content-Type") or "application/json", **cors}) except Exception as e: route.fulfill(status=502, headers={"Content-Type": "application/json", **cors}, body=json.dumps({"code": -1, "success": False, "message": f"bridge: {e}"})) def toast_text(pg, timeout=6000): try: pg.wait_for_selector("#toast.show", timeout=timeout) return (pg.text_content("#toast") or "").strip() except Exception: return "" def reload_idle(pg): pg.reload(); pg.wait_for_load_state("networkidle") # ---------------- 步骤 ---------------- def s0_open(pg): print("\n== 0 打开 + 初始加载 ==") pg.goto(DEMO.as_uri() + f"?api={BASE}&customerId={CID}") pg.wait_for_load_state("networkidle") pg.screenshot(path=str(SHOTS / "00-open.png")) cnt = (pg.text_content("#contactCount") or "").strip() step("打开 demo: 字典+图谱真实 API 加载(4 位联系人)", "共 4 位联系人" in cnt, cnt) def s1_views(pg): print("\n== 1 票① 双视图切换 + 图谱元素 ==") lv = pg.locator("#listView").is_visible() rows = pg.locator("#contactTbody tr").count() step("默认列表视图: 4 行", lv and rows == 4, f"visible={lv} rows={rows}") pg.click("#pillGraph") pg.wait_for_selector("#graphView", state="visible") gv = pg.locator("#graphView").is_visible() and not pg.locator("#listView").is_visible() comp = pg.locator(".gcard.company").count() cpos = pg.evaluate("JSON.stringify(COMPANY_POS)") step("切图谱: company 节点唯一且固定布局位", gv and comp == 1 and cpos, f"company={comp} pos={cpos}") rowok = pg.evaluate("([a,b,c]) => cardPos[a].y < cardPos[b].y && cardPos[b].y < cardPos[c].y", [ID_CHAIR, ID_MGR, ID_SPEC]) step("等级归行: 董事长(10)在市场经理(6)上方,经理在专员(3)上方", rowok) zone = pg.locator("#unrecZone").count() ztxt = (pg.text_content("#unrecZone") or "").strip() unrec = pg.evaluate("([i]) => cardPos[i].x >= document.getElementById('unrecZone').offsetLeft", [ID_MISC]) step("未设置等级栏: 其他职务卡停右栏", zone == 1 and "未设置等级" in ztxt and unrec, f"zone={zone} {ztxt[:20]}") chips = pg.locator(".gcard .lv-chip").count() step("等级 chip 数 = 3(4 卡,其他职务未设)", chips == 3, chips) pg.screenshot(path=str(SHOTS / "01-graph-view.png")) pg.click("#pillList") pg.wait_for_selector("#listView", state="visible") step("切回列表视图", pg.locator("#listView").is_visible()) def s2_editor(pg): print("\n== 2 票② 编辑器闭环 ==") pg.click("#pillGraph") pg.click("#btnDraw") btn = (pg.text_content("#btnDraw") or "").strip() editing = pg.evaluate("graphEditing") orange = pg.evaluate("document.getElementById('btnDraw').classList.contains('btn-orange')") step("进入编辑: 按钮→完成绘制+变橙+graphEditing", btn == "完成绘制" and editing and orange, f"{btn} editing={editing} orange={orange}") # 拖卡位移(同级行内水平拖,x 变等级不变;同格释放=tryLevelDrop no-op) box = pg.locator(f'.gcard[data-id="{ID_SPEC}"]').bounding_box() x0 = pg.evaluate(f"cardPos['{ID_SPEC}'].x") pg.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) pg.mouse.down() pg.mouse.move(box["x"] + box["width"] / 2 + 130, box["y"] + box["height"] / 2, steps=10) pg.mouse.up() pg.wait_for_timeout(250) x1 = pg.evaluate(f"cardPos['{ID_SPEC}'].x") lvl = pg.evaluate(f"byId('{ID_SPEC}').level") step("拖卡位移: 行内 x 位移+等级保持 3", abs(x1 - x0) > 50 and lvl == 3, f"x {x0}->{x1} lvl={lvl}") # 真实鼠标拖线: 市场经理 dot-bottom → 市场专员卡(等级 6>3 自动定向 parent=经理) dot = pg.locator(f'.gcard[data-id="{ID_MGR}"] .dot-bottom').bounding_box() tgt = pg.locator(f'.gcard[data-id="{ID_SPEC}"]').bounding_box() pg.mouse.move(dot["x"] + dot["width"] / 2, dot["y"] + dot["height"] / 2) pg.mouse.down() pg.mouse.move(tgt["x"] + tgt["width"] / 2, tgt["y"] + tgt["height"] / 2, steps=12) pg.mouse.up() pg.wait_for_timeout(300) pend = pg.evaluate("pendingAdds.length") pair = pg.evaluate("pendingAdds.length ? pendingAdds[0].parent + '>' + pendingAdds[0].child : ''") svg_pend = pg.locator("#edgeSvg path.edge-pending").count() step("拖线: dot→目标卡 pending(经理>专员)", pend == 1 and pair == f"{ID_MGR}>{ID_SPEC}" and svg_pend >= 1, f"pair={pair[-18:]} svg={svg_pend}") pg.screenshot(path=str(SHOTS / "02-editor-drag.png")) # 保存 → DB 取证 page_before = dbq("SELECT version FROM customer_contact_graph WHERE customer_id=%s", (int(CID),)) pg.click("#btnGraphSave") t = toast_text(pg) edges_db = dbq("SELECT parent_id, child_id FROM customer_contact_edge WHERE customer_id=%s", (int(CID),)) dbp = [(int(r["parent_id"]), int(r["child_id"])) for r in edges_db] step("保存: toast(版本)+DB edge 落库(经理>专员)", "保存成功(版本" in t and dbp == [(int(ID_MGR), int(ID_SPEC))] and page_before, f"{t[:30]} db={dbp}") # 重进持久 reload_idle(pg) committed = pg.evaluate("effectiveEdges().length") step("重进: 手动边持久(committed=1)", committed == 1, f"committed={committed}") pg.click("#pillGraph") pg.wait_for_selector("#graphView", state="visible") pg.screenshot(path=str(SHOTS / "03-persist.png")) # × 删除手动边 → 保存 → 删净 pg.click("#btnDraw") pg.locator("#edgeSvg g.edge-g:not(.edge-company)").locator(".edge-del").click() pg.wait_for_timeout(200) pdel = pg.evaluate("pendingDeletes.length") pg.click("#btnGraphSave") t2 = toast_text(pg) pg.wait_for_timeout(300) edges_db2 = dbq("SELECT id FROM customer_contact_edge WHERE customer_id=%s", (int(CID),)) step("× 删除手动边→保存: DB 删净", pdel == 1 and "保存成功" in t2 and not edges_db2, f"pdel={pdel} db={len(edges_db2)}") reload_idle(pg) committed2 = pg.evaluate("effectiveEdges().length") step("重进: 手动边已删净(0)", committed2 == 0, f"committed={committed2}") pg.click("#pillGraph") pg.wait_for_selector("#graphView", state="visible") pg.screenshot(path=str(SHOTS / "04-deleted.png")) def s3_dialogs(pg): print("\n== 3 票③ 三弹窗 ==") # a. 重置确认(有 pending 时弹确认,确定→回进编辑快照) pg.click("#btnDraw") pg.evaluate(f"tryConnect('{ID_MGR}','{ID_SPEC}')") pg.wait_for_timeout(200) pg.click("#btnGraphReset") pg.wait_for_selector("#dlgOverlay.show") title = (pg.text_content("#dlgTitle") or "").strip() pg.click("#dlgFoot button.btn-primary") pg.wait_for_timeout(300) pend_after = pg.evaluate("pendingAdds.length") step("重置确认弹窗: 确定→pending 清零", title == "重置" and pend_after == 0, f"title={title} pend={pend_after}") # b. 视图切换未保存三选: 继续编辑保持 → 再触发 → 放弃修改退出+切视图 pg.evaluate(f"tryConnect('{ID_MGR}','{ID_SPEC}')") pg.wait_for_timeout(200) pg.click("#pillList") pg.wait_for_selector("#dlgOverlay.show") labels = pg.locator("#dlgFoot button").all_text_contents() pg.locator("#dlgFoot button").first.click() # 继续编辑 pg.wait_for_timeout(250) keep = pg.evaluate("graphEditing") and not pg.evaluate( "document.getElementById('dlgOverlay').classList.contains('show')") pg.click("#pillList") pg.wait_for_selector("#dlgOverlay.show") pg.locator("#dlgFoot button.btn-danger-ghost").click() # 放弃修改 pg.wait_for_timeout(350) t = toast_text(pg) dropped = (not pg.evaluate("graphEditing")) and pg.locator("#listView").is_visible() \ and pg.evaluate("pendingAdds.length") == 0 step("未保存三选: 继续编辑保持态/放弃修改退出+切列表", labels == ["继续编辑", "放弃修改", "保存修改"] and keep and dropped, f"labels={labels} keep={keep} dropped={dropped} toast={t[:24]}") pg.screenshot(path=str(SHOTS / "05-dialogs.png")) # c. 完成绘制退出(保存修改路径) → 边落库+退编辑+重进持久 pg.click("#pillGraph") pg.click("#btnDraw") pg.evaluate(f"tryConnect('{ID_MGR}','{ID_SPEC}')") pg.wait_for_timeout(200) pg.click("#btnDraw") # 完成绘制 → 未保存三选 pg.wait_for_selector("#dlgOverlay.show") pg.locator("#dlgFoot button.btn-primary").click() # 保存修改 pg.wait_for_timeout(500) t3 = toast_text(pg) back = (not pg.evaluate("graphEditing")) and (pg.text_content("#btnDraw") or "").strip() == "绘制自定义关系" edges_db = dbq("SELECT parent_id, child_id FROM customer_contact_edge WHERE customer_id=%s", (int(CID),)) dbp = [(int(r["parent_id"]), int(r["child_id"])) for r in edges_db] step("完成绘制退出: 保存修改→边落库+退编辑", back and dbp == [(int(ID_MGR), int(ID_SPEC))], f"toast={t3[:24]} db={dbp}") reload_idle(pg) step("保存修改路径: 重进边持久", pg.evaluate("effectiveEdges().length") == 1) def s4_bubble(pg): print("\n== 4 票④ 悬停气泡 ==") pg.click("#pillGraph") pg.wait_for_selector("#graphView", state="visible") card = pg.locator(f'.gcard[data-id="{ID_MGR}"]') card.hover() pg.wait_for_timeout(350) vis = pg.evaluate("document.getElementById('bubble').style.display") txt = (pg.text_content("#bubble") or "").strip() step("悬停气泡: 显示+姓名+字段", vis == "block" and N2 in txt and "人脉关系情况" in txt, f"display={vis} {txt[:40]}") pg.screenshot(path=str(SHOTS / "06-bubble.png")) wrap = pg.locator("#graphWrap").bounding_box() pg.mouse.move(wrap["x"] + 6, wrap["y"] + 6) pg.wait_for_timeout(250) vis2 = pg.evaluate("document.getElementById('bubble').style.display") step("移出隐藏气泡", vis2 == "none", f"display={vis2}") def s5_list(pg): print("\n== 5 票⑤ 列表侧 ==") pg.click("#pillList") pg.wait_for_selector("#listView", state="visible") # 页内编辑 pg.click("#btnEditList") btn = (pg.text_content("#btnEditList") or "").strip() step("进入页内编辑: 按钮→退出编辑", btn == "退出编辑", btn) new_name = f"{PFX}{TS}专员乙" pg.fill(f'tr[data-id="{ID_SPEC}"] input[data-f="name"]', new_name) pg.wait_for_timeout(200) dirty = pg.locator(f'tr[data-id="{ID_SPEC}"].dirty-row').count() undo = pg.locator(f'tr[data-id="{ID_SPEC}"] .undo-link').count() step("改姓名: dirty 行标记+撤销链接", dirty == 1 and undo == 1, f"dirty={dirty} undo={undo}") pg.locator(f'tr[data-id="{ID_SPEC}"] .undo-link').click() t = toast_text(pg) back = pg.locator(f'tr[data-id="{ID_SPEC}"] input[data-f="name"]').input_value() step("撤销修改: 姓名恢复+toast", "已撤销该行修改" in t and back == N3, f"{back} | {t[:20]}") pg.fill(f'tr[data-id="{ID_SPEC}"] input[data-f="name"]', new_name) pg.evaluate("document.getElementById('toast').className=''") # 清残留 toast,避免旧 toast 干扰断言 pg.click("#btnSaveList") t2 = toast_text(pg) pg.wait_for_timeout(400) row = dbq("SELECT name FROM customer_contact WHERE id=%s", (int(ID_SPEC),)) step("保存全部: toast+DB 姓名更新(脱敏电话原样回传=不改)", "保存全部成功" in t2 and row and row[0]["name"] == new_name, f"{t2[:20]} db={row[0]['name'] if row else '?'}") pg.click("#btnEditList") # 保存后 demo 仍处编辑态(listEditing=true) → 退出恢复只读列表 pg.wait_for_timeout(350) pg.screenshot(path=str(SHOTS / "07-list-edit.png")) # 快速添加多行(前端查重 → ack → 服务端部分成功) pg.click("#btnAddContact") pg.wait_for_selector("#qaOverlay.show") pg.click("#qaAddRow") # 默认 2 行,补第 3 行(重复演示行) na, nb, nc = f"{PFX}{TS}新增A", f"{PFX}{TS}新增B", f"{PFX}{TS}新增C" pg.fill('#qaTbody tr[data-qa="0"] input[data-qf="name"]', na) pg.select_option('#qaTbody tr[data-qa="0"] select[data-qf="cat"]', "marketing") pg.wait_for_timeout(150) pg.select_option('#qaTbody tr[data-qa="0"] select[data-qf="job"]', "marketing_specialist") pg.fill('#qaTbody tr[data-qa="0"] input[data-qf="phone"]', "13800001005") pg.select_option('#qaTbody tr[data-qa="0"] select[data-qf="source"]', "group_meeting") pg.check('#qaTbody tr[data-qa="0"] input[data-qf="key"]') pg.fill('#qaTbody tr[data-qa="1"] input[data-qf="name"]', nb) pg.select_option('#qaTbody tr[data-qa="1"] select[data-qf="cat"]', "purchase") pg.wait_for_timeout(150) pg.select_option('#qaTbody tr[data-qa="1"] select[data-qf="job"]', "purchase_specialist") pg.fill('#qaTbody tr[data-qa="1"] input[data-qf="phone"]', "13800001006") pg.fill('#qaTbody tr[data-qa="2"] input[data-qf="name"]', nc) pg.select_option('#qaTbody tr[data-qa="2"] select[data-qf="cat"]', "marketing") pg.wait_for_timeout(150) pg.select_option('#qaTbody tr[data-qa="2"] select[data-qf="job"]', "marketing_manager") pg.fill('#qaTbody tr[data-qa="2"] input[data-qf="phone"]', "13800001001") # 重复=董事长 pg.evaluate("document.getElementById('toast').className=''") pg.click("#qaSubmit") t3 = toast_text(pg) hint = (pg.text_content('#qaTbody tr[data-qa="2"] .qa-hint') or "").strip() dup = pg.locator('#qaTbody tr[data-qa="2"].qa-dup').count() step("quickAdd 前端查重: 重复行拦截+确认知晓提示", "提交失败" in t3 and "疑似重复" in hint and dup == 1, f"{t3[:30]} | hint={hint[:36]}") pg.check('#qaTbody tr[data-qa="2"] input[data-qf="ack"]') pg.evaluate("document.getElementById('toast').className=''") pg.click("#qaSubmit") t4 = toast_text(pg) hint2 = (pg.text_content('#qaTbody tr[data-qa="2"] .qa-hint') or "").strip() step("quickAdd 部分成功: 2 成 1 败(服务端硬拒重复,失败行保留)", "2 行成功" in t4 and "1 行失败" in t4, f"{t4[:34]} | {hint2[:30]}") rows_qa = dbq("SELECT id FROM customer_contact WHERE customer_id=%s AND name IN (%s,%s,%s)", (int(CID), na, nb, nc)) step("quickAdd 落库: 2 新联系人在库", len(rows_qa) == 2, f"db={len(rows_qa)}") pg.screenshot(path=str(SHOTS / "08-qa.png")) pg.click("#qaClose") pg.wait_for_timeout(250) # 脱敏点击明文 + DB reveal_log 埋点 pg.locator(f'tr[data-id="{ID_MGR}"] .phone-masked').click() pg.wait_for_timeout(500) plain = (pg.text_content(f'tr[data-id="{ID_MGR}"] .phone-plain') or "").strip() logs = dbq("SELECT id FROM customer_contact_reveal_log WHERE customer_id=%s AND contact_id=%s", (int(CID), int(ID_MGR))) step("脱敏点击明文: 明文渲染+DB reveal_log 埋点", plain == "13800001002" and len(logs) >= 1, f"plain={plain} logs={len(logs)}") pg.screenshot(path=str(SHOTS / "09-reveal.png")) # 模板下载 + 批量导出下载 pg.click("#btnImport") pg.wait_for_selector("#impOverlay.show") with pg.expect_download() as dl: pg.click("#impTemplate") d0 = dl.value step("导入模板下载(xlsx)", d0.suggested_filename == "联系人导入模板-V1.xlsx" and os.path.getsize(d0.path()) > 0, f"{d0.suggested_filename} {os.path.getsize(d0.path())}B") pg.click("#impClose") pg.wait_for_timeout(250) pg.check("#chkAll") pg.wait_for_timeout(250) step("勾选全选: 导出按钮解锁", pg.locator("#btnExport").is_enabled()) with pg.expect_download() as dl2: pg.click("#btnExport") d1 = dl2.value step("批量导出下载(联系人导出.xlsx)", d1.suggested_filename == "联系人导出.xlsx" and os.path.getsize(d1.path()) > 0, f"{d1.suggested_filename} {os.path.getsize(d1.path())}B") def s6_import(pg): print("\n== 6 票⑥ 导入 UI 三段式 ==") from openpyxl import Workbook good, bad = f"{PFX}{TS}导入甲", f"{PFX}{TS}导入乙" wb = Workbook(); ws = wb.active ws.append(["姓名", "职务（字典编码）", "手机号", "来源（字典编码）", "是否关键联系人(是/否)"]) ws.append([good, "chairman", "13800001007", "other", "否"]) ws.append([bad, "marketing_manager", "", "other", ""]) wb.save(XLSX) pg.click("#btnImport") pg.wait_for_selector("#impOverlay.show") pg.set_input_files("#impFile", str(XLSX)) pg.click("#impUpload") pg.wait_for_selector("#impStep2", state="visible", timeout=30000) head = (pg.text_content("#impPreviewHead") or "").strip() step("预检计数头: 共 2 行/新增 1/失败 1", "共 2 行" in head and "新增 1" in head and "失败 1" in head, head[:70]) prows = pg.locator("#impPreviewTbody tr").count() has_fail = "FAIL" in (pg.text_content("#impPreviewTbody") or "") step("预检明细 2 行含 FAIL", prows == 2 and has_fail, f"rows={prows}") pg.screenshot(path=str(SHOTS / "10-import-preview.png")) pg.click("#impConfirm") pg.wait_for_selector("#impStep3", state="visible") head3 = "" for _ in range(30): pg.wait_for_timeout(1000) head3 = (pg.text_content("#impResultHead") or "").strip() if "已完成" in head3 or "已失败" in head3: break step("确认执行→轮询完成: 已完成 新增 1", "已完成:新增 1" in head3, head3[:60]) fails = pg.locator("#impFailTbody tr").count() ftxt = (pg.text_content("#impFailTbody") or "").strip() step("失败明细 1 行(手机号必填)", fails == 1 and "手机号" in ftxt, f"rows={fails} {ftxt[:40]}") g_rows = dbq("SELECT id FROM customer_contact WHERE customer_id=%s AND name=%s", (int(CID), good)) b_rows = dbq("SELECT id FROM customer_contact WHERE customer_id=%s AND name=%s", (int(CID), bad)) step("DB 执行面: 好行入库坏行未入", len(g_rows) == 1 and not b_rows, f"good={len(g_rows)} bad={len(b_rows)}") pg.screenshot(path=str(SHOTS / "11-import-done.png")) pg.click("#impDone") pg.wait_for_timeout(300) cnt = (pg.text_content("#contactCount") or "").strip() step("导入后联系人计数 = 7(4+2qa+1导入)", "共 7 位" in cnt, cnt) def s7_network(pg): print("\n== 7 网络层附带断言 ==") step("console error = 0", not console_errors, "; ".join(console_errors[:3])) step("page error = 0", not page_errors, "; ".join(page_errors[:3])) bad = [u for u in api_urls if re.match(r"^(PUT|DELETE) ", u)] step("网络层零 PUT/DELETE", not bad, "; ".join(bad[:3])) print(f" (捕获 /api/** 响应 {len(api_urls)} 条)") # ---------------- 主流程 ---------------- def main(): try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) pg = browser.new_page() pg.set_viewport_size({"width": 1720, "height": 1080}) # 图谱视图整体在视口内,鼠标拖拽坐标才有效 pg.set_default_timeout(20000) pg.on("console", lambda m: console_errors.append(m.text) if m.type == "error" else None) pg.on("pageerror", lambda e: page_errors.append(str(e))) pg.on("response", lambda r: api_urls.append(f"{r.request.method} {r.url}") if "/api/" in r.url else None) pg.route("**/api/**", handle_api) s0_open(pg) s1_views(pg) s2_editor(pg) s3_dialogs(pg) s4_bubble(pg) s5_list(pg) s6_import(pg) s7_network(pg) pg.screenshot(path=str(SHOTS / "99-final.png"), full_page=True) browser.close() finally: n = sweep() print(f"\nteardown: sweep e2c-cgd* 客户组 {n} 组(DB 硬删正门)") n_pass = sum(1 for r in results if r["ok"]) print(f"\n=== demo_smoke_graph: {n_pass}/{len(results)} PASS ===") for r in results: if not r["ok"]: print(f" FAIL {r['step']} | {r['note'][:120]}") out = ROOT / "demo_smoke_graph-results.json" out.write_text(json.dumps(results, ensure_ascii=False, indent=1), encoding="utf-8") print(f"results → {out}") sys.exit(0 if n_pass == len(results) else 1) if __name__ == "__main__": main()