# -*- coding: utf-8 -*- """demo_t04_smoke.py — customer-frontend-handover 票 04 · 主 demo 冒烟（真实 API 反代形态, chromium headless） 被测: .scratch/customer-e2e/demo/index.html?api=http://localhost:8080（本地起服 verify profile） 认证: demo 无登录 UI → page.route 反代 localhost:8080/api/**（剥 origin 头 + 注入 debug token Bearer + 补 CORS） 夹具: e2c-t04* 客户（归属 ADMIN）+ quickAdd 4 联系人（董事长/市场经理/市场专员/其他职务） 覆盖（票 02 收编后全交互 + 图谱编辑器真实鼠标拖拽/定级, 对齐 demo_smoke_graph 形态）: 1 列表检索夹具 + focusFlag 星标渲染 ☆/★（票 01 列） 2 toggleFocus 双向（DB customer_focus 取证） 3 进详情 + 快速添加按钮替换旧「+联系人」 4 contactsShell 常驻面板 + 联系人列表 4 行（旧端点全字段）+ 脱敏 5 图谱视图（等级归行 + 公司派生边无 ×） 6 reveal 明文（列表+图谱双写, DB reveal_log 埋点取证） 7 quickAdd 快速添加（DB 联系人 +1, 双刷回显） 8 batchEdit 页内编辑（DB 回显取证） 9 图谱编辑器: 拖卡位移 / .dot 真实鼠标拖线 → save（DB customer_contact_edge 取证）→ 切走重进持久 10 定级: 拖「其他职务」卡入等级行 → level 落库 → save（DB level 取证） 11 删边 × → save → DB 删净 12 导入三段式 cimp（openpyxl xlsx; multipart 分流 _py_upload; 预检计数 → 确认 → 轮询 → 失败明细 → DB 执行面） 13 export 下载 / delContact gConfirm 双刷 / 页签显隐互斥 网络层: console/page error=0 + 零 PUT/DELETE + 12 端点族触发 清理: e2c-t04* DB 硬删正门（含 customer_focus） 产出: shots-t04/*.png + t04-smoke-result.json """ import io, os, sys, json, re, time, urllib.request, urllib.parse, urllib.error sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') import pymysql from playwright.sync_api import sync_playwright ROOT = r"e:\code\crm-backend-matt\.scratch\customer-frontend-handover" SHOTS = os.path.join(ROOT, "shots-t04"); os.makedirs(SHOTS, exist_ok=True) URL = 'file:///' + os.path.join(r"e:\code\crm-backend-matt\.scratch\customer-e2e", 'demo', 'index.html').replace('\\', '/') + '?api=http://localhost:8080' BASE = "http://localhost:8080" ADMIN = "739564171091247104" PFX = "e2c-t04" TS = time.strftime("%H%M%S") steps, errors, api_urls = [], [], [] def step(name, ok, note=""): steps.append({"step": name, "ok": bool(ok), "note": str(note)}) print(("PASS " if ok else "FAIL ") + name + (" | " + str(note) if note else "")) # ---------------- HTTP 直调（夹具/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, tries=5): for _ in range(tries): try: d = get_json(f"{BASE}/api/auth/debug/token?userId={uid}").get("data") if d: return d if isinstance(d, str) else d.get("token") except Exception: pass time.sleep(2) raise RuntimeError("debug/token 重试后仍失败") 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 sweep(): """清历史 e2c-t04* 残留（幂等重跑）。本冒烟有 focus 交互 → 清单含 customer_focus。""" 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", "customer_focus"]: dbx(f"DELETE FROM {t} WHERE customer_id=%s", (cid,)) dbx("DELETE FROM customer WHERE id=%s", (cid,)) return len(rows) def dict_one(group, token): try: d = get_json(f"{BASE}/api/dict/item/enabled-list?groupCode={group}", token).get("data") or [] return (d[0] or {}).get("code") if d else None except Exception: return None def mk_customer(name, token, ctype, gov): 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, token) d = r.get("data") or {} if isinstance(d, dict) and d.get("needConfirm"): r = post(f"{BASE}/api/customer/create", dict(form, confirmSimilar="true"), token) 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, token): 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, token) # ---------------- 浏览器反代（page.route 注入 Bearer；multipart upload 分流 Python 直调） ---------------- _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 _py_upload(cid, mode, strategy, token): """Python 侧直调真实 upload（完整 multipart, 重读同一 xlsx）。Chromium post_data_buffer 不含上传文件内容（仅文件引用）, 任何转发方式都丢 file part → 67023, 与 graph 冒烟同款分流。""" boundary = "----smokeBoundaryT04001234567890" 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="{os.path.basename(XLSX)}"\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 " + token) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read().decode()) def make_handle(route_token): 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"", 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_token) 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 " + route_token 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}"})) return handle_api # ================= 主流程 ================= XLSX = os.path.join(ROOT, "_t04-contacts.xlsx") tok = get_token(ADMIN) removed = sweep() print(f"前置: 清历史 e2c-t04* 残留 {removed} 条") CTYPE = dict_one("customer_type", tok) or "customer_type_01" GOV = dict_one("industry", tok) or "gov" N1, N2, N3, N4 = f"{PFX}{TS}董", f"{PFX}{TS}经", f"{PFX}{TS}专", f"{PFX}{TS}杂" CNAME = f"{PFX}-{TS}" CID = mk_customer(CNAME, tok, CTYPE, GOV) 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"}, ], tok) 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} contacts 董事长={ID_CHAIR[-6:]} 经理={ID_MGR[-6:]} 专员={ID_SPEC[-6:]} 杂={ID_MISC[-6:]}") from openpyxl import Workbook wb = Workbook(); ws = wb.active ws.append(["姓名", "职务（字典编码）", "手机号", "来源（字典编码）", "是否关键联系人(是/否)"]) ws.append([f"{PFX}{TS}导入甲", "chairman", "13800001007", "other", "否"]) ws.append(["", "marketing_manager", "13800001008", "other", ""]) # 缺姓名 → 预检 FAIL wb.save(XLSX) with sync_playwright() as p: b = p.chromium.launch() pg = b.new_page(viewport={'width': 1720, 'height': 1080}, accept_downloads=True) pg.on('console', lambda m: errors.append('console.error: ' + m.text) if m.type == 'error' else None) pg.on('pageerror', lambda e: errors.append('pageerror: ' + str(e))) pg.on('request', lambda r: api_urls.append(r.method + ' ' + r.url) if 'localhost:8080' in r.url and '/api/' in r.url else None) pg.route('http://localhost:8080/**', make_handle(tok)) me = get_json(f"{BASE}/api/auth/me", tok).get("data") or {"userId": ADMIN, "userName": "演示管理员"} pg.add_init_script("localStorage.setItem('e2c_token', '%s');" "localStorage.setItem('e2c_me', %s);" % (tok, json.dumps(me, ensure_ascii=False))) pg.goto(URL) pg.wait_for_timeout(1500) # 1 列表检索夹具 + focusFlag 星标渲染 pg.fill('#fKeyword', PFX) pg.click('#tab-list .viewbar button:has-text("查询")') pg.wait_for_selector('#listBody tr td:not(.empty)', timeout=15000) row1 = pg.locator('#listBody tr').first name_txt = (row1.locator('td').nth(1).text_content() or '').strip() step('检索命中夹具客户', CNAME in pg.content(), name_txt[:40]) star_off = row1.locator('.star-link.off').count() star_on = row1.locator('.star-link.on').count() + row1.locator('.star-link:not(.off)').count() step('focusFlag 0 → 星标 ☆（票 01 列）', star_off == 1 and star_on == 0, f'off={star_off} on={star_on}') pg.screenshot(path=os.path.join(SHOTS, '01-list.png'), full_page=True) # 2 toggleFocus 双向（DB 取证） row1.locator('.star-link').click() pg.wait_for_selector(".toast-item:has-text('已关注')", timeout=8000) pg.wait_for_timeout(800) frows = dbq("SELECT id FROM customer_focus WHERE customer_id=%s AND user_id=%s", (int(CID), ADMIN)) ok_focus = len(frows) == 1 row1 = pg.locator('#listBody tr').first star_on2 = row1.locator('.star-link:not(.off)').count() step('focus → toast+DB 单行+星标 ★', ok_focus and star_on2 == 1, f'db={len(frows)} on={star_on2}') pg.locator('#listBody tr').first.locator('.star-link').click() pg.wait_for_selector(".toast-item:has-text('已取关')", timeout=8000) pg.wait_for_timeout(800) frows2 = dbq("SELECT id FROM customer_focus WHERE customer_id=%s AND user_id=%s", (int(CID), ADMIN)) step('unfocus → DB 行删除', len(frows2) == 0, f'db={len(frows2)}') # 3 进详情 + 快速添加按钮 pg.locator('#listBody td.row').first.click() pg.wait_for_selector('#headBox b', timeout=15000) qabtn = pg.locator('#headBox button:has-text("快速添加")').count() oldbtn = pg.locator('#headBox button:has-text("+联系人")').count() step('loadHead + 快速添加按钮替换旧+联系人', qabtn == 1 and oldbtn == 0, f'qa={qabtn} old={oldbtn}') # 4 contactsShell 常驻面板 + 联系人列表 pg.click("#detailPanel .subtabs button[data-s='contacts']") pg.wait_for_selector('#contactTbody tr', timeout=15000) shell_visible = pg.evaluate("document.getElementById('contactsShell').style.display !== 'none'") sub_hidden = pg.evaluate("document.getElementById('subBox').style.display === 'none'") step('contactsShell 显示 + subBox 隐藏', shell_visible and sub_hidden, f'shell={shell_visible} sub={sub_hidden}') crows = pg.locator('#contactTbody tr').count() masked = pg.locator('#contactTbody .phone-masked').count() step('联系人列表 4 行（旧端点全字段）', crows == 4, crows) step('列表电话脱敏 == 4', masked == 4, masked) pg.screenshot(path=os.path.join(SHOTS, '02-contacts.png'), full_page=True) # 5 图谱视图 pg.click('#pillGraph') pg.wait_for_timeout(900) cards = pg.locator('#canvasInner .gcard:not(.company)').count() comp = pg.locator('#edgeSvg .edge-company').count() compdel = pg.locator('#edgeSvg .edge-company .edge-del').count() manual = pg.locator('#edgeSvg g.edge-g:not(.edge-company)').count() step('图谱卡片 4', cards == 4, cards) step('公司派生边 1（顶层董事长 lv10）+ 手动边 0', comp == 1 and manual == 0, f'comp={comp} manual={manual}') step('公司派生边无 ×（修订⑧ D8）', compdel == 0, compdel) pg.screenshot(path=os.path.join(SHOTS, '03-graph.png'), full_page=True) # 6 reveal 明文（列表+图谱双写 + DB 埋点） pg.click('#pillList') pg.wait_for_timeout(300) pg.locator('#contactTbody .phone-masked').first.click() pg.wait_for_timeout(600) plain = pg.locator('#contactTbody .phone-plain').count() step('reveal 明文（列表）', plain >= 1, plain) pg.click('#pillGraph') pg.wait_for_timeout(500) gplain = pg.locator('.gcard .gphone.revealed').count() step('reveal 双写图谱节点', gplain >= 1, gplain) rl = dbq("SELECT id FROM customer_contact_reveal_log WHERE customer_id=%s", (int(CID),)) step('DB reveal_log 埋点', len(rl) >= 1, len(rl)) pg.click('#pillList') pg.wait_for_timeout(300) # 7 quickAdd（DB +1, 双刷） pg.click('#btnAddContact') pg.wait_for_selector('#qaOverlay.show', timeout=8000) pg.fill('#qaTbody tr[data-qa="0"] input[data-qf="name"]', f'{PFX}{TS}速加') pg.select_option('#qaTbody tr[data-qa="0"] select[data-qf="cat"]', 'marketing') pg.wait_for_timeout(200) 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"]', '13900001111') pg.click('#qaOverlay .modal-foot .btn-primary') pg.wait_for_selector(".toast-item:has-text('成功添加')", timeout=10000) pg.wait_for_timeout(1000) crows2 = pg.locator('#contactTbody tr').count() cdb = dbq("SELECT id FROM customer_contact WHERE customer_id=%s AND deleted=0", (int(CID),)) step('quickAdd → 双刷 5 行 + DB 5 行', crows2 == 5 and len(cdb) == 5, f'ui={crows2} db={len(cdb)}') pg.screenshot(path=os.path.join(SHOTS, '04-qa.png'), full_page=True) # 8 batchEdit 页内编辑（DB 取证） new_id = pg.evaluate("Array.from(document.querySelectorAll('#contactTbody tr')).map(r => r.dataset.id).find(id => !['%s','%s','%s','%s'].includes(id))" % (ID_CHAIR, ID_MGR, ID_SPEC, ID_MISC)) pg.click('#btnEditList') pg.wait_for_timeout(300) pg.fill(f'#contactTbody tr[data-id="{new_id}"] input[data-f="name"]', f'{PFX}{TS}改名') pg.wait_for_timeout(200) dirty = pg.locator('.dirty-row').count() pg.click('#btnSaveList') pg.wait_for_selector(".toast-item:has-text('保存全部成功')", timeout=10000) pg.wait_for_timeout(800) nm = pg.locator(f'#contactTbody tr[data-id="{new_id}"] input[data-f="name"]').input_value() ndb = dbq("SELECT name FROM customer_contact WHERE id=%s", (int(new_id),)) step('batchEdit 保存 + DB 回显', dirty == 1 and nm == f'{PFX}{TS}改名' and ndb and ndb[0]['name'] == f'{PFX}{TS}改名', f'dirty={dirty} db={ndb[0]["name"] if ndb else None}') pg.click('#btnEditList') pg.wait_for_timeout(300) # 9 图谱编辑器: 拖卡位移 / 真实鼠标拖线 → save → DB 取证 → 重进持久 pg.click('#pillGraph') pg.wait_for_timeout(700) pg.click('#btnGraphEdit') pg.wait_for_timeout(400) dots = pg.locator('.gcard .dot').count() step('编辑态圆点出现', dots > 0, dots) box = pg.locator(f'.gcard[data-id="{ID_SPEC}"]').bounding_box() x0 = pg.evaluate(f"G.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(300) x1 = pg.evaluate(f"G.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 = 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(400) pend = pg.evaluate("G.pendingAdds.length") pair = pg.evaluate("G.pendingAdds.length ? G.pendingAdds[0].parent + '>' + G.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=os.path.join(SHOTS, '05-editor-drag.png')) pg.click('#btnGraphSave') pg.wait_for_selector(".toast-item:has-text('保存成功')", timeout=10000) pg.wait_for_timeout(500) edges_db = dbq("SELECT parent_id, child_id FROM customer_contact_edge WHERE customer_id=%s", (int(CID),)) pair_db = {f"{e['parent_id']}>{e['child_id']}" for e in edges_db} step('save → DB 手动边落库（经理>专员）', f'{ID_MGR}>{ID_SPEC}' in pair_db, f'db_edges={sorted(p[-12:] for p in pair_db)}') pg.click('#pillList') pg.wait_for_timeout(300) pg.click('#pillGraph') pg.wait_for_timeout(900) manual2 = pg.locator('#edgeSvg g.edge-g:not(.edge-company)').count() step('切走重进: 手动边持久渲染', manual2 == 1, manual2) # 10 定级: 拖「其他职务」卡入等级 6 行 → save → DB level 取证（toggle 语义守卫） if not pg.evaluate("G.editing"): pg.click('#btnGraphEdit') pg.wait_for_timeout(400) mcard = pg.locator(f'.gcard[data-id="{ID_MISC}"]').bounding_box() lv6row = pg.locator('.lv-row[data-lv="6"]').bounding_box() pg.mouse.move(mcard["x"] + mcard["width"] / 2, mcard["y"] + mcard["height"] / 2) pg.mouse.down() pg.mouse.move(lv6row["x"] + 260, lv6row["y"] + lv6row["height"] / 2, steps=14) pg.wait_for_timeout(200) pg.mouse.up() pg.wait_for_timeout(400) mlevel = pg.evaluate(f"byId('{ID_MISC}').level") step('拖卡入等级行: 杂工定级 6', mlevel == 6, f'level={mlevel}') pg.screenshot(path=os.path.join(SHOTS, '06-editor-level.png')) pg.click('#btnGraphSave') pg.wait_for_selector(".toast-item:has-text('保存成功')", timeout=10000) pg.wait_for_timeout(500) ldb = dbq("SELECT job_title_level l FROM customer_contact WHERE id=%s", (int(ID_MISC),)) lv_db = (ldb[0]['l'] if ldb and ldb[0]['l'] is not None else None) # job_title_level 列为档位 code/level 视实现而定, 允许 6 或映射值; 断言非空即定级落库 step('save → DB level 落库', lv_db is not None, f'db={lv_db}') # 11 删边 × → save → DB 删净（toggle 语义守卫: 仅非编辑态才点进入） if not pg.evaluate("G.editing"): pg.click('#btnGraphEdit') pg.wait_for_timeout(300) pg.locator('#edgeSvg g.edge-g:not(.edge-company)').first.locator('.edge-del').click() pg.wait_for_timeout(300) pg.click('#btnGraphSave') pg.wait_for_selector(".toast-item:has-text('保存成功')", timeout=10000) pg.wait_for_timeout(500) edges_db2 = dbq("SELECT parent_id, child_id FROM customer_contact_edge WHERE customer_id=%s", (int(CID),)) step('删边 save → DB 删净', len(edges_db2) == 0, f'db_edges={len(edges_db2)}') if pg.evaluate("G.editing"): pg.click('#btnGraphEdit') # 退出编辑（toggle; pending 空 → 直接退出） pg.wait_for_timeout(300) pg.click('#pillList') pg.wait_for_timeout(300) # 12 导入三段式 cimp（真实 upload 分流 → 确认 → 轮询 → 失败明细 → DB 执行面） pg.click('#contactsShell button[onclick="openCimp()"]') pg.wait_for_selector('#cimpOverlay.show', timeout=8000) pg.set_input_files('#cimpFile', {'name': 'contacts.xlsx', 'mimeType': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'buffer': open(XLSX, 'rb').read()}) pg.click('#cimpUploadBtn') pg.wait_for_timeout(1500) head2 = pg.text_content('#cimpPreviewHead') or '' step('cimp 上传预检（共 2 行）', '共 2 行' in head2, head2[:80]) pre_fail = pg.locator('#cimpFailTbody tr').count() if pg.locator('#cimpFailTbody').count() else 0 pg.click('#cimpConfirmBtn') head3 = '' for _ in range(20): pg.wait_for_timeout(800) head3 = pg.text_content('#cimpResultHead') or '' if '已完成' in head3 or '已失败' in head3: break step('cimp 轮询终态已完成', '已完成' in head3, head3[:80]) fails = pg.locator('#cimpFailTbody tr').count() step('失败明细 1 行', fails == 1, fails) imp_db = dbq("SELECT name FROM customer_contact WHERE customer_id=%s AND name LIKE %s", (int(CID), f'{PFX}{TS}导入甲')) step('DB 执行面: 导入甲已落库', len(imp_db) == 1, f'db={len(imp_db)}') pg.screenshot(path=os.path.join(SHOTS, '07-cimp.png'), full_page=True) pg.click('#cimpStep3 button.btn-primary') pg.wait_for_timeout(400) # 13 export / delContact / 页签互斥 pg.click('#contactsShell button[onclick="exportContacts()"]') pg.wait_for_selector(".toast-item:has-text('导出已提交')", timeout=8000) step('export POST 已发出', any('POST' in u and '/api/customer/contact/export' in u for u in api_urls), '') before = pg.locator('#contactTbody tr').count() pg.click(f'#contactTbody tr[data-id="{new_id}"] [data-op="del"]') pg.wait_for_selector('dialog#dlg[open]', timeout=8000) dlg_title = (pg.text_content('#dlgTitle') or '').strip() pg.click('.dfoot button.ok') pg.wait_for_selector(".toast-item:has-text('已删除')", timeout=10000) pg.wait_for_timeout(1000) after = pg.locator('#contactTbody tr').count() step('delContact gConfirm 弹窗 + 双刷', dlg_title == '删除联系人' and before == 6 and after == 5, f'{before}→{after} dlg={dlg_title}') pg.screenshot(path=os.path.join(SHOTS, '08-deleted.png'), full_page=True) pg.click("#detailPanel .subtabs button[data-s='follow']") pg.wait_for_timeout(600) shell_hidden = pg.evaluate("document.getElementById('contactsShell').style.display === 'none'") sub_visible = pg.evaluate("document.getElementById('subBox').style.display !== 'none'") step('切走 follow: shell 隐藏 subBox 显示', shell_hidden and sub_visible, f'shell={shell_hidden} sub={sub_visible}') b.close() # ---- 网络层断言 ---- step('console/pageerror 零', not errors, '; '.join(errors[:3])) bad_method = [u for u in api_urls if u.startswith(('PUT ', 'DELETE '))] step('零 PUT/DELETE', not bad_method, str(bad_method[:3])) twelve = ['contact/graph/detail', 'contact/graph/save', 'contact/reveal', 'contact/quickAdd', 'contact/batchEdit', 'contact/export', 'contact/import/template', 'contact/import/upload', 'contact/import/confirm', 'contact/import/result', 'contact/import/failures', '/contacts?'] hit = sum(1 for ep in twelve if any(ep in u for u in api_urls)) step('12 端点族全部被触发（本冒烟可及 11 + contacts）', hit >= 11, f'{hit}/{len(twelve)}') sweep2 = sweep() print(f"清扫: e2c-t04* 夹具硬删 {sweep2} 条（幂等正门）") step('DB 正门清扫完成', sweep2 >= 1, f'removed={sweep2}') print('\n==== summary ====%d steps, %d failed' % (len(steps), sum(1 for s in steps if not s['ok']))) with open(os.path.join(ROOT, 't04-smoke-result.json'), 'w', encoding='utf-8') as f: json.dump({'steps': steps, 'errors': errors, 'api_urls': api_urls}, f, ensure_ascii=False, indent=1) print('ALL_PASS:', all(s['ok'] for s in steps))