uce/tests/wasm_benchmark.py

#!/usr/bin/env python3
"""Phase 5 native/WASM benchmark harness.

This runner intentionally has no third-party dependencies. It records a warmed
native HTTP baseline now and can compare against a future wasm worker by passing
--wasm-base-url. Results are written as JSON plus a small Markdown table.
"""
from __future__ import annotations

import argparse
import http.client
import json
import statistics
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from urllib.parse import urlparse


@dataclass
class Target:
    name: str
    url: str
    expected: str


@dataclass
class Measurement:
    backend: str
    target: str
    url: str
    ok: bool
    status: int
    samples_ms: list[float]
    median_ms: float
    mean_ms: float
    min_ms: float
    max_ms: float
    note: str = ""


def request_once(base_url: str, host_header: str, path: str, timeout: float) -> tuple[int, bytes, float]:
    parsed = urlparse(base_url)
    if parsed.scheme not in ("http", "https") or not parsed.hostname:
        raise SystemExit(f"invalid base url: {base_url}")
    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    conn_cls = http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection
    headers = {}
    if host_header:
        headers["Host"] = host_header
    started = time.perf_counter()
    conn = conn_cls(parsed.hostname, port, timeout=timeout)
    try:
        conn.request("GET", path, headers=headers)
        response = conn.getresponse()
        body = response.read()
        elapsed_ms = (time.perf_counter() - started) * 1000.0
        return response.status, body, elapsed_ms
    finally:
        conn.close()


def measure_backend(backend: str, base_url: str, host_header: str, targets: list[Target], warmups: int, samples: int, timeout: float) -> list[Measurement]:
    results: list[Measurement] = []
    for target in targets:
        for _ in range(warmups):
            request_once(base_url, host_header, target.url, timeout)
        durations: list[float] = []
        status = 0
        ok = True
        note = ""
        for _ in range(samples):
            status, body, elapsed_ms = request_once(base_url, host_header, target.url, timeout)
            durations.append(elapsed_ms)
            text = body.decode("utf-8", errors="replace")
            if status != 200:
                ok = False
                note = f"HTTP {status}"
            elif target.expected and target.expected not in text:
                ok = False
                note = f"missing marker {target.expected!r}"
        results.append(Measurement(
            backend=backend,
            target=target.name,
            url=target.url,
            ok=ok,
            status=status,
            samples_ms=durations,
            median_ms=statistics.median(durations),
            mean_ms=statistics.fmean(durations),
            min_ms=min(durations),
            max_ms=max(durations),
            note=note,
        ))
    return results


def compare(results: list[Measurement]) -> list[str]:
    native = {r.target: r for r in results if r.backend == "native"}
    lines = ["| target | backend | median ms | mean ms | budget | status |", "|---|---:|---:|---:|---|---|"]
    for result in results:
        budget = "baseline"
        status = "PASS" if result.ok else "FAIL"
        if result.backend != "native" and result.target in native:
            limit = native[result.target].median_ms * 2.0
            budget = f"≤ {limit:.1f} ms"
            if result.median_ms > limit:
                status = "FAIL"
        lines.append(f"| {result.target} | {result.backend} | {result.median_ms:.1f} | {result.mean_ms:.1f} | {budget} | {status} {result.note} |")
    return lines


def build_targets() -> list[Target]:
    return [
        Target("template-heavy-doc", "/doc/singlepage.uce", "UCE API"),
        Target("sqlite-page", "/demo/sqlite.uce", "SQLite"),
        Target("component-heavy-starter", "/examples/uce-starter/?dashboard", "Dashboard"),
    ]


def main() -> int:
    parser = argparse.ArgumentParser(description="Phase 5 native/wasm benchmark harness")
    parser.add_argument("--native-base-url", default="http://localhost:80")
    parser.add_argument("--wasm-base-url", default="", help="optional wasm worker URL; omitted until worker exists")
    parser.add_argument("--backend-label", default="native", help="label for --native-base-url measurements when running one backend at a time")
    parser.add_argument("--compare-native-json", default="", help="optional prior native benchmark.json for budget comparison")
    parser.add_argument("--host-header", default="uce.openfu.com")
    parser.add_argument("--warmups", type=int, default=2)
    parser.add_argument("--samples", type=int, default=20)
    parser.add_argument("--timeout", type=float, default=10.0)
    parser.add_argument("--out-dir", default="/tmp/uce/wasm-phase5")
    args = parser.parse_args()

    targets = build_targets()
    results: list[Measurement] = []
    if args.compare_native_json:
        for row in json.loads(Path(args.compare_native_json).read_text()):
            results.append(Measurement(**row))
    results.extend(measure_backend(args.backend_label, args.native_base_url, args.host_header, targets, args.warmups, args.samples, args.timeout))
    if args.wasm_base_url:
        results.extend(measure_backend("wasm", args.wasm_base_url, args.host_header, targets, args.warmups, args.samples, args.timeout))

    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "benchmark.json").write_text(json.dumps([asdict(r) for r in results], indent=2) + "\n")
    md_lines = ["# Phase 5 benchmark report", "", *compare(results), ""]
    if not args.wasm_base_url and args.backend_label == "native" and not args.compare_native_json:
        md_lines.append("WASM worker URL was not provided; this report is the native baseline that future wasm runs compare against.")
    (out_dir / "benchmark.md").write_text("\n".join(md_lines) + "\n")
    print("\n".join(md_lines))
    print(f"wrote {out_dir / 'benchmark.json'} and {out_dir / 'benchmark.md'}")
    return 0 if all(r.ok for r in results) else 1


if __name__ == "__main__":
    raise SystemExit(main())