GitHub Viewer
#!/usr/bin/env python3
"""Run the actual preview renderer in headless Chrome, with local Yoga assets."""
import functools
import json
import os
import shutil
import signal
import subprocess
import tempfile
import threading
import time
import urllib.request
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from websockets.exceptions import WebSocketException
from websockets.sync.client import connect
from pythonnative.sdk.schema import manifest
ROOT = Path(__file__).resolve().parents[1]
class Handler(SimpleHTTPRequestHandler):
"""Serve the checked-in renderer and the current Python contract."""
def do_GET(self) -> None:
"""Generate the schema route used by the production preview server."""
if self.path == "/src/pythonnative/devserver/static/schema.js":
payload = ("export default " + json.dumps(manifest()) + ";").encode()
self.send_response(200)
self.send_header("Content-Type", "text/javascript")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
else:
super().do_GET()
def log_message(self, *_args: object) -> None:
"""Keep successful asset requests out of test output."""
def stop_browser(process: subprocess.Popen, port_file: Path) -> None:
"""Finish Chrome's profile writes before its temporary directory is removed."""
try:
if process.poll() is None:
try:
port, endpoint = port_file.read_text().splitlines()[:2]
with connect(f"ws://127.0.0.1:{port}{endpoint}", open_timeout=5, close_timeout=5) as socket:
socket.send(json.dumps({"id": 1, "method": "Browser.close"}))
socket.recv(timeout=5)
except (OSError, ValueError, TimeoutError, WebSocketException):
# Chrome can close the socket before acknowledging shutdown.
# Give it time to exit before the process-group fallback.
pass
try:
process.wait(timeout=10)
except subprocess.TimeoutExpired:
pass # The process-group fallback below also stops its children.
finally:
try:
if os.name == "posix":
os.killpg(process.pid, signal.SIGKILL)
else:
process.kill()
except ProcessLookupError:
pass
process.wait(timeout=10)
def main() -> None:
"""Execute browser tests and propagate failures to CI."""
chrome = os.environ.get("CHROME_BIN") or shutil.which("google-chrome") or shutil.which("chromium")
if not chrome:
chrome = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
if not Path(chrome).is_file():
raise SystemExit("Install Chrome or set CHROME_BIN to run renderer tests")
server = ThreadingHTTPServer(("127.0.0.1", 0), functools.partial(Handler, directory=str(ROOT)))
worker = threading.Thread(target=server.serve_forever, daemon=True)
worker.start()
try:
with tempfile.TemporaryDirectory(prefix="pn-browser-") as profile:
with tempfile.TemporaryFile(mode="w+") as log:
process = subprocess.Popen(
[
chrome,
"--headless",
"--no-sandbox",
"--disable-gpu",
"--no-first-run",
"--disable-background-networking",
"--disable-extensions",
"--no-default-browser-check",
f"--user-data-dir={profile}",
"--remote-debugging-port=0",
"about:blank",
],
stdout=log,
stderr=log,
start_new_session=True,
)
port_file = Path(profile) / "DevToolsActivePort"
try:
deadline = time.monotonic() + 45
while not port_file.exists() and process.poll() is None and time.monotonic() < deadline:
time.sleep(0.1)
if not port_file.exists():
log.seek(0)
raise RuntimeError("Chrome didn't start: " + log.read()[-5000:])
port = int(port_file.read_text().splitlines()[0])
# Chrome writes the port file before DevTools reliably answers on slow CI runners.
target = None
deadline = time.monotonic() + 30
while target is None:
try:
with urllib.request.urlopen(f"http://127.0.0.1:{port}/json/list", timeout=5) as response:
targets = json.load(response)
target = next((item for item in targets if item.get("type") == "page"), None)
except OSError:
if process.poll() is not None or time.monotonic() >= deadline:
raise
if target is None:
if time.monotonic() >= deadline:
raise RuntimeError("Chrome DevTools didn't list a page target")
time.sleep(0.5)
with connect(target["webSocketDebuggerUrl"], open_timeout=5) as socket:
request_id = 0
def command(method: str, params: dict) -> dict:
nonlocal request_id
request_id += 1
socket.send(json.dumps({"id": request_id, "method": method, "params": params}))
while True:
message = json.loads(socket.recv(timeout=10))
if message.get("id") == request_id:
if "error" in message:
raise RuntimeError(str(message["error"]))
return message.get("result", {})
command(
"Page.navigate",
{"url": f"http://127.0.0.1:{server.server_port}/tests/browser/renderer.html"},
)
deadline = time.monotonic() + 45
output = ""
while time.monotonic() < deadline:
result = command(
"Runtime.evaluate",
{"expression": "document.getElementById('result')?.outerHTML", "returnByValue": True},
)
output = result.get("result", {}).get("value", "")
if 'data-status="passed"' in output:
break
if 'data-status="failed"' in output:
raise RuntimeError(output)
time.sleep(0.1)
else:
raise RuntimeError(
"Browser tests timed out: "
+ output
+ str(
command(
"Runtime.evaluate",
{"expression": "document.documentElement.outerHTML", "returnByValue": True},
)
)
)
finally:
stop_browser(process, port_file)
print("Browser renderer acceptance tests passed (Chrome + Yoga WASM)")
finally:
server.shutdown()
server.server_close()
if __name__ == "__main__":
main()