#!/usr/bin/env python3 """Paired measurement of returning to each already opened operations page. The gateway is fixed and synthetic. We time the visible heading and count business data requests before and after return, excluding lightweight version checks. This is a navigation and redundant-request test, not a live-data test. """ import argparse import asyncio import json import statistics import sys import threading from collections import Counter from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlsplit from playwright.async_api import async_playwright SESSION = {"objectId": "perf-test", "username": "perf-test", "sessionToken": "local-perf-test", "displayName": "测速账号", "roles": [], "operationsRole": "ops-manager", "isSuperAdmin": False} DASHBOARD = { "identity": {"objectId": "perf-test", "username": "perf-test", "displayName": "运营经理", "roles": [], "operationsRole": "ops-manager", "isSuperAdmin": False}, "date": "2026-09-26", "metrics": [], "alerts": [], "refreshedAt": "2026-09-26T06:00:00.000Z", "liveSourceAvailable": True, "scheduleTrend": {"period": "week", "periodLabel": "近 7 天", "dateFrom": "2026-09-20", "dateTo": "2026-09-26", "points": [{"key": "2026-09-26", "label": "9月26日", "shortLabel": "周六", "courseCount": 2, "studentCount": 2, "completedCount": 1, "isToday": True}], "todayCourseCount": 2, "todayStudentCount": 2, "totalCourseCount": 2, "totalStudentCount": 2, "maxValue": 2, "refreshedAt": "2026-09-26T06:00:00.000Z", "liveSourceAvailable": True}, } MODULES = { "dashboard": "运营经理，欢迎回来", "members": "会员管理", "stores": "门店管理", "schedule": "排课中心", "learning-reports": "学习报表", "coaches": "陪练老师", "payroll": "工资结算", "vocabulary": "词库管理", "special-training": "专项训练", } SELECTORS = {module: ("app-admin-dashboard" if module == "dashboard" else f"app-operations-{module}") for module in MODULES} SELECTORS["members"] = "app-operations-users" class AppHandler(SimpleHTTPRequestHandler): def do_GET(self): if not (Path(self.directory) / urlsplit(self.path).path.lstrip("/")).is_file(): self.path = "/index.html" return super().do_GET() def log_message(self, *_args): pass def serve(directory): server = ThreadingHTTPServer(("127.0.0.1", 0), partial(AppHandler, directory=str(directory))) threading.Thread(target=server.serve_forever, daemon=True).start() return server def p75(values): if len(values) == 1: return values[0] return round(statistics.quantiles(sorted(values), n=100, method="inclusive")[74], 1) async def measure(browser, url): context = await browser.new_context(viewport={"width": 1440, "height": 900}) await context.add_init_script( "sessionStorage.setItem('xiaoshu_admin_session_v1', " + json.dumps(json.dumps(SESSION)) + ");" ) page = await context.new_page() counts = Counter() errors = [] page.on("pageerror", lambda error: errors.append(str(error))) async def mock_gateway(route): operation = (route.request.post_data_json or {}).get("params", {}).get("operation", "") if operation not in ("meta", "ops/live/versions"): counts[operation] += 1 await asyncio.sleep(0.12 if operation != "ops/dashboard/summary" else 0.7) if operation == "meta": payload = {"identity": DASHBOARD["identity"]} elif operation == "ops/dashboard/summary": payload = DASHBOARD elif operation == "ops/live/versions": payload = {"version": "fixed-v1", "scopeVersions": {}, "refreshedAt": "2026-09-26T06:00:00.000Z", "syncStatus": {"state": "current"}} else: payload = {"items": [], "total": 0, "page": 1, "pageSize": 20} await route.fulfill(status=200, content_type="application/json", body=json.dumps({"success": True, "data": payload})) await page.route("**/cloud-functions/**", mock_gateway) try: await page.goto(url + "/admin/dashboard", wait_until="commit") await page.get_by_role("heading", name=MODULES["dashboard"]).wait_for(timeout=30000) await page.locator(".trend-stat.courses strong").wait_for(timeout=30000) results = {} for module, title in MODULES.items(): target = page.locator(f'#admin-sidebar a[href="/admin/{module}"]').last first_counts = counts.copy() if module != "dashboard": await target.click() await page.locator(SELECTORS[module]).wait_for(timeout=30000) await page.wait_for_timeout(220) target_operations = set((counts - first_counts).keys()) if module != "dashboard" else {"ops/dashboard/summary"} await page.locator('#admin-sidebar a[href="/admin/dashboard"]').last.click() await page.locator(SELECTORS["dashboard"]).wait_for(timeout=30000) if module == "dashboard": await page.locator('#admin-sidebar a[href="/admin/schedule"]').last.click() await page.locator(SELECTORS["schedule"]).wait_for(timeout=30000) before_counts = counts.copy() start = await page.evaluate("performance.now()") await target.click() await page.locator(SELECTORS[module]).wait_for(timeout=30000) if module == "dashboard": await page.locator(".trend-stat.courses strong").wait_for(timeout=30000) elapsed = round(await page.evaluate("performance.now()") - start, 1) await page.wait_for_timeout(220) results[module] = {"returnMs": elapsed, "repeatBusinessRequests": sum((counts - before_counts)[operation] for operation in target_operations)} return {"modules": results, "pageErrors": errors} finally: await context.close() async def main(): parser = argparse.ArgumentParser() parser.add_argument("--baseline", type=Path, required=True) parser.add_argument("--candidate", type=Path, required=True) parser.add_argument("--runs", type=int, default=10) parser.add_argument("--out", type=Path) parser.add_argument("--ceilings", type=Path) args = parser.parse_args() roots = {"A": args.baseline, "B": args.candidate} servers = {name: serve(path) for name, path in roots.items()} results = {name: [] for name in roots} try: async with async_playwright() as playwright: browser = await playwright.chromium.launch() try: for index in range(args.runs): for name in (("A", "B") if index % 2 == 0 else ("B", "A")): results[name].append(await measure(browser, f"http://127.0.0.1:{servers[name].server_port}")) finally: await browser.close() finally: for server in servers.values(): server.shutdown() summary = {module: {name: { "p75Ms": p75([run["modules"][module]["returnMs"] for run in results[name]]), "repeatBusinessRequests": max(run["modules"][module]["repeatBusinessRequests"] for run in results[name]), } for name in roots} for module in MODULES} print(json.dumps(summary, ensure_ascii=False, indent=2)) failures = [f"{module}: {entry['B']}" for module, entry in summary.items() if entry["B"]["repeatBusinessRequests"] != 0] if any(run["pageErrors"] for run in results["B"]): failures.append("candidate browser page errors") if args.ceilings: limits = json.loads(args.ceilings.read_text(encoding="utf-8")) failures.extend(f"{module}: {summary[module]['B']['p75Ms']} > {maximum} ms" for module, maximum in limits.items() if summary[module]["B"]["p75Ms"] > maximum) if args.out: args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps({"summary": summary, "runs": results}, ensure_ascii=False, indent=2), encoding="utf-8") if failures: print("Return-navigation ceiling exceeded: " + "; ".join(failures), file=sys.stderr) raise SystemExit(1) if __name__ == "__main__": asyncio.run(main())