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

#!/usr/bin/env python3
"""Measure first sidebar navigation to each operations module after the dashboard settles.

Runs the production bundles against a local, fixed gateway stub. This measures
route-code availability and first page heading; it is not a live API benchmark.
"""

import argparse
import asyncio
import json
import statistics
import sys
import threading
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,
}
MODULES = {
    "stores": "门店管理", "schedule": "排课中心", "coaches": "陪练老师",
    "payroll": "工资结算", "vocabulary": "词库管理", "special-training": "专项训练",
}


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):
    return round(statistics.quantiles(sorted(values), n=100, method="inclusive")[74], 1)


async def measure(browser, base_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()
    cdp = await context.new_cdp_session(page)
    await cdp.send("Network.enable")
    await cdp.send("Network.setCacheDisabled", {"cacheDisabled": False})
    await cdp.send("Network.emulateNetworkConditions", {
        "offline": False, "latency": 20,
        "downloadThroughput": 4 * 1024 * 1024 / 8,
        "uploadThroughput": 3 * 1024 * 1024 / 8,
    })

    async def mock_gateway(route):
        await asyncio.sleep(0.15)
        await route.fulfill(status=200, content_type="application/json", body=json.dumps({"success": True, "data": {}}))

    await page.route("**/cloud-functions/**", mock_gateway)
    try:
        await page.goto(base_url + "/admin/dashboard", wait_until="commit")
        await page.locator('#admin-sidebar a[href="/admin/stores"]').wait_for(timeout=30000)
        await page.wait_for_timeout(4500)
        results = {}
        for route, title in MODULES.items():
            before = await page.evaluate("performance.now()")
            await page.locator(f'#admin-sidebar a[href="/admin/{route}"]').click()
            await page.get_by_role("heading", name=title, exact=True).first.wait_for(timeout=30000)
            after = await page.evaluate("performance.now()")
            results[route] = round(after - before, 1)
        return results
    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, help="JSON map of maximum candidate p75 milliseconds by route")
    args = parser.parse_args()
    servers = {"A": serve(args.baseline), "B": serve(args.candidate)}
    urls = {name: f"http://127.0.0.1:{server.server_port}" for name, server in servers.items()}
    runs = {name: {route: [] for route in MODULES} for name in servers}
    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")):
                        result = await measure(browser, urls[name])
                        for route, value in result.items():
                            runs[name][route].append(value)
            finally:
                await browser.close()
    finally:
        for server in servers.values():
            server.shutdown()
    summary = {route: {name: {"p75Ms": p75(runs[name][route]), "runs": runs[name][route]}
                       for name in servers} for route in MODULES}
    print(json.dumps({route: {name: entry["p75Ms"] for name, entry in results.items()}
                      for route, results in summary.items()}, ensure_ascii=False, indent=2))
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    if args.ceilings:
        ceilings = json.loads(args.ceilings.read_text(encoding="utf-8"))
        failures = [f"{route}: {summary[route]['B']['p75Ms']} > {maximum} ms"
                    for route, maximum in ceilings.items() if summary[route]["B"]["p75Ms"] > maximum]
        if failures:
            print("Menu-switch ceiling exceeded: " + "; ".join(failures), file=sys.stderr)
            raise SystemExit(1)


if __name__ == "__main__":
    asyncio.run(main())