Project-SDE-WP-Suite/tests/cdp.py

"""Minimal Chrome DevTools Protocol client. Stdlib only — no pip, no Selenium.

Enough CDP to load a page in a headless browser as a signed-in user, capture any
JavaScript that failed, and interrogate the rendered DOM. Same no-dependency rule
as server/smoketest.py, for the same reason: these tools have to run on a plain
Python install on whatever machine is to hand.

The WebSocket bits are hand-rolled because there is no stdlib ws client and
http.client cannot upgrade: handshake, masked client frames out, unmasked in.

Used by tests/browser_check.py. Nothing in the app imports this.
"""
import base64
import json
import os
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import time
import urllib.request

# Where to find a headless-capable browser. Edge ships with Windows, so it is
# first; Chrome is accepted too. WP_BROWSER overrides everything.
_CANDIDATES = [
    r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
    r"C:\Program Files\Microsoft\Edge\Application\msedge.exe",
    r"C:\Program Files\Google\Chrome\Application\chrome.exe",
    r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
    "/usr/bin/microsoft-edge",
    "/usr/bin/google-chrome",
    "/usr/bin/chromium",
    "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
    "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
]


def find_browser():
    """Path to a usable browser, or None. Check this before running: a missing
    browser is 'could not run', not 'the app is broken'."""
    env = os.getenv("WP_BROWSER")
    if env:
        return env if os.path.exists(env) else None
    for p in _CANDIDATES:
        if os.path.exists(p):
            return p
    for name in ("msedge", "google-chrome", "chromium", "chrome"):
        found = shutil.which(name)
        if found:
            return found
    return None


def free_port():
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


class WS:
    """One WebSocket connection, speaking CDP's request/response + event mix."""

    def __init__(self, url, timeout=25):
        assert url.startswith("ws://"), url
        hostport, _, path = url[5:].partition("/")
        host, _, port = hostport.partition(":")
        self.sock = socket.create_connection((host, int(port or 80)), timeout=timeout)
        self.sock.settimeout(timeout)
        key = base64.b64encode(os.urandom(16)).decode()
        self.sock.sendall((
            f"GET /{path} HTTP/1.1\r\nHost: {hostport}\r\nUpgrade: websocket\r\n"
            f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\n"
            f"Sec-WebSocket-Version: 13\r\n\r\n").encode())
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = self.sock.recv(4096)
            if not chunk:
                raise EOFError("handshake closed")
            buf += chunk
        head, _, rest = buf.partition(b"\r\n\r\n")
        if b" 101 " not in head.split(b"\r\n")[0]:
            raise RuntimeError("upgrade refused: " + head.decode(errors="replace")[:200])
        self.buf = rest
        self._id = 0
        self.events = []

    def _send_frame(self, payload: bytes):
        mask = os.urandom(4)
        n = len(payload)
        h = bytearray([0x81])
        if n < 126:
            h.append(0x80 | n)
        elif n < 1 << 16:
            h.append(0x80 | 126); h += struct.pack(">H", n)
        else:
            h.append(0x80 | 127); h += struct.pack(">Q", n)
        h += mask
        self.sock.sendall(bytes(h) + bytes(b ^ mask[i % 4] for i, b in enumerate(payload)))

    def _read(self, n):
        while len(self.buf) < n:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise EOFError("socket closed")
            self.buf += chunk
        out, self.buf = self.buf[:n], self.buf[n:]
        return out

    def _recv_frame(self):
        while True:
            h = self._read(2)
            op, ln = h[0] & 0x0F, h[1] & 0x7F
            if ln == 126:
                ln = struct.unpack(">H", self._read(2))[0]
            elif ln == 127:
                ln = struct.unpack(">Q", self._read(8))[0]
            data = self._read(ln)
            if op == 1:
                return json.loads(data.decode())
            if op == 8:
                raise EOFError("browser closed the connection")
            if op == 9:
                self._send_frame(b"")           # ping -> pong

    def call(self, method, params=None, timeout=25):
        self._id += 1
        mine = self._id
        self._send_frame(json.dumps({"id": mine, "method": method,
                                     "params": params or {}}).encode())
        deadline = time.time() + timeout
        while time.time() < deadline:
            msg = self._recv_frame()
            if msg.get("id") == mine:
                if "error" in msg:
                    raise RuntimeError(f"{method}: {msg['error']}")
                return msg.get("result", {})
            if "method" in msg:
                self.events.append(msg)
        raise TimeoutError(method)

    def drain(self, seconds=0.4):
        """Collect pending events without blocking on a reply."""
        end = time.time() + seconds
        self.sock.settimeout(0.15)
        try:
            while time.time() < end:
                try:
                    msg = self._recv_frame()
                except (socket.timeout, TimeoutError):
                    break
                if "method" in msg:
                    self.events.append(msg)
        finally:
            self.sock.settimeout(25)

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


