art-dag/mono
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SSE, response headers, view transitions, and 5 new sx attributes

Browse Source 
Implement missing SxEngine features:
- SSE (sx-sse, sx-sse-swap) with EventSource management and auto-cleanup
- Response headers: SX-Trigger, SX-Retarget, SX-Reswap, SX-Redirect,
  SX-Refresh, SX-Location, SX-Replace-Url, SX-Trigger-After-Swap/Settle
- View Transitions API: transition:true swap modifier + global config
- every:<time> trigger for polling (setInterval)
- sx-replace-url (replaceState instead of pushState)
- sx-disabled-elt (disable elements during request)
- sx-prompt (window.prompt, value sent as SX-Prompt header)
- sx-params (filter form parameters: *, none, not x,y, x,y)

Adds docs (ATTR_DETAILS, BEHAVIOR_ATTRS, headers, events), demo
components in reference.sx, API endpoints (prompt-echo, sse-time),
and 27 new unit tests for engine logic.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-03-04 11:55:21 +00:00
parent 3bffc212cc
commit 213421516e
6 changed files with 888 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
shared/static/scripts/sx.js
View File

@@ -1379,6 +1379,16 @@
var PROCESSED = "_sxBound";
var VERBS = ["get", "post", "put", "delete", "patch"];
var DEFAULT_SWAP = "outerHTML";
var _config = { globalViewTransitions: false };
/** Wrap a function in View Transition API if supported and enabled. */
function _withTransition(enabled, fn) {
if (enabled && document.startViewTransition) {
document.startViewTransition(fn);
} else {
fn();
}
}
function dispatch(el, name, detail) {
@@ -1386,6 +1396,26 @@
return el.dispatchEvent(evt);
}
/** Parse and dispatch SX-Trigger header events.
* Value can be: "myEvent" (plain string), '{"myEvent": {"key": "val"}}' (JSON). */
function _dispatchTriggerEvents(el, headerVal) {
if (!headerVal) return;
try {
var parsed = JSON.parse(headerVal);
if (typeof parsed === "object" && parsed !== null) {
for (var evtName in parsed) dispatch(el, evtName, parsed[evtName]);
} else {
dispatch(el, String(parsed), {});
}
} catch (e) {
// Plain string — may be comma-separated event names
headerVal.split(",").forEach(function (name) {
var n = name.trim();
if (n) dispatch(el, n, {});
});
}
}
function csrfToken() {
var m = document.querySelector('meta[name="csrf-token"]');
return m ? m.getAttribute("content") : null;
@@ -1458,6 +1488,15 @@
if (!window.confirm(confirmMsg)) return Promise.resolve();
}
// sx-prompt: show prompt dialog, send result as SX-Prompt header
var promptMsg = el.getAttribute("sx-prompt");
if (promptMsg) {
var promptVal = window.prompt(promptMsg);
if (promptVal === null) return Promise.resolve(); // cancelled
extraParams = extraParams || {};
extraParams.promptValue = promptVal;
}
return _doFetch(el, method, url, extraParams);
}
@@ -1496,6 +1535,11 @@
} catch (e) { /* ignore */ }
}
// SX-Prompt header from sx-prompt dialog result
if (extraParams && extraParams.promptValue !== undefined) {
headers["SX-Prompt"] = extraParams.promptValue;
}
// CSRF for same-origin mutating requests
if (method !== "GET" && sameOrigin(url)) {
var csrf = csrfToken();
@@ -1529,6 +1573,24 @@
}
}
// sx-params: filter form parameters
var paramsSpec = el.getAttribute("sx-params");
if (paramsSpec && body instanceof URLSearchParams) {
if (paramsSpec === "none") {
body = new URLSearchParams();
} else if (paramsSpec.indexOf("not ") === 0) {
var excluded = paramsSpec.substring(4).split(",").map(function (s) { return s.trim(); });
excluded.forEach(function (k) { body.delete(k); });
} else if (paramsSpec !== "*") {
var allowed = paramsSpec.split(",").map(function (s) { return s.trim(); });
var filtered = new URLSearchParams();
allowed.forEach(function (k) {
body.getAll(k).forEach(function (v) { filtered.append(k, v); });
});
body = filtered;
}
}
// Include extra inputs
var includeSel = el.getAttribute("sx-include");
if (includeSel && method !== "GET") {
@@ -1587,6 +1649,11 @@
indicatorEl.style.display = "";
}
// sx-disabled-elt: disable elements during request
var disabledEltSel = el.getAttribute("sx-disabled-elt");
var disabledElts = disabledEltSel ? Array.prototype.slice.call(document.querySelectorAll(disabledEltSel)) : [];
disabledElts.forEach(function (e) { e.disabled = true; });
var fetchOpts = { method: method, headers: headers, signal: ctrl.signal };
// Cross-origin credentials for known subdomains
try {
@@ -1608,6 +1675,7 @@
el.classList.remove("sx-request");
el.removeAttribute("aria-busy");
if (indicatorEl) { indicatorEl.classList.remove("sx-request"); indicatorEl.style.display = "none"; }
disabledElts.forEach(function (e) { e.disabled = false; });
if (!resp.ok) {
dispatch(el, "sx:responseError", { response: resp, status: resp.status });
@@ -1617,11 +1685,42 @@
return resp.text().then(function (text) {
dispatch(el, "sx:afterRequest", { response: resp });
// --- Response header processing ---
// SX-Redirect: navigate away (skip swap entirely)
var hdrRedirect = resp.headers.get("SX-Redirect");
if (hdrRedirect) { location.assign(hdrRedirect); return; }
// SX-Refresh: reload page (skip swap entirely)
var hdrRefresh = resp.headers.get("SX-Refresh");
if (hdrRefresh === "true") { location.reload(); return; }
// SX-Trigger: dispatch custom events on target
var hdrTrigger = resp.headers.get("SX-Trigger");
if (hdrTrigger) _dispatchTriggerEvents(el, hdrTrigger);
// Process the response
var swapStyle = el.getAttribute("sx-swap") || DEFAULT_SWAP;
var rawSwap = el.getAttribute("sx-swap") || DEFAULT_SWAP;
var target = resolveTarget(el, null);
var selectSel = el.getAttribute("sx-select");
// SX-Retarget: server overrides target
var hdrRetarget = resp.headers.get("SX-Retarget");
if (hdrRetarget) target = document.querySelector(hdrRetarget) || target;
// SX-Reswap: server overrides swap strategy
var hdrReswap = resp.headers.get("SX-Reswap");
if (hdrReswap) rawSwap = hdrReswap;
// Parse swap style and modifiers (e.g. "innerHTML transition:true")
var swapParts = rawSwap.split(/\s+/);
var swapStyle = swapParts[0];
var useTransition = _config.globalViewTransitions;
for (var sp = 1; sp < swapParts.length; sp++) {
if (swapParts[sp] === "transition:true") useTransition = true;
else if (swapParts[sp] === "transition:false") useTransition = false;
}
// Check for text/sx content type — use direct DOM rendering path
var ct = resp.headers.get("Content-Type") || "";
if (ct.indexOf("text/sx") >= 0) {
@@ -1663,8 +1762,10 @@
// Main swap using DOM morph
if (swapStyle !== "none" && target) {
_swapDOM(target, selectedDOM, swapStyle);
_hoistHeadElements(target);
_withTransition(useTransition, function () {
_swapDOM(target, selectedDOM, swapStyle);
_hoistHeadElements(target);
});
}
} catch (err) {
console.error("sx.js render error [v2]:", err, "\nsxSource first 500:", sxSource ? sxSource.substring(0, 500) : "(empty)");
@@ -1697,34 +1798,67 @@
// Main swap
if (swapStyle !== "none" && target) {
_swapContent(target, content, swapStyle);
_hoistHeadElements(target);
_withTransition(useTransition, function () {
_swapContent(target, content, swapStyle);
_hoistHeadElements(target);
});
}
}
// History
// SX-Location: server-driven client-side navigation
var hdrLocation = resp.headers.get("SX-Location");
if (hdrLocation) {
var locUrl = hdrLocation;
try { var locObj = JSON.parse(hdrLocation); locUrl = locObj.path || locObj; } catch (e) {}
fetch(locUrl, { headers: { "SX-Request": "true" } }).then(function (r) {
return r.text().then(function (t) {
var main = document.getElementById("main-panel");
if (main) { _swapContent(main, t, "innerHTML"); _postSwap(main); }
try { history.pushState({ sxUrl: locUrl }, "", locUrl); } catch (e) {}
});
});
return;
}
// History: sx-push-url (pushState) and sx-replace-url (replaceState)
var pushUrl = el.getAttribute("sx-push-url");
if (pushUrl === "true" || (pushUrl && pushUrl !== "false")) {
var replaceUrl = el.getAttribute("sx-replace-url");
// SX-Replace-Url response header overrides client-side attribute
var hdrReplaceUrl = resp.headers.get("SX-Replace-Url");
if (hdrReplaceUrl) {
try { history.replaceState({ sxUrl: hdrReplaceUrl, scrollY: window.scrollY }, "", hdrReplaceUrl); } catch (e) {}
} else if (pushUrl === "true" || (pushUrl && pushUrl !== "false")) {
var pushTarget = pushUrl === "true" ? url : pushUrl;
try {
history.pushState({ sxUrl: pushTarget, scrollY: window.scrollY }, "", pushTarget);
} catch (e) {
// Cross-origin pushState not allowed — full navigation
location.assign(pushTarget);
return;
}
} else if (replaceUrl === "true" || (replaceUrl && replaceUrl !== "false")) {
var replTarget = replaceUrl === "true" ? url : replaceUrl;
try {
history.replaceState({ sxUrl: replTarget, scrollY: window.scrollY }, "", replTarget);
} catch (e) { /* ignore */ }
}
dispatch(el, "sx:afterSwap", { target: target });
// SX-Trigger-After-Swap
var hdrTriggerSwap = resp.headers.get("SX-Trigger-After-Swap");
if (hdrTriggerSwap) _dispatchTriggerEvents(el, hdrTriggerSwap);
// Settle tick
requestAnimationFrame(function () {
dispatch(el, "sx:afterSettle", { target: target });
// SX-Trigger-After-Settle
var hdrTriggerSettle = resp.headers.get("SX-Trigger-After-Settle");
if (hdrTriggerSettle) _dispatchTriggerEvents(el, hdrTriggerSettle);
});
});
}).catch(function (err) {
el.classList.remove("sx-request");
el.removeAttribute("aria-busy");
if (indicatorEl) { indicatorEl.classList.remove("sx-request"); indicatorEl.style.display = "none"; }
disabledElts.forEach(function (e) { e.disabled = false; });
if (err.name === "AbortError") return;
dispatch(el, "sx:sendError", { error: err });
return _handleRetry(el, verbInfo, extraParams);
@@ -2012,6 +2146,14 @@
// ---- Trigger system ---------------------------------------------------
function _parseTime(s) {
// Parse time string: "2s" → 2000, "500ms" → 500, "1.5s" → 1500
if (!s) return 0;
if (s.indexOf("ms") >= 0) return parseInt(s, 10);
if (s.indexOf("s") >= 0) return parseFloat(s) * 1000;
return parseInt(s, 10);
}
function parseTrigger(spec) {
if (!spec) return null;
var triggers = [];
@@ -2020,12 +2162,17 @@
var p = parts[i].trim();
if (!p) continue;
var tokens = p.split(/\s+/);
// Handle "every <time>" as a special trigger
if (tokens[0] === "every" && tokens.length >= 2) {
triggers.push({ event: "every", modifiers: { interval: _parseTime(tokens[1]) } });
continue;
}
var trigger = { event: tokens[0], modifiers: {} };
for (var j = 1; j < tokens.length; j++) {
var tok = tokens[j];
if (tok === "once") trigger.modifiers.once = true;
else if (tok === "changed") trigger.modifiers.changed = true;
else if (tok.indexOf("delay:") === 0) trigger.modifiers.delay = parseInt(tok.substring(6), 10);
else if (tok.indexOf("delay:") === 0) trigger.modifiers.delay = _parseTime(tok.substring(6));
else if (tok.indexOf("from:") === 0) trigger.modifiers.from = tok.substring(5);
}
triggers.push(trigger);
@@ -2051,7 +2198,10 @@
}
triggers.forEach(function (trig) {
if (trig.event === "intersect") {
if (trig.event === "every") {
var ms = trig.modifiers.interval || 1000;
setInterval(function () { executeRequest(el, verbInfo); }, ms);
} else if (trig.event === "intersect") {
_bindIntersect(el, verbInfo, trig.modifiers);
} else if (trig.event === "load") {
setTimeout(function () { executeRequest(el, verbInfo); }, 0);
@@ -2394,6 +2544,64 @@
}
}
// ---- SSE (Server-Sent Events) ----------------------------------------
function _processSSE(root) {
var sseEls = root.querySelectorAll("[sx-sse]");
if (root.matches && root.matches("[sx-sse]")) _bindSSE(root);
for (var i = 0; i < sseEls.length; i++) _bindSSE(sseEls[i]);
}
function _bindSSE(el) {
if (el._sxSSE) return; // already connected
var url = el.getAttribute("sx-sse");
if (!url) return;
var source = new EventSource(url);
el._sxSSE = source;
// Bind swap handlers for sx-sse-swap="eventName" attributes on el and descendants
var swapEls = el.querySelectorAll("[sx-sse-swap]");
if (el.hasAttribute("sx-sse-swap")) _bindSSESwap(el, source);
for (var i = 0; i < swapEls.length; i++) _bindSSESwap(swapEls[i], source);
source.addEventListener("error", function () { dispatch(el, "sx:sseError", {}); });
source.addEventListener("open", function () { dispatch(el, "sx:sseOpen", {}); });
// Cleanup: close EventSource when element is removed from DOM
if (typeof MutationObserver !== "undefined") {
var obs = new MutationObserver(function () {
if (!document.body.contains(el)) {
source.close();
el._sxSSE = null;
obs.disconnect();
}
});
obs.observe(document.body, { childList: true, subtree: true });
}
}
function _bindSSESwap(el, source) {
var eventName = el.getAttribute("sx-sse-swap") || "message";
source.addEventListener(eventName, function (e) {
var target = resolveTarget(el, null) || el;
var swapStyle = el.getAttribute("sx-swap") || "innerHTML";
var data = e.data;
if (data.trim().charAt(0) === "(") {
try {
var dom = Sx.render(data);
_swapDOM(target, dom, swapStyle);
} catch (err) {
_swapContent(target, data, swapStyle);
}
} else {
_swapContent(target, data, swapStyle);
}
_postSwap(target);
dispatch(el, "sx:sseMessage", { data: data, event: eventName });
});
}
// ---- Process function -------------------------------------------------
function process(root) {
@@ -2415,6 +2623,9 @@
// Process sx-boost containers
_processBoosted(root);
// Process SSE connections
_processSSE(root);
// Bind sx-on:* handlers on all elements
var allOnEls = root.querySelectorAll("[sx-on\\:beforeRequest],[sx-on\\:afterRequest],[sx-on\\:afterSwap],[sx-on\\:afterSettle],[sx-on\\:responseError]");
allOnEls.forEach(function (el) {
@@ -2442,6 +2653,7 @@
var engine = {
process: process,
executeRequest: executeRequest,
config: _config,
version: "1.0.0"
};

394
shared/sx/tests/test_sx_engine.py Normal file
View File

@@ -0,0 +1,394 @@
"""Test SxEngine features in sx.js — trigger parsing, param filtering, etc.
Runs pure-logic SxEngine functions through Node.js (no DOM required).
"""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
import pytest
SX_JS = Path(__file__).resolve().parents[2] / "static" / "scripts" / "sx.js"
def _run_engine_js(js_code: str) -> str:
"""Run a JS snippet that has access to SxEngine internals.
We load sx.js with a minimal DOM stub so the IIFE doesn't crash,
then expose internal functions via a test harness.
"""
stub = """
// Minimal DOM stub for SxEngine initialisation
global.document = {
readyState: "complete",
head: { querySelector: function() { return null; } },
body: null,
querySelector: function() { return null; },
querySelectorAll: function() { return []; },
getElementById: function() { return null; },
createElement: function(t) {
return {
tagName: t, attributes: [], childNodes: [],
setAttribute: function() {},
appendChild: function() {},
querySelectorAll: function() { return []; },
};
},
createTextNode: function(t) { return { nodeType: 3, nodeValue: t }; },
createDocumentFragment: function() { return { nodeType: 11, childNodes: [], appendChild: function() {} }; },
addEventListener: function() {},
title: "",
cookie: "",
};
global.window = global;
global.window.addEventListener = function() {};
global.window.matchMedia = function() { return { matches: false }; };
global.window.confirm = function() { return true; };
global.window.prompt = function() { return ""; };
global.window.scrollTo = function() {};
global.requestAnimationFrame = function(fn) { fn(); };
global.setTimeout = global.setTimeout || function(fn) { fn(); };
global.setInterval = global.setInterval || function() {};
global.clearTimeout = global.clearTimeout || function() {};
global.console = { log: function() {}, error: function() {}, warn: function() {} };
global.CSS = { escape: function(s) { return s; } };
global.location = { href: "http://localhost/", hostname: "localhost", origin: "http://localhost", assign: function() {}, reload: function() {} };
global.history = { pushState: function() {}, replaceState: function() {} };
global.fetch = function() { return Promise.resolve({ ok: true, headers: new Map(), text: function() { return Promise.resolve(""); } }); };
global.Headers = function(o) { this._h = o || {}; this.get = function(k) { return this._h[k] || null; }; };
global.URL = function(u, b) { var full = u.indexOf("://") >= 0 ? u : b + u; this.origin = "http://localhost"; this.hostname = "localhost"; };
global.CustomEvent = function(n, o) { this.type = n; this.detail = (o || {}).detail; };
global.AbortController = function() { this.signal = {}; this.abort = function() {}; };
global.URLSearchParams = function(init) {
this._data = [];
if (init) {
if (typeof init.forEach === "function") {
var self = this;
init.forEach(function(v, k) { self._data.push([k, v]); });
}
}
this.append = function(k, v) { this._data.push([k, v]); };
this.delete = function(k) { this._data = this._data.filter(function(p) { return p[0] !== k; }); };
this.getAll = function(k) { return this._data.filter(function(p) { return p[0] === k; }).map(function(p) { return p[1]; }); };
this.toString = function() { return this._data.map(function(p) { return p[0] + "=" + p[1]; }).join("&"); };
this.forEach = function(fn) { this._data.forEach(function(p) { fn(p[1], p[0]); }); };
};
global.FormData = function() { this._data = []; this.forEach = function(fn) { this._data.forEach(function(p) { fn(p[1], p[0]); }); }; };
global.MutationObserver = function() { this.observe = function() {}; this.disconnect = function() {}; };
global.EventSource = function(url) { this.url = url; this.addEventListener = function() {}; this.close = function() {}; };
global.IntersectionObserver = function() { this.observe = function() {}; };
"""
script = f"""
{stub}
{SX_JS.read_text()}
// --- test code ---
{js_code}
"""
result = subprocess.run(
["node", "-e", script],
capture_output=True, text=True, timeout=5,
)
if result.returncode != 0:
pytest.fail(f"Node.js error:\n{result.stderr}")
return result.stdout
# ---------------------------------------------------------------------------
# parseTrigger tests
# ---------------------------------------------------------------------------
class TestParseTrigger:
"""Test the parseTrigger function for various trigger specifications."""
def _parse(self, spec: str) -> list[dict]:
out = _run_engine_js(f"""
// Access parseTrigger via the IIFE's internal scope isn't possible directly,
// but we can test it indirectly. Actually, we need to extract it.
// Since SxEngine is built as an IIFE, we need to re-expose parseTrigger.
// Let's test via a workaround: add a test method.
// Actually, parseTrigger is captured in the closure. Let's hook into process.
// Better approach: just re-parse the function from sx.js source.
// Simplest: duplicate parseTrigger logic for testing (not ideal).
// Best: we patch SxEngine to expose it before the IIFE closes.
// Actually, the simplest approach: the _parseTime and parseTrigger functions
// are inside the SxEngine IIFE. We can test them by examining the behavior
// through the process() function, but that needs DOM.
//
// Instead, let's just eval the same code to test the logic:
var _parseTime = function(s) {{
if (!s) return 0;
if (s.indexOf("ms") >= 0) return parseInt(s, 10);
if (s.indexOf("s") >= 0) return parseFloat(s) * 1000;
return parseInt(s, 10);
}};
var parseTrigger = function(spec) {{
if (!spec) return null;
var triggers = [];
var parts = spec.split(",");
for (var i = 0; i < parts.length; i++) {{
var p = parts[i].trim();
if (!p) continue;
var tokens = p.split(/\\s+/);
if (tokens[0] === "every" && tokens.length >= 2) {{
triggers.push({{ event: "every", modifiers: {{ interval: _parseTime(tokens[1]) }} }});
continue;
}}
var trigger = {{ event: tokens[0], modifiers: {{}} }};
for (var j = 1; j < tokens.length; j++) {{
var tok = tokens[j];
if (tok === "once") trigger.modifiers.once = true;
else if (tok === "changed") trigger.modifiers.changed = true;
else if (tok.indexOf("delay:") === 0) trigger.modifiers.delay = _parseTime(tok.substring(6));
else if (tok.indexOf("from:") === 0) trigger.modifiers.from = tok.substring(5);
}}
triggers.push(trigger);
}}
return triggers;
}};
process.stdout.write(JSON.stringify(parseTrigger({json.dumps(spec)})));
""")
return json.loads(out)
def test_click(self):
result = self._parse("click")
assert len(result) == 1
assert result[0]["event"] == "click"
def test_every_seconds(self):
result = self._parse("every 2s")
assert len(result) == 1
assert result[0]["event"] == "every"
assert result[0]["modifiers"]["interval"] == 2000
def test_every_milliseconds(self):
result = self._parse("every 500ms")
assert len(result) == 1
assert result[0]["event"] == "every"
assert result[0]["modifiers"]["interval"] == 500
def test_delay_modifier(self):
result = self._parse("input changed delay:300ms")
assert result[0]["event"] == "input"
assert result[0]["modifiers"]["changed"] is True
assert result[0]["modifiers"]["delay"] == 300
def test_multiple_triggers(self):
result = self._parse("click, every 5s")
assert len(result) == 2
assert result[0]["event"] == "click"
assert result[1]["event"] == "every"
assert result[1]["modifiers"]["interval"] == 5000
def test_once_modifier(self):
result = self._parse("click once")
assert result[0]["modifiers"]["once"] is True
def test_from_modifier(self):
result = self._parse("keyup from:#search")
assert result[0]["event"] == "keyup"
assert result[0]["modifiers"]["from"] == "#search"
def test_load_trigger(self):
result = self._parse("load")
assert result[0]["event"] == "load"
def test_intersect(self):
result = self._parse("intersect once")
assert result[0]["event"] == "intersect"
assert result[0]["modifiers"]["once"] is True
def test_delay_seconds(self):
result = self._parse("click delay:1s")
assert result[0]["modifiers"]["delay"] == 1000
# ---------------------------------------------------------------------------
# sx-params filtering tests
# ---------------------------------------------------------------------------
class TestParamsFiltering:
"""Test the sx-params parameter filtering logic."""
def _filter(self, params_spec: str, form_data: dict[str, str]) -> dict[str, str]:
fd_entries = json.dumps([[k, v] for k, v in form_data.items()])
out = _run_engine_js(f"""
var body = new URLSearchParams();
var entries = {fd_entries};
entries.forEach(function(p) {{ body.append(p[0], p[1]); }});
var paramsSpec = {json.dumps(params_spec)};
if (paramsSpec === "none") {{
body = new URLSearchParams();
}} else if (paramsSpec.indexOf("not ") === 0) {{
var excluded = paramsSpec.substring(4).split(",").map(function(s) {{ return s.trim(); }});
excluded.forEach(function(k) {{ body.delete(k); }});
}} else if (paramsSpec !== "*") {{
var allowed = paramsSpec.split(",").map(function(s) {{ return s.trim(); }});
var filtered = new URLSearchParams();
allowed.forEach(function(k) {{ body.getAll(k).forEach(function(v) {{ filtered.append(k, v); }}); }});
body = filtered;
}}
var result = {{}};
body.forEach(function(v, k) {{ result[k] = v; }});
process.stdout.write(JSON.stringify(result));
""")
return json.loads(out)
def test_all(self):
result = self._filter("*", {"a": "1", "b": "2"})
assert result == {"a": "1", "b": "2"}
def test_none(self):
result = self._filter("none", {"a": "1", "b": "2"})
assert result == {}
def test_include(self):
result = self._filter("name", {"name": "Alice", "secret": "123"})
assert result == {"name": "Alice"}
def test_include_multiple(self):
result = self._filter("name,email", {"name": "Alice", "email": "a@b.c", "secret": "123"})
assert "name" in result
assert "email" in result
assert "secret" not in result
def test_exclude(self):
result = self._filter("not secret", {"name": "Alice", "secret": "123", "email": "a@b.c"})
assert "name" in result
assert "email" in result
assert "secret" not in result
# ---------------------------------------------------------------------------
# _dispatchTriggerEvents parsing tests
# ---------------------------------------------------------------------------
class TestTriggerEventParsing:
"""Test SX-Trigger header value parsing."""
def _parse_trigger(self, header_val: str) -> list[dict]:
out = _run_engine_js(f"""
var events = [];
// Stub dispatch to capture events
function dispatch(el, name, detail) {{
events.push({{ name: name, detail: detail }});
return true;
}}
function _dispatchTriggerEvents(el, headerVal) {{
if (!headerVal) return;
try {{
var parsed = JSON.parse(headerVal);
if (typeof parsed === "object" && parsed !== null) {{
for (var evtName in parsed) dispatch(el, evtName, parsed[evtName]);
}} else {{
dispatch(el, String(parsed), {{}});
}}
}} catch (e) {{
headerVal.split(",").forEach(function(name) {{
var n = name.trim();
if (n) dispatch(el, n, {{}});
}});
}}
}}
_dispatchTriggerEvents(null, {json.dumps(header_val)});
process.stdout.write(JSON.stringify(events));
""")
return json.loads(out)
def test_plain_string(self):
events = self._parse_trigger("myEvent")
assert len(events) == 1
assert events[0]["name"] == "myEvent"
def test_comma_separated(self):
events = self._parse_trigger("eventA, eventB")
assert len(events) == 2
assert events[0]["name"] == "eventA"
assert events[1]["name"] == "eventB"
def test_json_object(self):
events = self._parse_trigger('{"myEvent": {"key": "val"}}')
assert len(events) == 1
assert events[0]["name"] == "myEvent"
assert events[0]["detail"]["key"] == "val"
def test_json_multiple(self):
events = self._parse_trigger('{"a": {}, "b": {"x": 1}}')
assert len(events) == 2
names = [e["name"] for e in events]
assert "a" in names
assert "b" in names
# ---------------------------------------------------------------------------
# _parseTime tests
# ---------------------------------------------------------------------------
class TestParseTime:
"""Test the time parsing utility."""
def _parse_time(self, s: str) -> int:
out = _run_engine_js(f"""
var _parseTime = function(s) {{
if (!s) return 0;
if (s.indexOf("ms") >= 0) return parseInt(s, 10);
if (s.indexOf("s") >= 0) return parseFloat(s) * 1000;
return parseInt(s, 10);
}};
process.stdout.write(String(_parseTime({json.dumps(s)})));
""")
return int(out)
def test_seconds(self):
assert self._parse_time("2s") == 2000
def test_milliseconds(self):
assert self._parse_time("500ms") == 500
def test_fractional_seconds(self):
assert self._parse_time("1.5s") == 1500
def test_plain_number(self):
assert self._parse_time("100") == 100
def test_empty(self):
assert self._parse_time("") == 0
# ---------------------------------------------------------------------------
# View Transition parsing tests
# ---------------------------------------------------------------------------
class TestSwapParsing:
"""Test sx-swap value parsing with transition modifier."""
def _parse_swap(self, raw_swap: str) -> dict:
out = _run_engine_js(f"""
var rawSwap = {json.dumps(raw_swap)};
var swapParts = rawSwap.split(/\\s+/);
var swapStyle = swapParts[0];
var useTransition = false;
for (var sp = 1; sp < swapParts.length; sp++) {{
if (swapParts[sp] === "transition:true") useTransition = true;
else if (swapParts[sp] === "transition:false") useTransition = false;
}}
process.stdout.write(JSON.stringify({{ style: swapStyle, transition: useTransition }}));
""")
return json.loads(out)
def test_plain_swap(self):
result = self._parse_swap("innerHTML")
assert result["style"] == "innerHTML"
assert result["transition"] is False
def test_transition_true(self):
result = self._parse_swap("innerHTML transition:true")
assert result["style"] == "innerHTML"
assert result["transition"] is True
def test_transition_false(self):
result = self._parse_swap("outerHTML transition:false")
assert result["style"] == "outerHTML"
assert result["transition"] is False

34
shared/tests/test_sx_app_pages.py
View File

@@ -165,6 +165,19 @@ ATTRIBUTES = [
"sx-media",
"sx-disable",
"sx-on", # URL slug for sx-on:*
"sx-boost",
"sx-preload",
"sx-preserve",
"sx-indicator",
"sx-validate",
"sx-ignore",
"sx-optimistic",
"sx-replace-url",
"sx-disabled-elt",
"sx-prompt",
"sx-params",
"sx-sse",
"sx-sse-swap",
"sx-retry",
"data-sx",
"data-sx-env",
@@ -441,3 +454,24 @@ class TestReferenceAPIs:
def test_flaky(self):
r = _get("/reference/api/flaky")
assert r.status_code in (200, 503)
def test_prompt_echo(self):
r = httpx.get(
f"{SX_BASE}/reference/api/prompt-echo",
headers={**HEADERS, "SX-Prompt": "Alice"},
timeout=TIMEOUT,
)
assert r.status_code == 200
assert "Alice" in r.text
def test_sse_time(self):
"""SSE endpoint returns event-stream content type."""
with httpx.stream("GET", f"{SX_BASE}/reference/api/sse-time",
headers=HEADERS, timeout=TIMEOUT) as r:
assert r.status_code == 200
ct = r.headers.get("content-type", "")
assert "text/event-stream" in ct
# Read just the first chunk to verify format
for chunk in r.iter_text():
assert "event:" in chunk or "data:" in chunk
break # only need the first chunk