278 lines
13 KiB
Python
278 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the real React/WASM 3axis tutorial panel through Chromium CDP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
REPORT = ROOT / "coverage/axis3-tutorial-web-manual-verification.report.json"
|
|
PORT = 5175
|
|
CDP_PORT = 9223
|
|
CASE = "linuxcnc/configs/sim/axis/vismach/3axis-tutorial/3axis-tutorial.ini"
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def sha256_value(value: Any) -> str:
|
|
return sha256_bytes(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode())
|
|
|
|
|
|
class WebSocket:
|
|
def __init__(self, url: str):
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(url)
|
|
self.sock = socket.create_connection((parsed.hostname, parsed.port), timeout=10)
|
|
key = base64.b64encode(secrets.token_bytes(16)).decode()
|
|
path = parsed.path or "/"
|
|
self.sock.sendall((f"GET {path} HTTP/1.1\r\nHost: {parsed.hostname}:{parsed.port}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
|
|
response = b""
|
|
while b"\r\n\r\n" not in response:
|
|
response += self.sock.recv(4096)
|
|
if b"101" not in response.split(b"\r\n", 1)[0]:
|
|
raise RuntimeError(f"cdp_websocket_handshake_failed:{response[:120]!r}")
|
|
self.buffer = response.split(b"\r\n\r\n", 1)[1]
|
|
|
|
def send(self, value: str) -> None:
|
|
payload = value.encode()
|
|
length = len(payload)
|
|
mask = secrets.token_bytes(4)
|
|
masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
|
|
if length < 126:
|
|
header = bytes([0x81, 0x80 | length])
|
|
elif length < 65536:
|
|
header = bytes([0x81, 0x80 | 126]) + struct.pack("!H", length)
|
|
else:
|
|
header = bytes([0x81, 0x80 | 127]) + struct.pack("!Q", length)
|
|
self.sock.sendall(header + mask + masked)
|
|
|
|
def recv(self) -> tuple[int, bytes]:
|
|
def read(size: int) -> bytes:
|
|
while len(self.buffer) < size:
|
|
self.buffer += self.sock.recv(65536)
|
|
value, self.buffer = self.buffer[:size], self.buffer[size:]
|
|
return value
|
|
first, second = read(2)
|
|
opcode = first & 0x0F
|
|
length = second & 0x7F
|
|
if length == 126:
|
|
length = struct.unpack("!H", read(2))[0]
|
|
elif length == 127:
|
|
length = struct.unpack("!Q", read(8))[0]
|
|
masked = second & 0x80
|
|
mask = read(4) if masked else b""
|
|
payload = read(length)
|
|
if masked:
|
|
payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
|
|
return opcode, payload
|
|
|
|
def close(self) -> None:
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
class CDP:
|
|
def __init__(self, ws_url: str):
|
|
self.ws = WebSocket(ws_url)
|
|
self.counter = 0
|
|
self.events: list[dict[str, Any]] = []
|
|
|
|
def call(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
self.counter += 1
|
|
self.ws.send(json.dumps({"id": self.counter, "method": method, "params": params or {}}))
|
|
while True:
|
|
opcode, payload = self.ws.recv()
|
|
if opcode != 1:
|
|
continue
|
|
message = json.loads(payload.decode())
|
|
if message.get("id") == self.counter:
|
|
if "error" in message:
|
|
raise RuntimeError(f"cdp_error:{message['error']}")
|
|
return message.get("result", {})
|
|
self.events.append(message)
|
|
|
|
def evaluate(self, expression: str, await_promise: bool = False) -> Any:
|
|
result = self.call("Runtime.evaluate", {"expression": expression, "returnByValue": True, "awaitPromise": await_promise, "userGesture": True})
|
|
exception = result.get("exceptionDetails")
|
|
if exception:
|
|
raise RuntimeError(f"page_evaluation_failed:{exception}")
|
|
return result.get("result", {}).get("value")
|
|
|
|
|
|
def fetch_json(url: str) -> Any:
|
|
with urllib.request.urlopen(url, timeout=5) as response:
|
|
return json.loads(response.read().decode())
|
|
|
|
|
|
def terminate_group(process: subprocess.Popen[str] | None) -> None:
|
|
if not process or process.poll() is not None:
|
|
return
|
|
try:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
return
|
|
try:
|
|
process.wait(timeout=8)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
def run(timeout: float = 40.0) -> dict[str, Any]:
|
|
preview = subprocess.Popen(["npm", "--prefix", "web/react", "run", "preview", "--", "--host", "127.0.0.1", "--port", str(PORT)], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, start_new_session=True)
|
|
browser: subprocess.Popen[str] | None = None
|
|
browser_profile = tempfile.mkdtemp(prefix="linuxcnc-web-axis-cdp-")
|
|
cdp: CDP | None = None
|
|
screenshots: list[dict[str, Any]] = []
|
|
states: list[dict[str, Any]] = []
|
|
preview_output = ""
|
|
start = time.monotonic()
|
|
try:
|
|
while time.monotonic() - start < 15:
|
|
try:
|
|
urllib.request.urlopen(f"http://127.0.0.1:{PORT}/", timeout=1).close()
|
|
break
|
|
except OSError:
|
|
time.sleep(0.2)
|
|
browser = subprocess.Popen(["google-chrome", "--headless=new", "--no-sandbox", "--disable-gpu", "--enable-unsafe-swiftshader", "--use-angle=swiftshader", "--use-gl=swiftshader", "--disable-dev-shm-usage", "--disable-extensions", "--no-first-run", "--no-default-browser-check", "--window-size=1440,1000", f"--remote-debugging-port={CDP_PORT}", f"--user-data-dir={browser_profile}", f"http://127.0.0.1:{PORT}/"], cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, start_new_session=True)
|
|
ws_url = None
|
|
for _ in range(100):
|
|
try:
|
|
targets = fetch_json(f"http://127.0.0.1:{CDP_PORT}/json")
|
|
page = next(item for item in targets if item.get("type") == "page")
|
|
ws_url = page["webSocketDebuggerUrl"]
|
|
break
|
|
except (OSError, StopIteration, KeyError):
|
|
time.sleep(0.2)
|
|
if not ws_url:
|
|
raise RuntimeError("chrome_cdp_target_missing")
|
|
cdp = CDP(ws_url)
|
|
cdp.call("Runtime.enable")
|
|
cdp.call("Log.enable")
|
|
cdp.call("Page.enable")
|
|
page_ready = None
|
|
for _ in range(75):
|
|
page_ready = cdp.evaluate("({href: location.href, readyState: document.readyState, root: Boolean(document.querySelector('#root')), app: Boolean(document.querySelector('.axis-app')), htmlLength: document.documentElement?.outerHTML?.length || 0})")
|
|
if page_ready.get("app"):
|
|
break
|
|
if _ in {10, 30}:
|
|
cdp.call("Page.reload", {"ignoreCache": True})
|
|
time.sleep(0.2)
|
|
if not page_ready or not page_ready.get("app"):
|
|
diagnostics = [event for event in (cdp.events if cdp else []) if event.get("method") in {"Runtime.exceptionThrown", "Runtime.consoleAPICalled", "Log.entryAdded"}]
|
|
raise RuntimeError(f"web_page_boot_failed:{page_ready}:diagnostics={diagnostics[-8:]}")
|
|
|
|
def snapshot(label: str) -> dict[str, Any]:
|
|
value = cdp.evaluate("""(() => ({
|
|
label: %s,
|
|
href: location.href,
|
|
ready_state: document.readyState,
|
|
root_html: document.querySelector('#root')?.innerHTML?.slice(0, 120) || '',
|
|
title: document.title,
|
|
case_status: document.querySelector('.case-status')?.textContent?.trim() || '',
|
|
connection: document.querySelector('.connection-state')?.textContent?.trim() || '',
|
|
axis_state: document.querySelector('.axis-state-badge')?.textContent?.trim() || '',
|
|
machine_state: document.querySelector('.machine-state')?.textContent?.trim() || '',
|
|
operation_state: document.querySelector('.operation-state')?.textContent?.trim() || '',
|
|
program_status: document.querySelector('[data-axis-program-status]')?.getAttribute('data-axis-program-status') || '',
|
|
position: document.querySelector('.position-state')?.textContent?.trim() || '',
|
|
footer: document.querySelector('.footer-status')?.textContent?.trim() || '',
|
|
error_text: document.querySelector('.diagnostic-error')?.textContent?.trim() || document.querySelector('.program-diagnostic')?.textContent?.trim() || document.querySelector('.axis-diagnostics')?.textContent?.trim() || '',
|
|
task: document.querySelector('.task-state')?.textContent?.trim() || '',
|
|
canvases: [...document.querySelectorAll('canvas')].map(canvas => ({ width: canvas.width, height: canvas.height }))
|
|
}))()""" % json.dumps(label), await_promise=False)
|
|
states.append(value)
|
|
return value
|
|
|
|
def screenshot(label: str) -> None:
|
|
data = cdp.call("Page.captureScreenshot", {"format": "png", "captureBeyondViewport": False})["data"]
|
|
path = ROOT / "coverage" / f"axis3-tutorial-web-{label}.png"
|
|
path.write_bytes(base64.b64decode(data))
|
|
screenshots.append({"label": label, "path": str(path.relative_to(ROOT)), "bytes": path.stat().st_size, "sha256": sha256_bytes(path.read_bytes())})
|
|
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
state = snapshot("ready_poll")
|
|
if state["connection"] in {"READY", "RUNNING", "PAUSED"} or state["case_status"]:
|
|
break
|
|
time.sleep(0.5)
|
|
screenshot("ready")
|
|
snapshot("ready")
|
|
cdp.evaluate("document.querySelector('button[aria-label^=\"Emergency Stop\"]')?.click()")
|
|
time.sleep(0.8)
|
|
snapshot("estop")
|
|
cdp.evaluate("document.querySelector('button[aria-label^=\"Power on\"]')?.click()")
|
|
time.sleep(0.8)
|
|
snapshot("power")
|
|
cdp.evaluate("[...document.querySelectorAll('button')].find(button => button.textContent?.trim() === 'Home All')?.click()")
|
|
time.sleep(1.5)
|
|
screenshot("homed")
|
|
snapshot("homed")
|
|
cdp.evaluate("document.querySelector('button[aria-label^=\"Begin executing current file\"]')?.click()")
|
|
time.sleep(2.0)
|
|
screenshot("running")
|
|
snapshot("running")
|
|
complete_deadline = time.monotonic() + timeout
|
|
final = None
|
|
while time.monotonic() < complete_deadline:
|
|
final = snapshot("complete_poll")
|
|
if final["program_status"] == "completed" and final["operation_state"] in {"Operation IDLE", "IDLE"} and final["connection"] == "READY":
|
|
break
|
|
time.sleep(0.5)
|
|
screenshot("complete")
|
|
final = snapshot("complete")
|
|
status = "verified" if final["canvases"] and any(item["width"] > 0 and item["height"] > 0 for item in final["canvases"]) and final["program_status"] == "completed" and final["operation_state"] in {"Operation IDLE", "IDLE"} and final["connection"] == "READY" else "blocked"
|
|
diagnostics = [event for event in cdp.events if event.get("method") in {"Runtime.exceptionThrown", "Runtime.consoleAPICalled", "Log.entryAdded"}] if cdp else []
|
|
return {"schema": "axis3-tutorial-web-manual-verification.v1", "status": status, "case": CASE, "tool": "tool/linuxcnc-web-wasm-manual-execution", "browser": "Google Chrome headless", "url": f"http://127.0.0.1:{PORT}/", "input_policy": "real browser DOM click events through CDP; real Vite/WASM/Dedicated Worker", "states": states, "screenshots": screenshots, "browser_diagnostics": diagnostics[-40:], "preview_output_sha256": sha256_bytes(preview_output.encode()), "cross_checks": {"native_tool": "tool/linuxcnc-axis-manual-execution", "parity_report": "coverage/axis3-tutorial-web-wasm-complete-parity.report.json"}, "boundaries": ["selected 3axis-tutorial case", "real hardware excluded", "browser run is Chromium evidence"]}
|
|
finally:
|
|
if cdp:
|
|
cdp.ws.close()
|
|
terminate_group(browser)
|
|
terminate_group(preview)
|
|
if preview:
|
|
try:
|
|
preview_output = preview.communicate(timeout=5)[0]
|
|
except subprocess.TimeoutExpired:
|
|
preview_output = ""
|
|
try:
|
|
import shutil
|
|
shutil.rmtree(browser_profile, ignore_errors=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--timeout", type=float, default=40.0)
|
|
args = parser.parse_args()
|
|
report = run(args.timeout)
|
|
report["report_sha256"] = sha256_value({key: value for key, value in report.items() if key != "report_sha256"})
|
|
REPORT.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps({key: report[key] for key in ("schema", "status", "case", "browser", "screenshots", "report_sha256")}, ensure_ascii=False, sort_keys=True))
|
|
return 0 if report["status"] == "verified" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|