# -*- coding: utf-8 -*- """a7-heavy · demo 浏览器级 E2E（票 05：补码三域冒烟，Playwright chromium headless） 被测: .scratch/a7-heavy/demo/index.html?api=http://localhost:8080（ts= 注入脚本侧时间戳，两侧一致） 布局: 3 tab —— 同步任务(票01) / 全局日志(票02) / 公海绑定数(票03) 覆盖: 票01 任务化契约(SYNC-MMdd-序号/61023 两态容忍/分页倒序/详情五块/设置读写+61022 双路+还原/ lastSyncTime 切表口径/下次执行时间推导) + 票02 AOP 采集行/退出埋点/导出导入留档/结果文件 ZIP/ ids[] 打包/不存在文件错误形态/BUDDY 403 + 票03 部门专用规则发布/增量计数 base+n/默认兜底降级/清理正门 附带断言: 网络层零 PUT/DELETE、零 /api/// 路径变量、console/page error 零 清理: pre_clean 幂等（演示角色删 / 规则草稿删+发布中停用 / 导入池删）； 任务行与登录/操作/批量日志行 = 审计数据，留存不删；同步设置由 demo 步骤7 自还原原值； 商机 close 正门终态（已关闭仍计数 → 增量断言 base+n 跨轮安全） """ import json, re, time, pathlib, sys, io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') from playwright.sync_api import sync_playwright ROOT = pathlib.Path(__file__).resolve().parent SHOTS = ROOT / "shots"; SHOTS.mkdir(exist_ok=True) DEMO = ROOT / "demo" / "index.html" API = "http://localhost:8080" ADMIN = "739564171091247104" BUDDY = "744842318024015872" TS = time.strftime("%H%M%S") results = [] def step(name, ok, note=""): results.append({"step": name, "ok": bool(ok), "note": note}) print(("PASS " if ok else "FAIL ") + name + (" | " + note if note else "")) import urllib.request, urllib.parse, urllib.error def post(url, data=None, token=None): body = urllib.parse.urlencode(data or {}).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/x-www-form-urlencoded;charset=UTF-8") if token: req.add_header("Authorization", "Bearer " + token) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: return {"code": -1, "message": f"HTTP {e.code}"} def get(url, token=None): req = urllib.request.Request(url) if token: req.add_header("Authorization", "Bearer " + token) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: return {"code": -1, "message": f"HTTP {e.code}"} def token_of(uid=ADMIN): return json.loads(urllib.request.urlopen(f"{API}/api/auth/debug/token?userId={uid}", timeout=15).read().decode())["data"] def pre_clean(): """中断轮次残留清理（幂等重跑保障）。 留守原则：任务行与登录/操作/批量日志行 = 审计数据，留存不删；设置由 demo 步骤7 还原。""" try: tok = token_of() except Exception as ex: print("pre_clean token fail:", ex); return removed = {"roles": 0, "rules_deleted": 0, "rules_disabled": 0, "pools": 0} # 演示角色（logPrepare 正常轮次自建自删，此处兜中断残留） for kw in ("e2d-日志演示角色",): pg = post(f"{API}/api/roles/page", {"current": 1, "size": 100, "roleName": kw}, tok) for r in (pg.get("data") or {}).get("content", []): if post(f"{API}/api/roles/delete", {"id": str(r["id"])}, tok).get("code") == 0: removed["roles"] += 1 # 绑定数演示规则：草稿删 / 发布中停用（停用行留痕可辨，同 demo prPrepare 幂等口径） pg = post(f"{API}/api/rule/opp-pool-rule/page", {"current": 1, "size": 100, "ruleName": "e2d-绑定数规则"}, tok) for r in (pg.get("data") or {}).get("content", []): if not str(r.get("ruleName", "")).startswith("e2d-绑定数规则"): continue if str(r.get("status")) == "1" and post(f"{API}/api/rule/opp-pool-rule/delete", {"id": str(r["id"])}, tok).get("code") == 0: removed["rules_deleted"] += 1 elif str(r.get("status")) == "2" and post(f"{API}/api/rule/opp-pool-rule/disable", {"id": str(r["id"])}, tok).get("code") == 0: removed["rules_disabled"] += 1 # 导入池（e2d-导入池；上轮已 close 不存在——池本体删除即净） pg = post(f"{API}/api/rule/pool/page", {"current": 1, "size": 100, "poolName": "e2d-导入池"}, tok) for r in (pg.get("data") or {}).get("content", []): if str(r.get("poolName", "")).startswith("e2d-导入池") and post(f"{API}/api/rule/pool/delete", {"id": str(r["id"])}, tok).get("code") == 0: removed["pools"] += 1 print("pre_clean:", removed) pre_clean() console_errors, page_errors, api_urls = [], [], [] def run_step(page, tab, label=None, btn=None, outsel=None, timeout=30000): """切到 tab → 清空旧输出防串步 → 点步骤按钮 → 等 out 容器出现 okbox/errbox/infobox""" outsel = outsel or f"#{tab}Out" page.click(f"nav.tabs button[data-tab='{tab}']") page.evaluate(f"document.querySelector('{outsel}').innerHTML = ''") page.click(btn or f"#tab-{tab} button:has-text('{label}')") page.wait_for_selector(f"{outsel} .okbox, {outsel} .errbox, {outsel} .infobox", timeout=timeout) cls = page.get_attribute(f"{outsel} > div", "class") text = page.text_content(f"{outsel} > div").strip() is_ok = "okbox" in cls or "infobox" in cls return is_ok, text def free_depts(tok): """树序部门 id 中未被线索池占用的（导入池 xlsx 用 free[0]；demo 无建池步骤，无错开需求）""" tree = get(f"{API}/api/system/depts/tree", tok).get("data") or [] ids = [] def walk(ns): for n in ns or []: ids.append(str(n["id"])); walk(n.get("children")) walk(tree) pg = post(f"{API}/api/rule/pool/page", {"current": 1, "size": 200}, tok) used = {str(r["deptId"]) for r in (pg.get("data") or {}).get("content", []) if r.get("deptId")} return [i for i in ids if i not in used] def make_import_xlsx(ts, dept_id): """12 列模板与 LeadPoolImportRow @ExcelProperty 逐字一致；1 行合法数据（部门=空闲部门顺位避让 64003）""" import openpyxl wb = openpyxl.Workbook(); ws = wb.active ws.append(["池名称", "归属部门ID", "部门名称", "负责人用户ID", "负责人姓名", "协作人用户ID（逗号分隔）", "部门成员用户ID（逗号分隔）", "领取规则(1=成员可领+管理员分配 2=仅管理员分配)", "每日领取上限", "个人持有上限", "超时回收天数", "线索失效天数"]) ws.append([f"e2d-导入池-{ts}", dept_id, "e2d部门", ADMIN, "罗伟健", "", BUDDY, 1, 5, 100, 7, 30]) p = ROOT / f"_log_import_{ts}.xlsx" wb.save(p) return p with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.set_default_timeout(20000) page.on("console", lambda m: console_errors.append(m.text) if m.type == "error" else None) page.on("pageerror", lambda e: page_errors.append(str(e))) def on_resp(r): u = r.url if "/api/" in u: api_urls.append(f"{r.request.method} {u}") page.on("response", on_resp) # 0. 打开 demo（ts 参数注入脚本侧时间戳，保证演示角色/规则/商机/导入池名两侧一致） page.goto(DEMO.as_uri() + "?api=" + API + "&ts=" + TS) page.wait_for_load_state("networkidle") page.screenshot(path=str(SHOTS / "00-open.png")) # 1. 登录（登录后 demo 自动预载同步任务列表） page.click("button:has-text('uid 登录')") page.wait_for_selector(".toast-item:has-text('登录成功')", timeout=15000) step("登录 debug/token + /auth/me", True) # ===== Tab 同步任务（票 01）===== # 步骤1 发起同步（61023 两态容忍；异步终态轮询最长 75s → 步骤超时放大到 100s） is_ok, text = run_step(page, "sync", "发起同步任务", timeout=100000) step("同步任务 launch（任务编号格式 + 61023 两态容忍）", is_ok and ("任务化契约验证" in text or "61023 并发占位生效" in text), text[:100].replace("\n", " ")) page.screenshot(path=str(SHOTS / "01-sync-launch.png")) # 步骤2 分页+倒序（syncPage 渲染 #sBody 不走 out 容器，单独驱动；window._syncRows 服务端原序断言倒序） page.click("nav.tabs button[data-tab='sync']") page.click("#tab-sync .viewbar button:has-text('查询')") page.wait_for_selector("#sBody tr td:not(.empty)", state="attached", timeout=15000) rows = page.evaluate("() => window._syncRows || []") times = [str(r.get("startTime") or "") for r in rows] desc = all(times[i] >= times[i + 1] for i in range(len(times) - 1)) step("同步记录分页 + 开始时间倒序", len(rows) > 0 and desc, f"rows={len(rows)} desc={desc}" + (f" first={times[0]}" if times else "")) # 步骤3 详情五块（taskNo/triggerType/status/startTime/progress + failDetails 三字段） is_ok, text = run_step(page, "sync", "任务详情") step("任务详情五块结构（含 failDetails 失败明细）", is_ok and "详情五块结构在位" in text, text[:120].replace("\n", " ")) # 步骤4-7 设置读/存/边界/还原 is_ok, text = run_step(page, "sync", "读取同步设置") step("同步设置回显（派生 plan/nextExecuteTime/lastModifierName）", is_ok and "设置回显" in text, text[:120].replace("\n", " ")) page.screenshot(path=str(SHOTS / "02-sync-settings.png")) is_ok, text = run_step(page, "sync", "保存合法设置") step("设置保存回显（interval=2 次数=3 并发=1）", is_ok and "合法设置保存并回显" in text, text[:90]) is_ok, text = run_step(page, "sync", "边界校验") step("设置边界 61022 双路（间隔<1min / 次数<1 拒）", is_ok and "61022 边界校验两路生效" in text, text[:100].replace("\n", " ")) is_ok, text = run_step(page, "sync", "还原设置") step("设置还原（不污染调度原值）", is_ok and "设置已还原" in text, text[:90]) # 步骤8 lastSyncTime 切表口径 + 下次执行时间推导 is_ok, text = run_step(page, "sync", "lastSyncTime 数据源切换") step("lastSyncTime 切表口径 + 下次执行时间推导", is_ok and "lastSyncTime 验证" in text, text[:130].replace("\n", " ")) page.screenshot(path=str(SHOTS / "03-sync-stats.png")) # ===== Tab 全局日志（票 02）===== page.click("nav.tabs button[data-tab='glog']") _tok = token_of() _free = free_depts(_tok) assert len(_free) >= 1, f"空闲部门不足: {_free}" xlsx = make_import_xlsx(TS, _free[0]) page.set_input_files("#logImportFile", str(xlsx)) # 步骤1 AOP 素材（建/删演示角色；采集落库由步骤2 轮询等待） is_ok, text = run_step(page, "glog", "建/删演示角色") step("AOP 写端点采集素材（建/删演示角色）", is_ok and "写操作完成" in text, text[:100]) page.screenshot(path=str(SHOTS / "04-glog-prepare.png")) # 步骤2 操作日志分页（operatorName/opType 命中 + 时间范围逐字段） is_ok, text = run_step(page, "glog", "操作日志分页", timeout=40000) step("操作日志分页（新增/删除行命中 + 时间范围逐字段）", is_ok and "AOP 采集行命中" in text, text[:130].replace("\n", " ")) # 步骤3 退出埋点（debug token logout 落「退出」行 + token 重铸） is_ok, text = run_step(page, "glog", "退出登录", timeout=40000) step("登录日志「退出」行（内联埋点 + token 重铸）", is_ok and "退出埋点命中" in text, text[:100].replace("\n", " ")) # 步骤4 线索池导出 → 批量「导出」行（result + 结果文件 id + ZIP 魔数） is_ok, text = run_step(page, "glog", "线索池导出", timeout=40000) step("批量日志「导出」行（result+fileId+ZIP 魔数）", is_ok and "导出留档行命中" in text, text[:120].replace("\n", " ")) page.screenshot(path=str(SHOTS / "05-glog-export.png")) # 步骤5 导入 xlsx → 批量「导入」行（BatchResult 计数） is_ok, text = run_step(page, "glog", "导入上方文件", timeout=40000) step("批量日志「导入」行（BatchResult 计数）", is_ok and "导入行命中" in text, text[:120].replace("\n", " ")) # 步骤6 结果文件下载（/api/log/batch/file?id= ZIP 魔数） is_ok, text = run_step(page, "glog", "结果文件下载") step("结果文件下载 /api/log/batch/file（ZIP 魔数）", is_ok and "结果文件下载成功" in text, text[:90]) # 步骤7 打包下载（/api/log/batch/download ids[] ZIP 魔数） is_ok, text = run_step(page, "glog", "打包下载") step("打包下载 /api/log/batch/download ids[]（ZIP 魔数）", is_ok and "打包下载成功" in text, text[:90]) # 步骤8 不存在文件 id 错误形态（40401/66001/HTTP404 容忍） is_ok, text = run_step(page, "glog", "不存在文件 id") step("不存在文件 id 错误形态（40401/66001/404）", is_ok and "错误形态生效" in text, text[:100].replace("\n", " ")) # 步骤9 BUDDY 三 page 权限拒绝 is_ok, text = run_step(page, "glog", "非管理员", timeout=40000) step("BUDDY 三 page 权限拒绝（ADMIN 门禁）", is_ok and "非管理员权限口径生效" in text, text[:130].replace("\n", " ")) page.screenshot(path=str(SHOTS / "06-glog-denied.png")) # ===== Tab 公海绑定数（票 03）===== # 步骤1 建演示部门专用规则并发布（部门被占回退既有发布中专用规则）→ 基线 base is_ok, text = run_step(page, "poolrule", "建演示部门专用规则", timeout=60000) step("公海规则发布（部门专用）+ boundOppCount 基线", is_ok and "规则就绪" in text, text[:130].replace("\n", " ")) page.screenshot(path=str(SHOTS / "07-poolrule-base.png")) # 步骤2/3 建演示商机 A/B → 增量断言 base+1 / base+2 is_ok, text = run_step(page, "poolrule", "建演示商机 A", timeout=60000) step("商机 A → boundOppCount=base+1（match 语义）", is_ok and "base+1" in text, text[:110]) is_ok, text = run_step(page, "poolrule", "建演示商机 B", timeout=60000) step("商机 B → boundOppCount=base+2（动态统计非快照）", is_ok and "base+2" in text, text[:110]) # 步骤4 默认规则兜底口径（无发布中默认规则 → infobox 降级提示） is_ok, text = run_step(page, "poolrule", "默认规则兜底口径", timeout=60000) step("默认规则兜底口径（无默认规则降级提示）", is_ok and ("兜底口径生效" in text or "降级为提示" in text), text[:120].replace("\n", " ")) # 步骤5 清理（商机 close 正门终态 / 演示规则停用） is_ok, text = run_step(page, "poolrule", "关停演示商机", timeout=60000) step("清理（商机 close 正门 + 规则停用）", is_ok and "清理完成" in text, text[:120].replace("\n", " ")) page.screenshot(path=str(SHOTS / "08-cleanup.png")) # N. 网络层断言：零 PUT/DELETE、零路径变量形态 bad_method = [u for u in api_urls if u.startswith(("PUT ", "DELETE "))] old_style = [u for u in api_urls if re.search(r"/api/[a-z-]+/\d+/", u)] step("网络层零旧契约形态", not bad_method and not old_style, f"bad_method={bad_method[:2]} old_style={old_style[:2]} (共{len(api_urls)}请求)") browser.close() # ---- 汇总 ---- xlsx.unlink(missing_ok=True) # BUDDY 403 权限拒绝步骤会产生浏览器级 "Failed to load resource: 403" console error # （HTTP 层 4xx 响应浏览器必然记录，JS 无法阻止）——属预期拒绝产物，豁免后仍要求真实 console error 为零 expected_403 = [t for t in console_errors if "status of 403" in t] real_console = [t for t in console_errors if "status of 403" not in t] step("console/page error 双零（预期 403 拒绝豁免）", not real_console and not page_errors, f"raw_console={len(console_errors)} exempt_403={len(expected_403)} real={len(real_console)} page_errors={len(page_errors)}") summary = {"ts": TS, "console_errors": console_errors, "page_errors": page_errors, "exempt_403": expected_403, "real_console_errors": real_console, "results": results, "all_pass": all(r["ok"] for r in results) and not real_console and not page_errors} (ROOT / "demo-smoke-result.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1), encoding="utf-8") print("\n== SUMMARY ==") for r in results: print(("PASS " if r["ok"] else "FAIL ") + r["step"] + (" | " + r["note"] if r["note"] else "")) print("console_errors:", len(console_errors), "(exempt_403:", len(expected_403), ") page_errors:", len(page_errors)) print("ALL_PASS:", summary["all_pass"])