uce/tests/wasm_benchmark.py

#!/usr/bin/env python3
"""Current wasm-runtime HTTP benchmark harness.

This runner intentionally has no third-party dependencies. It measures one or
more HTTP endpoints served by the wasm-only UCE runtime. Optionally pass a prior
benchmark JSON as a baseline; old native baselines are treated as historical
reference data only, not as a runnable backend mode.
"""
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], baseline_label: str) -> list[str]:
    baseline = {r.target: r for r in results if r.backend == baseline_label}
    lines = ["| target | backend | median ms | mean ms | budget | status |", "|---|---:|---:|---:|---|---|"]
    for result in results:
        budget = "baseline" if result.backend == baseline_label else "n/a"
        status = "PASS" if result.ok else "FAIL"
        if result.backend != baseline_label and result.target in baseline:
            limit = baseline[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="UCE wasm-runtime HTTP benchmark harness")
    parser.add_argument("--base-url", "--native-base-url", dest="base_url", default="http://localhost:80", help="runtime base URL (legacy alias --native-base-url is accepted)")
    parser.add_argument("--backend-label", default="wasm", help="label for this run's measurements")
    parser.add_argument("--compare-baseline-json", "--compare-native-json", dest="compare_baseline_json", default="", help="optional prior benchmark.json for budget comparison")
    parser.add_argument("--baseline-label", default="native", help="backend label in the prior JSON to use as the budget baseline")
    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-benchmark")
    args = parser.parse_args()

    targets = build_targets()
    results: list[Measurement] = []
    if args.compare_baseline_json:
        for row in json.loads(Path(args.compare_baseline_json).read_text()):
            row.setdefault("backend", args.baseline_label)
            results.append(Measurement(**row))
    results.extend(measure_backend(args.backend_label, args.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 = ["# UCE wasm benchmark report", "", *compare(results, args.baseline_label), ""]
    (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())