class Browser:
    """A headless browser process and its debugging port.

    Owns teardown, which is the fiddly part: a browser spawns a tree of renderer
    and GPU processes, and killing the process we launched leaves the rest behind
    (one careless run left 98 strays). So we kill the tree AND sweep anything still
    holding our unique profile directory — matching on that path, never on the
    process name, so a real browser the user has open is never touched.
    """

    # Launching is occasionally flaky: the process we start can hand off to another
    # instance and exit rc=0 without ever binding the port, especially if a previous
    # run left processes behind. Retrying with a fresh profile and port clears it.
    ATTEMPTS = 3

    def __init__(self, exe=None, port=None):
        self.exe = exe or find_browser()
        if not self.exe:
            raise RuntimeError("no headless-capable browser found (set WP_BROWSER)")
        last = ""
        for attempt in range(1, self.ATTEMPTS + 1):
            self.port = port if (port and attempt == 1) else free_port()
            self.profile = tempfile.mkdtemp(prefix="wpsuite-cdp-")
            self.proc = subprocess.Popen(
                [self.exe, "--headless=new", f"--remote-debugging-port={self.port}",
                 f"--user-data-dir={self.profile}", "--remote-allow-origins=*",
                 "--no-first-run", "--no-default-browser-check", "--disable-gpu",
                 "--disable-extensions", "--disable-sync",
                 "--window-size=1400,1000", "about:blank"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            for _ in range(160):
                try:
                    with urllib.request.urlopen(
                            f"http://127.0.0.1:{self.port}/json/version", timeout=1) as r:
                        json.load(r)
                    return
                except Exception:
                    if self.proc.poll() is not None:
                        last = f"exited rc={self.proc.returncode} without binding the port"
                        break
                    time.sleep(0.25)
            else:
                last = "never bound the debugging port"
            self.close()
            time.sleep(1.5)          # let the old tree finish dying
        raise RuntimeError(f"browser would not start after {self.ATTEMPTS} attempts ({last})")

    def page(self):
        return Page(self.port)

    def close(self):
        pid = self.proc.pid
        try:
            self.proc.kill()
        except OSError:
            pass
        if sys.platform == "win32":
            subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"],
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            # Sweep any orphan that still has our profile open. Scoped to the temp
            # profile path, so it cannot match a browser window the user opened.
            leaf = os.path.basename(self.profile)
            subprocess.run(
                ["powershell", "-NoProfile", "-Command",
                 "Get-CimInstance Win32_Process | Where-Object { $_.CommandLine -like "
                 f"'*{leaf}*' }} | ForEach-Object {{ try {{ Stop-Process -Id "
                 "$_.ProcessId -Force -ErrorAction Stop } catch {} }"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        shutil.rmtree(self.profile, ignore_errors=True)


class Page:
    """One headless tab, with JS-error capture and a DOM query helper."""

    def __init__(self, port):
        self.ws = WS(self._page_ws(port))
        for domain in ("Page.enable", "Runtime.enable", "Log.enable", "Network.enable"):
            self.ws.call(domain)

    @staticmethod
    def _page_ws(port):
        for _ in range(40):
            with urllib.request.urlopen(f"http://127.0.0.1:{port}/json/list", timeout=2) as r:
                for t in json.load(r):
                    if t.get("type") == "page" and t.get("webSocketDebuggerUrl"):
                        return t["webSocketDebuggerUrl"]
            time.sleep(0.25)
        raise RuntimeError("no page target")

    def set_cookie(self, name, value, domain="127.0.0.1", path="/"):
        self.ws.call("Network.setCookie", {"name": name, "value": value,
                                          "domain": domain, "path": path})

    def clear_cookies(self):
        self.ws.call("Network.clearBrowserCookies")

    def goto(self, url, wait_for=None, timeout=20):
        """Navigate, then poll `wait_for` (a JS expression) until it is truthy.
        The pages fetch their own data after load, so waiting on the load event
        alone races the thing under test."""
        self.ws.events.clear()
        self.ws.call("Page.navigate", {"url": url})
        deadline = time.time() + timeout
        while time.time() < deadline:
            self.ws.drain(0.25)
            if any(e["method"] == "Page.loadEventFired" for e in self.ws.events):
                break
        if wait_for:
            while time.time() < deadline:
                try:
                    if self.eval(wait_for) is True:
                        break
                except Exception:
                    pass
                self.ws.drain(0.2)
        self.ws.drain(0.4)
        return self

    def eval(self, expr):
        r = self.ws.call("Runtime.evaluate", {
            "expression": expr, "returnByValue": True, "awaitPromise": True})
        if "exceptionDetails" in r:
            raise RuntimeError("JS threw: " + json.dumps(r["exceptionDetails"])[:300])
        return r.get("result", {}).get("value")

    def click(self, selector, settle=0.5):
        self.eval(f"document.querySelector({selector!r}).click()")
        time.sleep(settle)
        self.ws.drain(0.2)

    def key(self, name, settle=0.4):
        self.eval(f"document.dispatchEvent(new KeyboardEvent('keydown',{{key:{name!r}}}))")
        time.sleep(settle)

    def viewport(self, width, height=900, mobile=False, scale=1):
        """Emulate a viewport width. The two that matter are 390 (the gloved-hands
        field tablet, where the worst rendering was found) and 1440 (the desk).

        `mobile` also sets the mobile flag and a meta-viewport-aware layout, which
        is what a tablet actually reports; without it a 390px-wide desktop window
        is not the same test."""
        self.ws.call("Emulation.setDeviceMetricsOverride", {
            "width": int(width), "height": int(height),
            "deviceScaleFactor": scale, "mobile": bool(mobile),
        })
        time.sleep(0.35)                      # let media queries and reflow settle
        self.ws.drain(0.2)
        return self

    def screenshot(self, path, full_page=True):
        """Write a PNG to `path`, creating parent directories. Returns the path.

        full_page captures the whole document rather than the visible box: the
        creator form is roughly 4,700px tall and a viewport-sized shot of it would
        hide the thing being compared. Chrome refuses beyond 16,384px, so an
        over-tall page is clamped rather than failing the capture."""
        params = {"format": "png"}
        if full_page:
            try:
                m = self.ws.call("Page.getLayoutMetrics")
                size = m.get("cssContentSize") or m.get("contentSize") or {}
                w, h = size.get("width"), size.get("height")
                if w and h:
                    params["clip"] = {"x": 0, "y": 0, "width": w,
                                      "height": min(h, 16384), "scale": 1}
                    params["captureBeyondViewport"] = True
            except (RuntimeError, TimeoutError):
                pass                          # fall back to a viewport-sized shot
        shot = self.ws.call("Page.captureScreenshot", params, timeout=45)
        parent = os.path.dirname(os.path.abspath(path))
        if parent:
            os.makedirs(parent, exist_ok=True)
        with open(path, "wb") as fh:
            fh.write(base64.b64decode(shot["data"]))
        return path

    def js_errors(self):
        """Everything that means 'this page did not boot cleanly': uncaught
        exceptions, console.error calls, and browser-logged errors.

        Icon and manifest probes are ignored — they are not code faults. The URL is
        kept in the message because a bare '404 (Not Found)' is undiagnosable, and
        some log entries arrive with no url field at all."""
        out = []
        for e in self.ws.events:
            m, p = e["method"], e.get("params", {})
            if m == "Runtime.exceptionThrown":
                d = p.get("exceptionDetails", {})
                txt = d.get("exception", {}).get("description") or d.get("text", "")
                out.append("uncaught: " + str(txt).split("\n")[0])
            elif m == "Runtime.consoleAPICalled" and p.get("type") == "error":
                bits = " ".join(str(a.get("value", a.get("description", "")))
                                for a in p.get("args", []))
                out.append("console.error: " + bits[:200])
            elif m == "Log.entryAdded":
                entry = p.get("entry", {})
                if entry.get("level") != "error":
                    continue
                url, text = entry.get("url", "") or "", str(entry.get("text", ""))
                if any(s in url or s in text
                       for s in ("favicon", "manifest.webmanifest", "icon-")):
                    continue
                out.append(f"log: {text[:160]}" + (f"  [{url}]" if url else "  [no url]"))
        return out

    def close(self):
        self.ws.close()