xiaoshu-app/perf/bench/admin-menu-return.py

#!/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())