Files
site/tools/harness-safari.py
T

58 lines
3.5 KiB
Python

#!/usr/bin/env python3
"""tools/harness-safari.py [FRAGMENT] [TIMEOUT_S]
The loop harness in Safari (WebKit, the iPhone's engine: play() policy, read-only element volume, one audio session),
driven over WebDriver by safaridriver, in real time, over http from a temp server. Prints #diag once it says END.
Needs `safaridriver --enable` once (admin). Stdlib only.
Safari has no autoplay flag: the harness fragment gets a real click first (WebDriver clicks count as user activation),
so what runs is what a tapped phone runs.
"""
import json, pathlib, socket, subprocess, sys, tempfile, time, urllib.request
ROOT = pathlib.Path(__file__).resolve().parent.parent
FRAG = ROOT / (sys.argv[1] if len(sys.argv) > 1 else "tools/harness.html")
TIMEOUT = int(sys.argv[2]) if len(sys.argv) > 2 else 90
out = pathlib.Path(tempfile.mkdtemp())
page = (ROOT / "index.html").read_text()
h = FRAG.read_text()
script = h[h.index("<!-- harness -->"):]
# a click target that starts the harness: WebDriver's click is a user activation in Safari
gate = '<button id="go" style="position:fixed;top:0;left:0;z-index:99;padding:20px">GO</button><script>window.__gate=new Promise(r=>document.getElementById("go").addEventListener("click",()=>{document.getElementById("go").remove();r();}));</script>'
script = script.replace('(async () => {\n document.dispatchEvent(new Event("keydown"));', "(async () => {\n await window.__gate;", 1)
assert "await window.__gate" in script, "the fragment's opening line changed; update the gate"
(out / "run.html").write_text(page.replace("</body>", gate + script + "</body>"))
(out / "brain.js").write_text((ROOT / "brain.js").read_text())
for extra in ("sw.js", "manifest.json", "favicon.svg"):
if (ROOT / extra).exists(): (out / extra).write_text((ROOT / extra).read_text())
s = socket.socket(); s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]; s.close()
srv = subprocess.Popen([sys.executable, "-m", "http.server", str(port), "--bind", "127.0.0.1", "--directory", str(out)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
ds = socket.socket(); ds.bind(("127.0.0.1", 0)); dport = ds.getsockname()[1]; ds.close()
drv = subprocess.Popen(["/usr/bin/safaridriver", "-p", str(dport)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def call(method, path, body=None):
req = urllib.request.Request(f"http://127.0.0.1:{dport}{path}", data=json.dumps(body).encode() if body is not None else None, method=method, headers={"content-type": "application/json"})
with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read())["value"]
try:
for _ in range(40):
try: urllib.request.urlopen(f"http://127.0.0.1:{dport}/status", timeout=2); break
except Exception: time.sleep(0.5)
sess = call("POST", "/session", {"capabilities": {"alwaysMatch": {"browserName": "safari"}}})["sessionId"]
base = f"/session/{sess}"
call("POST", f"{base}/url", {"url": f"http://127.0.0.1:{port}/run.html"})
time.sleep(1.5)
el = call("POST", f"{base}/element", {"using": "css selector", "value": "#go"})
call("POST", f"{base}/element/{list(el.values())[0]}/click", {})
t0 = time.time(); diag = ""
while time.time() - t0 < TIMEOUT:
diag = call("POST", f"{base}/execute/sync", {"script": "const d=document.getElementById('diag');return d?d.textContent:''", "args": []})
if diag.rstrip().endswith("END"): break
time.sleep(1)
print(diag or "no diag: the harness did not run")
call("DELETE", base)
finally:
drv.terminate(); srv.terminate()