wasm-simulator/web/test-browser-wasm-smoke-cdp.py

#!/usr/bin/env python3
import base64
import http.client
import json
import os
import socket
import struct
import sys
import time
from urllib.parse import urlparse

debug_port = int(sys.argv[1])
dom_log = sys.argv[2]
timeout_ms = int(sys.argv[3])
deadline = time.monotonic() + (timeout_ms / 1000.0)


def http_json(path):
    conn = http.client.HTTPConnection("127.0.0.1", debug_port, timeout=1)
    try:
        conn.request("GET", path)
        response = conn.getresponse()
        body = response.read()
    finally:
        conn.close()
    if response.status != 200:
        raise RuntimeError(f"{path} returned HTTP {response.status}")
    return json.loads(body.decode("utf-8"))


def websocket_connect(url):
    parsed = urlparse(url)
    key = base64.b64encode(os.urandom(16)).decode("ascii")
    sock = socket.create_connection((parsed.hostname, parsed.port), timeout=2)
    request = (
        f"GET {parsed.path} HTTP/1.1\r\n"
        f"Host: {parsed.hostname}:{parsed.port}\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        f"Sec-WebSocket-Key: {key}\r\n"
        "Sec-WebSocket-Version: 13\r\n\r\n"
    )
    sock.sendall(request.encode("ascii"))
    response = sock.recv(4096)
    if b" 101 " not in response.split(b"\r\n", 1)[0]:
        raise RuntimeError("DevTools websocket handshake failed")
    sock.settimeout(1)
    return sock


def websocket_send(sock, payload):
    data = payload.encode("utf-8")
    header = bytearray([0x81])
    if len(data) < 126:
        header.append(0x80 | len(data))
    elif len(data) < 65536:
        header.append(0x80 | 126)
        header.extend(struct.pack("!H", len(data)))
    else:
        header.append(0x80 | 127)
        header.extend(struct.pack("!Q", len(data)))
    mask = os.urandom(4)
    header.extend(mask)
    masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(data))
    sock.sendall(header + masked)


def recv_exact(sock, size):
    chunks = []
    remaining = size
    while remaining:
        chunk = sock.recv(remaining)
        if not chunk:
            raise RuntimeError("DevTools websocket closed")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def websocket_recv(sock):
    first, second = recv_exact(sock, 2)
    opcode = first & 0x0F
    length = second & 0x7F
    if length == 126:
        length = struct.unpack("!H", recv_exact(sock, 2))[0]
    elif length == 127:
        length = struct.unpack("!Q", recv_exact(sock, 8))[0]
    masked = bool(second & 0x80)
    mask = recv_exact(sock, 4) if masked else b""
    payload = recv_exact(sock, length)
    if masked:
        payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
    if opcode == 8:
        raise RuntimeError("DevTools websocket closed")
    if opcode != 1:
        return None
    return json.loads(payload.decode("utf-8"))


def find_page_ws_url():
    while time.monotonic() < deadline:
        try:
            for target in http_json("/json/list"):
                if target.get("type") == "page" and target.get("url", "").endswith("/test-browser-wasm-smoke.html"):
                    return target["webSocketDebuggerUrl"]
        except Exception:
            pass
        time.sleep(0.1)
    raise RuntimeError("timed out waiting for browser DevTools page target")


sock = websocket_connect(find_page_ws_url())
next_id = 0


def cdp(method, params=None):
    global next_id
    next_id += 1
    message_id = next_id
    websocket_send(sock, json.dumps({"id": message_id, "method": method, "params": params or {}}))
    while time.monotonic() < deadline:
        try:
            message = websocket_recv(sock)
        except TimeoutError:
            continue
        except socket.timeout:
            continue
        if message and message.get("id") == message_id:
            if "error" in message:
                raise RuntimeError(message["error"])
            return message["result"]
    raise RuntimeError(f"timed out waiting for DevTools response to {method}")


def evaluate(expression):
    result = cdp("Runtime.evaluate", {"expression": expression, "returnByValue": True})
    return result.get("result", {}).get("value", "")


try:
    cdp("Runtime.enable")
    text = ""
    while time.monotonic() < deadline:
        text = evaluate('document.querySelector("#result")?.textContent || ""')
        if "browser wasm smoke passed" in text:
            html = evaluate("document.documentElement.outerHTML")
            with open(dom_log, "w", encoding="utf-8") as handle:
                handle.write(html)
            sys.exit(0)
        if "browser wasm smoke failed:" in text:
            print(text, file=sys.stderr)
            html = evaluate("document.documentElement.outerHTML")
            with open(dom_log, "w", encoding="utf-8") as handle:
                handle.write(html)
            sys.exit(1)
        time.sleep(0.2)
    print(f"timed out waiting for browser smoke result; last status: {text}", file=sys.stderr)
    html = evaluate("document.documentElement.outerHTML")
    with open(dom_log, "w", encoding="utf-8") as handle:
        handle.write(html)
    sys.exit(1)
finally:
    sock.close()