pkg-facet-wrapped/tests/robot_streamlit_host.py
"""Prove the installed wheel is still usable **the way Streamlit uses it**.
`robot_installed.py` drives the loader as a top-level page with a query string. That
is NOT how Streamlit mounts a component, and the difference is exactly where a
WebGPU port can break without any other test noticing:
1. Streamlit embeds the component in an **iframe**, served from a **different
origin** than the host app (`declare_component(path=…)` gets its own server).
2. Arguments arrive by **postMessage** (`streamlit:render`) after the component
announces `streamlit:componentReady` — never in the URL.
3. The host sizes the frame from the component's `streamlit:setFrameHeight`; without
it the component renders into a zero-height box and the user sees nothing.
WebGPU is a **secure-context, iframe-sensitive** API, so (1) is a real risk and not a
formality: a renderer that works top-level can still fail inside a cross-origin frame.
This test therefore stands up TWO origins on loopback — a fake "Streamlit app" and the
component server — and drives the real handshake between them.
No Streamlit is installed or imported: this crate must contain no Python, and the
wheel must stay dependency-free. What is asserted is the protocol Streamlit speaks,
which is the thing that actually has to keep working.
Run: python tests/robot_streamlit_host.py
"""
from __future__ import annotations
import http.server
import json
import socketserver
import threading
import urllib.parse
from pathlib import Path
import facett
from playwright.sync_api import sync_playwright
ASSETS = facett.assets_dir()
assert "site-packages" in str(ASSETS), f"not the installed copy: {ASSETS}"
# Same flags as robot_installed.py — see the rationale there.
WEBGPU_FLAGS = [
"--enable-unsafe-webgpu",
"--enable-features=Vulkan",
"--use-angle=swiftshader",
"--use-vulkan=swiftshader",
"--enable-unsafe-swiftshader",
]
def serve(directory: str | None = None, page: bytes | None = None) -> int:
"""Serve `directory` (or a single `page`) on a fresh loopback port; return it."""
class H(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=directory or str(ASSETS), **kw)
def do_GET(self): # noqa: N802
if page is None:
return super().do_GET()
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.send_header("Content-Length", str(len(page)))
self.end_headers()
self.wfile.write(page)
def log_message(self, *a):
pass
httpd = socketserver.TCPServer(("127.0.0.1", 0), H)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
return httpd.server_address[1]
COMPONENT_PORT = serve(directory=str(ASSETS))
# The fake Streamlit host page. It does what Streamlit's frontend does and nothing
# else: embed the component in an iframe, wait for `componentReady`, post a
# `streamlit:render` with the args, and record `setFrameHeight`.
HOST_PAGE = ("""<!doctype html><meta charset=utf-8><title>fake streamlit host</title>
<body style="margin:0">
<iframe id="f" src="http://127.0.0.1:__PORT__/index.html"
style="width:1000px;height:600px;border:0"></iframe>
<script>
window.__ready = false; // did the component send streamlit:componentReady?
window.__heights = []; // every streamlit:setFrameHeight we were sent
window.addEventListener("message", (ev) => {
const d = ev.data || {};
if (!d.isStreamlitMessage) return;
if (d.type === "streamlit:componentReady") window.__ready = true;
if (d.type === "streamlit:setFrameHeight") window.__heights.push(d.height);
});
// Exactly Streamlit's call shape: args under `args`, posted into the frame.
window.__render = (component, data) => {
document.getElementById("f").contentWindow.postMessage(
{ isStreamlitMessage: true, type: "streamlit:render",
args: { component, data, height: 520 } }, "*");
};
</script>
</body>""").replace("__PORT__", str(COMPONENT_PORT)).encode()
HOST_PORT = serve(page=HOST_PAGE)
HOST_URL = f"http://127.0.0.1:{HOST_PORT}/"
# A couple of representative payloads: one plain 2D component and one that leans on
# the GPU path, so "it works in an iframe" is not proven only for the easy case.
DRIVES = [
("table",
{"title": "Person", "columns": ["id", "name"], "rows": [["0", "a"], ["1", "b"]]},
lambda s: s.get("rows") == 2),
("osm3d",
{"title": "Sigtuna",
"ways": [{"osm_id": 1,
"coords": [17.71, 59.61, 17.72, 59.61, 17.72, 59.62, 17.71, 59.61],
"tags": {"building": "yes"}}]},
lambda s: s.get("triangles", 0) > 0),
]
results = []
def row(label: str, ok: bool, detail: str = "") -> None:
results.append({"check": label, "ok": bool(ok), "detail": detail})
print(f"{'PASS' if ok else 'FAIL'} {label:26} {detail}")
with sync_playwright() as p:
b = p.chromium.launch(args=WEBGPU_FLAGS)
pg = b.new_page(viewport={"width": 1200, "height": 800})
pg.goto(HOST_URL, wait_until="load", timeout=60000)
frame = pg.frame_locator("#f")
# The component lives on a DIFFERENT origin than the host page — assert that,
# so this test cannot silently degrade into a same-origin one.
fr = [f for f in pg.frames if "index.html" in f.url][0]
row("cross-origin-iframe", str(COMPONENT_PORT) in fr.url and HOST_PORT != COMPONENT_PORT,
f"host :{HOST_PORT} -> component :{COMPONENT_PORT}")
fr.wait_for_function("() => window.facet_ready === true", timeout=45000)
# 1. The handshake Streamlit will not mount without.
pg.wait_for_function("() => window.__ready === true", timeout=30000)
row("streamlit:componentReady", pg.evaluate("() => window.__ready"))
# 2. WebGPU must survive the cross-origin iframe.
ad = fr.evaluate("""async () => {
if (!window.isSecureContext) return { ok: false, why: 'not secure context' };
if (!navigator.gpu) return { ok: false, why: 'navigator.gpu undefined in iframe' };
const a = await navigator.gpu.requestAdapter();
return a ? { ok: true, architecture: (a.info || {}).architecture }
: { ok: false, why: 'requestAdapter() -> null in iframe' };
}""")
row("webgpu-in-iframe", ad.get("ok"), json.dumps(ad))
# 3. Nothing may render before the host sends args — the fixture-substitution bug.
row("awaits-before-render", fr.evaluate("() => window.facet_mounted()") == "",
"nothing mounted before streamlit:render")
# 4. The actual drives, over postMessage exactly as Streamlit sends them.
for name, payload, check in DRIVES:
pg.evaluate("([c, d]) => window.__render(c, d)", [name, payload])
fr.wait_for_function(f"() => window.facet_mounted() === {name!r}", timeout=45000)
pg.wait_for_timeout(900)
raw = fr.evaluate("() => window.facet_state()")
api = fr.evaluate("() => window.facet_canvas_api()")
err = fr.evaluate("() => window.facet_error()")
state = json.loads(raw) if raw and raw != "null" else None
ok = state is not None and not err and bool(check(state)) and api == "webgpu"
row(f"render:{name}", ok, f"api={api!r} err={err!r}")
# 5. Frame height: without this Streamlit shows a zero-height component.
heights = pg.evaluate("() => window.__heights")
row("streamlit:setFrameHeight", bool(heights) and all(h > 0 for h in heights),
f"heights={heights}")
b.close()
passed = sum(1 for r in results if r["ok"])
print(f"\n{passed}/{len(results)} Streamlit-host checks passed against the INSTALLED wheel")
raise SystemExit(0 if passed == len(results) else 1)