site/tools/harness-cdp.py

#!/usr/bin/env python3
"""tools/harness-cdp.py CHROME URL [TIMEOUT_S] [CHROME_FLAG ...]

Runs a harness page in headless Chrome in real time and prints its #diag once it says END (or the timeout passes).
--dump-dom with --virtual-time-budget (the loop harness) is deterministic but IndexedDB answers in real time and
virtual time does not wait for it, so the resume harness is driven over the DevTools protocol instead: start Chrome
with a debugging port, attach to the page, poll Runtime.evaluate. Stdlib only (a 40-line websocket client)."""
import base64, json, os, re, socket, struct, subprocess, sys, tempfile, time, urllib.request


class WS:
    def __init__(self, url):
        m = re.match(r"ws://([^:/]+):(\d+)(/.*)", url)
        self.s = socket.create_connection((m.group(1), int(m.group(2))), timeout=10)
        key = base64.b64encode(os.urandom(16)).decode()
        self.s.sendall((f"GET {m.group(3)} HTTP/1.1\r\nHost: {m.group(1)}:{m.group(2)}\r\nUpgrade: websocket\r\n"
                        f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
        buf = b""
        while b"\r\n\r\n" not in buf: buf += self.s.recv(4096)
        assert b" 101 " in buf.split(b"\r\n")[0], buf[:200]
        self.buf = buf.split(b"\r\n\r\n", 1)[1]
        self.n = 0

    def _read(self, n):
        while len(self.buf) < n: self.buf += self.s.recv(65536)
        out, self.buf = self.buf[:n], self.buf[n:]
        return out

    def send(self, obj):
        data = json.dumps(obj).encode()
        head = bytes([0x81])
        if len(data) < 126: head += bytes([0x80 | len(data)])
        elif len(data) < 65536: head += bytes([0x80 | 126]) + struct.pack(">H", len(data))
        else: head += bytes([0x80 | 127]) + struct.pack(">Q", len(data))
        mask = os.urandom(4)
        self.s.sendall(head + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))

    def recv(self):
        while True:
            b0, b1 = self._read(2)
            n = b1 & 0x7F
            if n == 126: n = struct.unpack(">H", self._read(2))[0]
            elif n == 127: n = struct.unpack(">Q", self._read(8))[0]
            if b1 & 0x80: self._read(4)
            data = self._read(n)
            op = b0 & 0x0F
            if op == 1: return json.loads(data)
            if op == 8: raise EOFError("closed")

    def call(self, method, **params):
        self.n += 1
        self.send({"id": self.n, "method": method, "params": params})
        while True:
            m = self.recv()
            if m.get("id") == self.n: return m.get("result", {})


def main():
    chrome, url = sys.argv[1], sys.argv[2]
    timeout = float(sys.argv[3]) if len(sys.argv) > 3 else 30
    flags = sys.argv[4:]
    prof = tempfile.mkdtemp()
    p = subprocess.Popen([chrome, "--headless=new", "--disable-gpu", "--remote-debugging-port=0", f"--user-data-dir={prof}", *flags, "about:blank"],
                         stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
    port = None
    for line in p.stderr:
        m = re.search(r"DevTools listening on ws://127\.0\.0\.1:(\d+)/", line)
        if m: port = m.group(1); break
    if not port: print("no diag: Chrome gave no DevTools port"); p.kill(); return 1
    try:
        targets = json.load(urllib.request.urlopen(f"http://127.0.0.1:{port}/json"))
        page = next(t for t in targets if t["type"] == "page")
        ws = WS(page["webSocketDebuggerUrl"])
        ws.call("Page.enable"); ws.call("Runtime.enable")
        ws.call("Page.navigate", url=url)
        t0 = time.time(); text = ""
        while time.time() - t0 < timeout:
            time.sleep(0.25)
            r = ws.call("Runtime.evaluate", expression="(document.getElementById('diag')||{}).textContent||''", returnByValue=True)
            text = r.get("result", {}).get("value", "") or ""
            if "\nEND" in text or text.startswith("END") or text.startswith("TOPERR"): break
        print(text if text else "no diag: the harness did not run")
    finally:
        p.kill()
        subprocess.run(["rm", "-rf", prof])
    return 0


if __name__ == "__main__":
    sys.exit(main())