pkg-facet-wrapped/tests/robot_installed.py
"""Robot-UI drive against the PIP-INSTALLED facet-wrapped artifact, RECORDED to saga.
Loads the installed package's loader page in a headless browser, mounts each wrapped
component with real data, and asserts on the component's own `state_json` — never a
substring, never a screenshot. Also drives the honest-refusal path.
The installed artifact is the thing under test: a wheel can ship a stale or missing
wasm and every source-level test still passes.
**Every drive is recorded as a saga tape** (`nornir.v1.Saga.RecordEvents`) and then
read back off the server (`ReadSession`) and asserted against. A drive that only
asserts in-process proves nothing after the process exits; a landed tape is a
durable, replayable record of which component mounted, with what state, and which
refusals fired — queryable in Mímir's `saga_events` long after this run.
The recorder is `facett_demo.saga` (test-time import, like playwright — the
wheel itself stays dependency-free). Point it elsewhere with `NORNIR_SERVER` /
`NORNIR_WORKSPACE`. Set `FACETT_ROBOT_SAGA=0` to drive without recording; the run
then reports the tape as SKIPPED rather than quietly passing an empty one.
"""
from __future__ import annotations
import http.server
import json
import os
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}"
# ── the saga tape: record every drive, then prove it landed ────────────────────
SAGA_ON = os.environ.get("FACETT_ROBOT_SAGA", "1") != "0"
saga = None
if SAGA_ON:
from facett_demo.saga import SagaRecorder
# strict=True: an unreachable tape sink FAILS the run. A robot that silently
# records nothing is exactly the hole this closes.
saga = SagaRecorder(app="facet-wrapped-robot", strict=True)
saga.record("nav", route="robot/start", action="start",
detail=json.dumps(facett.build_info()))
print(f"saga session {saga.session_id} → {saga.target} [{saga.workspace}]")
# Serve the installed assets over HTTP (wasm needs a real origin, not file://).
class Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=str(ASSETS), **kw)
def log_message(self, *a):
pass
httpd = socketserver.TCPServer(("127.0.0.1", 0), Handler)
PORT = httpd.server_address[1]
threading.Thread(target=httpd.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{PORT}/index.html"
# ── the drives: (component, payload, assertion on the live state_json) ──────────
def n_of(v):
"""Length of a list field, or the field itself when the component reports a count."""
if isinstance(v, int):
return v
if isinstance(v, (list, dict)):
return len(v)
return None
def has(d, *path):
cur = d
for p in path:
if isinstance(cur, dict) and p in cur:
cur = cur[p]
else:
return None
return cur
DRIVES = [
("table",
{"title": "Person", "columns": ["id", "name", "kommun"],
"rows": [["0", "n0", "Stockholm"], ["1", "n1", "Jonkoping"], ["2", "n2", "Lulea"]]},
lambda s: s.get("columns") == ["id", "name", "kommun"] and s.get("rows") == 3),
("icebergview",
{"title": "warehouse", "table": "Person", "columns": ["id", "name"],
"snapshots": [{"id": 7, "timestamp_ms": 1699999999000, "rows": [["0", "n0"], ["1", "n1"]]},
{"id": 6, "timestamp_ms": 1699000000000, "rows": [["0", "n0"]]}]},
lambda s: s.get("table") == "Person" and n_of(s.get("snapshots")) == 2),
("stat-grid",
{"title": "rows_by_table",
"bars": [{"key": "TaxRecord", "value": 1200.0}, {"key": "Person", "value": 220.0}]},
lambda s: n_of(s.get("rows")) == 2 or s.get("count") == 2),
("plot",
{"title": "throughput", "kind": "line",
"series": [{"name": "rows/s", "points": [[0, 12.0], [1, 18.5], [2, 21.0]]}]},
lambda s: n_of(s.get("series")) == 1),
("osm2d",
{"title": "Sweden", "points": [[18.06, 59.33], [13.0, 55.6]],
"ways": [{"osm_id": 7, "coords": [18.0, 59.3, 18.1, 59.4],
"tags": "{\"highway\":\"motorway\"}"}]},
lambda s: s.get("navigable") is True and "lod_sizes" in s),
("graphnav",
{"title": "cases",
"nodes": [{"id": 0, "label": "Company", "key": "Company/556"},
{"id": 1, "label": "Vessel", "key": "Vessel/265"}],
"edges": [{"src": 0, "dst": 1, "rel": "OWNS"}]},
lambda s: has(s, "canvas", "total_nodes") == 2 and has(s, "canvas", "total_edges") == 1),
("graph3d",
{"title": "cloud",
"nodes": [{"id": 0, "label": "A"}, {"id": 1, "label": "B"}, {"id": 2, "label": "C"}],
"edges": [{"src": 0, "dst": 1, "rel": "X"}, {"src": 1, "dst": 2, "rel": "X"}]},
lambda s: n_of(s.get("nodes")) == 3 or n_of(has(s, "graph", "nodes")) == 3),
# ── the 3D map: the SAME payload as osm2d, one dimension up ────────────────
("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"}},
{"osm_id": 2, "coords": [17.70, 59.60, 17.73, 59.63],
"tags": "{\"highway\":\"primary\"}"}]},
lambda s: s.get("triangles", 0) > 0 and s.get("vertices", 0) > 0),
# ── the ops family ─────────────────────────────────────────────────────────
("pipeline",
{"title": "ingest",
"nodes": [{"name": "scan", "status": "done", "rows": 1200},
{"name": "embed", "status": "running", "throughput_rows_s": 840.0},
{"name": "index", "status": "queued"}],
"edges": [[0, 1], [1, 2]]},
lambda s: n_of(s.get("nodes")) == 3 or s.get("node_count") == 3),
("jobview",
{"title": "dwarves jobs",
"jobs": [{"id": "42", "kind": "build", "target": "korp", "status": "running",
"elapsed_ms": 8200, "progress": 0.4},
{"id": "43", "kind": "test", "status": "queued"}]},
lambda s: n_of(s.get("jobs")) == 2 or s.get("job_count") == 2 or s.get("count") == 2),
("syschart",
{"title": "nordisk",
"nodes": [{"id": "korp", "label": "korp", "badge": 3},
{"id": "holger", "label": "holger"},
{"id": "nornir", "label": "nornir"}],
"edges": [{"a": "korp", "b": "holger"}, {"a": "korp", "b": "nornir"}]},
lambda s: n_of(s.get("nodes")) == 3 or s.get("node_count") == 3),
("kanban",
{"title": "sprint",
"columns": [{"id": "todo", "title": "To do", "wip_limit": 2}, "doing", "done"],
"cards": [{"id": "1", "title": "wrap osm3d", "column": "todo", "priority": "P1"},
{"id": "2", "title": "ship", "column": "doing"}]},
lambda s: s.get("card_count") == 2 and n_of(s.get("columns")) == 3),
("gantt",
{"title": "release",
"today": "2026-07-31",
"tasks": [{"id": "a", "label": "wrap", "start": "2026-07-01", "end": "2026-07-10",
"progress": 0.5},
{"id": "b", "label": "ship", "start": "2026-07-11", "end": "2026-07-12",
"deps": ["a"], "milestone": True}]},
lambda s: n_of(s.get("tasks")) == 2 or s.get("task_count") == 2),
]
# The refusal path: a fixture-only component asked for live data must REFUSE.
REFUSALS = [
("console", {"rows": [["x"]]}, "no data seam"),
("table", {}, "nothing renderable"),
("nope", {"rows": [["x"]]}, "unknown component"),
]
results = []
# ── the headless WEBGPU adapter ───────────────────────────────────────────────
# facet-wrapped renders on WebGPU and has NO WebGL fallback, so these flags are
# load-bearing: with the old `--use-gl=swiftshader` pair alone, `requestAdapter()`
# returns null in headless chromium and nothing mounts at all.
#
# What each one is for:
# --enable-unsafe-webgpu expose WebGPU where it is not enabled by default
# --enable-features=Vulkan Dawn reaches the GPU through Vulkan
# --use-vulkan=swiftshader ...and that Vulkan is SwiftShader, a CPU implementation,
# --use-angle=swiftshader so the run needs no physical GPU and no display
# --enable-unsafe-swiftshader allow the software rasteriser in a headless context
#
# Measured on this box (playwright chromium 1228 / HeadlessChrome 149): this set
# yields adapter vendor=google architecture=swiftshader, a live GPUDevice, and a
# completed render pass. The flag set was chosen by probing alternatives — the
# defaults and `--use-gl=swiftshader` both give a null adapter.
#
# One more precondition, and it is easy to lose: WebGPU is SECURE-CONTEXT ONLY.
# It works here because the assets are served over http://127.0.0.1 (loopback counts
# as trustworthy). Serving these same files from a `file://` path or a non-loopback
# `http://` host makes `navigator.gpu` undefined and every drive below fails.
WEBGPU_FLAGS = [
"--enable-unsafe-webgpu",
"--enable-features=Vulkan",
"--use-angle=swiftshader",
"--use-vulkan=swiftshader",
"--enable-unsafe-swiftshader",
]
with sync_playwright() as p:
b = p.chromium.launch(args=WEBGPU_FLAGS)
pg = b.new_page(viewport={"width": 1200, "height": 800})
# ── prove the renderer BEFORE asserting on components ──────────────────────
# If the adapter is missing, every drive below fails for the same single reason.
# Establishing that up front turns a wall of confusing component failures into
# one accurate diagnosis.
print("-- webgpu adapter --")
pg.goto(BASE, wait_until="load", timeout=60000)
pg.wait_for_function("() => window.facet_ready === true", timeout=45000)
adapter = pg.evaluate("""async () => {
if (!window.isSecureContext) return { ok: false, why: 'not a secure context' };
if (!navigator.gpu) return { ok: false, why: 'navigator.gpu undefined' };
const a = await navigator.gpu.requestAdapter();
if (!a) return { ok: false, why: 'requestAdapter() -> null' };
const i = a.info || {};
return { ok: true, vendor: i.vendor, architecture: i.architecture };
}""")
built_for = pg.evaluate("() => window.facet_backend()")
ok = bool(adapter.get("ok")) and built_for == "webgpu"
results.append({"component": "webgpu:adapter", "ok": ok, "detail": json.dumps(adapter)})
print(f"{'PASS' if ok else 'FAIL'} adapter {json.dumps(adapter)} built_for={built_for!r}")
if saga:
saga.record("log", action="webgpu-adapter", component="(renderer)",
detail=json.dumps(adapter), level="info" if ok else "error")
# Fail FAST and legibly. Without this, a missing adapter makes every drive below
# time out waiting for a mount that cannot happen, and the run ends in a 45s
# playwright traceback that names `wait_for_function` rather than the real cause.
# One accurate line beats thirteen misleading ones.
if not ok:
print(f"\nABORT: no usable WebGPU adapter ({adapter.get('why')}).\n"
"facet-wrapped renders on WebGPU and has no WebGL fallback, so no\n"
"component can mount. Check that the browser gets WEBGPU_FLAGS above and\n"
"that the page is served from http://127.0.0.1 (WebGPU needs a secure\n"
"context; file:// and non-loopback http:// do not qualify).")
if saga:
saga.flush()
saga.close()
b.close()
httpd.shutdown()
raise SystemExit(1)
for name, payload, check in DRIVES:
url = f"{BASE}?component={name}&data={urllib.parse.quote(json.dumps(payload))}"
if saga:
saga.record("nav", route=f"?component={name}", component=name, action="goto",
detail=json.dumps(payload)[:400])
pg.goto(url, wait_until="load", timeout=60000)
pg.wait_for_function("() => window.facet_ready === true", timeout=45000)
pg.wait_for_function("() => window.facet_mounted() !== ''", timeout=45000)
pg.wait_for_timeout(900) # let a frame or two republish state
mounted = pg.evaluate("() => window.facet_mounted()")
err = pg.evaluate("() => window.facet_error()")
raw = pg.evaluate("() => window.facet_state()")
# Which graphics API the canvas is REALLY on. Asserted per drive, not once:
# this is the row that would catch a silent per-component fallback, and it
# costs nothing to check every time.
api = pg.evaluate("() => window.facet_canvas_api()")
state = json.loads(raw) if raw and raw != "null" else None
try:
ok = (mounted == name and not err and state is not None
and bool(check(state)) and api == "webgpu")
except Exception as e:
ok = False
print(f" assertion error: {e}")
if api != "webgpu":
print(f" canvas API is {api!r}, expected 'webgpu'")
if saga:
# The component's OWN published state is the tape's payload — the tape
# therefore carries what was asserted on, not a paraphrase of it.
saga.mounted(name, raw or "", action="mount",
detail=f"assert={'pass' if ok else 'fail'}")
if err:
saga.record("error", component=name, action="mount", detail=str(err), level="error")
results.append({"component": name, "mounted": mounted, "error": err, "api": api,
"ok": ok, "state_keys": sorted(state.keys())[:8] if state else []})
print(f"{'PASS' if ok else 'FAIL'} {name:12} mounted={mounted!r} api={api!r} err={err!r}")
if not ok and state:
print(" state:", json.dumps(state)[:300])
# ── no component named: the loader must WAIT, never substitute a fixture ──────
# This is the Streamlit case (args arrive by postMessage, never in the URL) and
# it regressed once: the loader defaulted to manifest[0] and mounted the `table`
# component's own knut/korp fixture, which appeared in the slot the host meant
# for the Sweden map. A fixture standing in for a caller's data is the single
# worst failure this package can have — it looks exactly like success.
print("\n-- unnamed component --")
pg.goto(BASE, wait_until="load", timeout=60000)
pg.wait_for_function("() => window.facet_ready === true", timeout=45000)
pg.wait_for_timeout(900)
m = pg.evaluate("() => window.facet_mounted()")
body = pg.evaluate("() => document.body.innerText")
canvas_hidden = pg.evaluate(
"() => getComputedStyle(document.getElementById('facet_canvas')).display === 'none'")
ok = m == "" and "awaiting a component" in body and canvas_hidden
if saga:
# `log`, not `state`: nothing mounted, so there is no component state to
# carry — recording an empty `state` event would dilute the tape's meaning.
saga.record("log", component="(none)", action="await",
detail=f"mounted={m!r} canvas_hidden={canvas_hidden}", level="info")
results.append({"component": "unnamed:awaits", "mounted": None, "ok": ok, "detail": body[:120]})
print(f"{'PASS' if ok else 'FAIL'} no-component mounted={m!r} canvas_hidden={canvas_hidden} "
f"text={body.strip()[:60]!r}")
print("\n-- refusal path --")
for name, payload, expect in REFUSALS:
url = f"{BASE}?component={name}&data={urllib.parse.quote(json.dumps(payload))}"
pg.goto(url, wait_until="load", timeout=60000)
pg.wait_for_function("() => window.facet_ready === true", timeout=45000)
pg.wait_for_timeout(600)
why = pg.evaluate("() => { const r=document.getElementById('refusal');"
" return r && r.style.display !== 'none' ? r.textContent : ''; }")
mounted = pg.evaluate("() => window.facet_mounted()")
ok = expect in why and mounted != name
if saga:
saga.refused(name, why.strip() or "(component did not refuse)")
results.append({"component": f"refusal:{name}", "refusal": why, "ok": ok})
print(f"{'PASS' if ok else 'FAIL'} refusal {name:10} -> {why[:90]!r}")
b.close()
httpd.shutdown()
# ── prove the tape landed: read it back off the server and assert on IT ────────
if saga:
print("\n-- saga tape --")
saga.record("log", action="finish", detail=f"{sum(1 for r in results if r['ok'])}/{len(results)} drives ok")
captured = saga.status()["captured"]
accepted = saga.flush()
tape = saga.read_session()
def tape_ok(cond: bool, label: str, extra: str = "") -> None:
results.append({"component": f"saga:{label}", "ok": bool(cond), "detail": extra})
print(f"{'PASS' if cond else 'FAIL'} saga {label:22} {extra}")
tape_ok(accepted == captured, "accepted-all", f"{accepted}/{captured} events")
tape_ok(len(tape) == captured, "readback-count", f"read {len(tape)} of {captured}")
tape_ok([e["seq"] for e in tape] == sorted(e["seq"] for e in tape), "seq-ordered")
# Every component that mounted must have a `state` event whose state_json is
# the very JSON the assertion ran against — round-tripped through the server.
by_state = {e["component"]: e["state_json"] for e in tape if e["kind"] == "state"}
drove = [r["component"] for r in results if r.get("mounted")]
missing = [c for c in drove if c not in by_state]
tape_ok(not missing, "state-per-drive", f"missing={missing}" if missing else f"{len(by_state)} components")
bad = [c for c, s in by_state.items() if not s or json.loads(s) is None]
tape_ok(not bad, "state-json-parses", f"unparseable={bad}" if bad else "all parse")
refusals_on_tape = [e for e in tape if e["kind"] == "error" and e["action"] == "refuse"]
tape_ok(len(refusals_on_tape) == len(REFUSALS), "refusals-recorded",
f"{len(refusals_on_tape)}/{len(REFUSALS)}")
stamped = [e for e in tape if e["app"] == "facet-wrapped-robot" and e["git_sha"]]
tape_ok(len(stamped) == len(tape), "build-stamped", f"{len(stamped)}/{len(tape)} carry app+git_sha")
sessions = saga.list_sessions()
tape_ok(any(str(s.get("session_id", "")) == saga.session_id for s in sessions),
"in-list-sessions", f"{len(sessions)} sessions on server")
print(f" session_id={saga.session_id} (replay: nornir saga replay --session {saga.session_id})")
saga.close()
else:
print("\n-- saga tape -- SKIPPED (FACETT_ROBOT_SAGA=0): nothing was recorded")
out = Path(os.environ.get("FACETT_ROBOT_OUT", Path(__file__).resolve().parent / "robot_results.json"))
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(results, indent=2))
passed = sum(1 for r in results if r["ok"])
print(f"\n{passed}/{len(results)} drives passed against the INSTALLED package"
f"{' (recorded to saga)' if saga else ''}")
raise SystemExit(0 if passed == len(results) else 1)