8a0f6073f7
joule-ui takes in OpenUI Lang 1.0 the way it takes in A2UI: a model writes a surface as a program, one widget a line, and joule-ui-lang builds a widget tree against the host's registry as the program streams. After every chunk a snapshot holds the tree as far as it can be built, each widget checked by joule-ui's own validate, and every error with its source line and reason code. State, queries, operators, conditionals, built-in functions and sequences of actions are parsed and refused not-declarative. A canonical writer turns a tree back into the language; it parses back to the same tree. - joule-ui-core: WidgetSchema::positional, the order a positional call passes props in (SPEC §3); absent from the wire form when undeclared. - joule-ui-interaction: begin_interaction and Pending::finish, the lifecycle's three phases in two steps for a host whose handler is asynchronous (SPEC §6); run_interaction is built on them and seals what it sealed before. - conformance/lang/v1: L1 statements, L2 ingest, L3 streaming, L4 the canonical form, L5 refusals, with an independent Python verifier written from §11. - docs/create.md: the design of Create. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1207 lines
42 KiB
Python
1207 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""lang_verify.py: the independent verifier of joule-ui's line language (SPEC §11).
|
|
|
|
Written from SPEC.md §4 and §11 alone, sharing no code with the reference
|
|
implementation (joule-ui-rs/crates/joule-ui-lang). It recomputes every case of the
|
|
pack in conformance/lang/v1/ (L1 statements, L2 ingest, L3 streaming, L4 the
|
|
canonical form, L5 refusals) from the case's input and the pack's library, and
|
|
compares what it gets with what the pack states. Errors are compared without
|
|
their messages, which are not normative.
|
|
|
|
python3 -I lang_verify.py [<pack dir>] verify the pack
|
|
python3 -I lang_verify.py --mutations [<dir>] show that changed vectors are refused
|
|
|
|
Exit status: 0 when every case agrees (or, with --mutations, when every mutation is
|
|
caught); 1 otherwise. Standard library only.
|
|
"""
|
|
from decimal import Decimal
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DEFAULT_PACK = os.path.join(HERE, "lang", "v1")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §4 validation
|
|
|
|
def ptype(v):
|
|
return v["type"]
|
|
|
|
|
|
def validate(w, reg, path="$"):
|
|
"""SPEC §4: every error of the tree, pre-order."""
|
|
errs = []
|
|
s = reg.get(w["name"])
|
|
if s is None:
|
|
errs.append({"kind": "unknown_widget", "name": w["name"], "path": path})
|
|
return errs
|
|
props = w.get("props", {})
|
|
for p in sorted(s["props"]):
|
|
if s["props"][p].get("required") and p not in props:
|
|
errs.append({"kind": "missing_prop", "prop": p, "path": path})
|
|
for p in sorted(props):
|
|
if p in s["props"]:
|
|
want = s["props"][p]["ty"]
|
|
if want != "any" and want != ptype(props[p]):
|
|
errs.append({"kind": "wrong_prop_type", "prop": p, "path": path})
|
|
elif not s.get("allow_extra_props"):
|
|
errs.append({"kind": "extra_prop", "prop": p, "path": path})
|
|
ch = s["children"]
|
|
for i, c in enumerate(w.get("children", [])):
|
|
cp = "%s.children[%d]" % (path, i)
|
|
kind = ch["kind"]
|
|
ok = kind == "any" or (kind == "only_of" and c["name"] in ch["value"])
|
|
if not ok:
|
|
errs.append({"kind": "disallowed_child", "child": c["name"], "path": cp})
|
|
errs.extend(validate(c, reg, cp))
|
|
return errs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.1 source text
|
|
|
|
MARKER = "]]>openui:"
|
|
FENCE = "```"
|
|
|
|
|
|
def strip_comments(t):
|
|
out = []
|
|
quote = None
|
|
i = 0
|
|
n = len(t)
|
|
while i < n:
|
|
c = t[i]
|
|
if quote is not None:
|
|
out.append(c)
|
|
if c == "\\":
|
|
if i + 1 < n:
|
|
out.append(t[i + 1])
|
|
i += 2
|
|
continue
|
|
elif c == quote:
|
|
quote = None
|
|
i += 1
|
|
continue
|
|
if c in "\"'":
|
|
quote = c
|
|
out.append(c)
|
|
i += 1
|
|
continue
|
|
if c == "#" or (c == "/" and i + 1 < n and t[i + 1] == "/"):
|
|
while i < n and t[i] != "\n":
|
|
i += 1
|
|
continue
|
|
out.append(c)
|
|
i += 1
|
|
return "".join(out)
|
|
|
|
|
|
def prepare(source):
|
|
text = source.replace("\r\n", "\n")
|
|
lines = [(i + 1, l) for i, l in enumerate(text.split("\n"))]
|
|
if any(l.startswith(MARKER) for _, l in lines):
|
|
kept = []
|
|
content = False
|
|
for n, l in lines:
|
|
if l.startswith(MARKER):
|
|
content = l[len(MARKER):].startswith("content")
|
|
continue
|
|
if content:
|
|
kept.append((n, l))
|
|
lines = kept
|
|
if any(l.startswith(FENCE) for _, l in lines):
|
|
kept = []
|
|
inside = False
|
|
dq = False
|
|
skip = False
|
|
for n, l in lines:
|
|
if not inside:
|
|
if l.startswith(FENCE):
|
|
inside, dq, skip = True, False, False
|
|
continue
|
|
if not dq and l.startswith(FENCE):
|
|
inside = False
|
|
continue
|
|
kept.append((n, l))
|
|
for c in l:
|
|
if skip:
|
|
skip = False
|
|
elif dq and c == "\\":
|
|
skip = True
|
|
elif c == '"':
|
|
dq = not dq
|
|
skip = False
|
|
lines = kept
|
|
numbers = [n for n, _ in lines]
|
|
joined = strip_comments("\n".join(l for _, l in lines))
|
|
ws = " \t\n\r"
|
|
lead = len(joined) - len(joined.lstrip(ws))
|
|
numbers = numbers[joined[:lead].count("\n"):]
|
|
body = joined.strip(ws)
|
|
numbers = numbers[: len(body.split("\n"))]
|
|
return body, numbers
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.2 lexing
|
|
|
|
def is_start(c):
|
|
return ("a" <= c <= "z") or ("A" <= c <= "Z") or c == "_"
|
|
|
|
|
|
def is_part(c):
|
|
return is_start(c) or ("0" <= c <= "9")
|
|
|
|
|
|
def is_digit(c):
|
|
return "0" <= c <= "9"
|
|
|
|
|
|
VALUE_KINDS = {"ident", "state", "str", "num", "badnum", "true", "false", "null"}
|
|
|
|
|
|
def ends_value(tok):
|
|
return tok["k"] in VALUE_KINDS or (tok["k"] == "p" and tok["v"] in (")", "]"))
|
|
|
|
|
|
def decode_dq(raw):
|
|
try:
|
|
v = json.loads('"' + raw + '"')
|
|
except ValueError:
|
|
return raw
|
|
if any(0xD800 <= ord(ch) <= 0xDFFF for ch in v):
|
|
return raw
|
|
return v
|
|
|
|
|
|
def decode_sq(raw):
|
|
out = []
|
|
i = 0
|
|
while i < len(raw):
|
|
c = raw[i]
|
|
if c != "\\":
|
|
out.append(c)
|
|
i += 1
|
|
continue
|
|
if i + 1 < len(raw):
|
|
e = raw[i + 1]
|
|
out.append("\n" if e == "n" else "\t" if e == "t" else e)
|
|
i += 2
|
|
return "".join(out)
|
|
|
|
|
|
TWO = ("==", "!=", ">=", "<=", "&&", "||")
|
|
ONE = {"=": "=", ">": ">", "<": "<", "+": "+", "-": "-", "*": "*", "/": "/", "%": "%", "&": "&&", "|": "||",
|
|
"!": "!", "?": "?", ":": ":", ".": ".", ",": ",", "(": "(", ")": ")", "[": "[", "]": "]", "{": "{", "}": "}"}
|
|
|
|
|
|
def lex(t):
|
|
toks = []
|
|
i, n, line = 0, len(t), 0
|
|
|
|
def add(k, v=None, start=None, open_=False):
|
|
toks.append({"k": k, "v": v, "line": line if start is None else start, "open": open_})
|
|
|
|
while i < n:
|
|
c = t[i]
|
|
if c in " \t":
|
|
i += 1
|
|
elif c == "\n":
|
|
add("nl")
|
|
line += 1
|
|
i += 1
|
|
elif c in "\"'":
|
|
start = line
|
|
j = i + 1
|
|
closed = False
|
|
while j < n:
|
|
d = t[j]
|
|
if d == "\\":
|
|
if j + 1 < n and t[j + 1] == "\n":
|
|
line += 1
|
|
j += 2
|
|
continue
|
|
if d == "\n":
|
|
line += 1
|
|
if d == c:
|
|
closed = True
|
|
break
|
|
j += 1
|
|
raw = t[i + 1: min(j, n)]
|
|
if not closed:
|
|
trailing = len(raw) - len(raw.rstrip("\\"))
|
|
if trailing % 2 == 1:
|
|
raw += "\\"
|
|
add("str", decode_dq(raw) if c == '"' else decode_sq(raw), start, not closed)
|
|
i = j + 1 if closed else n
|
|
elif c in "$@" and i + 1 < n and is_start(t[i + 1]):
|
|
j = i + 1
|
|
while j < n and is_part(t[j]):
|
|
j += 1
|
|
if c == "$":
|
|
add("state", t[i:j])
|
|
else:
|
|
add("at", t[i + 1:j])
|
|
i = j
|
|
elif is_start(c):
|
|
j = i
|
|
while j < n and is_part(t[j]):
|
|
j += 1
|
|
w = t[i:j]
|
|
if w in ("true", "false", "null"):
|
|
add(w, w)
|
|
else:
|
|
add("ident", w)
|
|
i = j
|
|
elif is_digit(c) or (c == "-" and i + 1 < n and is_digit(t[i + 1]) and not (toks and ends_value(toks[-1]))):
|
|
j = i + 1 if c == "-" else i
|
|
while j < n and is_digit(t[j]):
|
|
j += 1
|
|
if j + 1 < n and t[j] == "." and is_digit(t[j + 1]):
|
|
j += 1
|
|
while j < n and is_digit(t[j]):
|
|
j += 1
|
|
bad = False
|
|
if j < n and t[j] in "eE":
|
|
j += 1
|
|
if j < n and t[j] in "+-":
|
|
j += 1
|
|
d0 = j
|
|
while j < n and is_digit(t[j]):
|
|
j += 1
|
|
bad = j == d0
|
|
text = t[i:j]
|
|
if bad:
|
|
add("badnum", text)
|
|
else:
|
|
toks.append({"k": "num", "v": float(text), "text": text, "line": line, "open": False})
|
|
i = j
|
|
else:
|
|
if t[i:i + 2] in TWO:
|
|
add("p", t[i:i + 2])
|
|
i += 2
|
|
continue
|
|
if c in ONE:
|
|
add("p", ONE[c])
|
|
i += 1
|
|
return toks
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.3 statements
|
|
|
|
def is_p(tok, v):
|
|
return tok["k"] == "p" and tok["v"] == v
|
|
|
|
|
|
def split(toks, finished):
|
|
out = []
|
|
pos, n = 0, len(toks)
|
|
while pos < n:
|
|
while pos < n and toks[pos]["k"] == "nl":
|
|
pos += 1
|
|
if pos >= n:
|
|
break
|
|
t0 = toks[pos]
|
|
if t0["k"] not in ("ident", "state") or pos + 1 >= n or not is_p(toks[pos + 1], "="):
|
|
while pos < n and toks[pos]["k"] != "nl":
|
|
pos += 1
|
|
continue
|
|
name, state, line = t0["v"], t0["k"] == "state", t0["line"]
|
|
pos += 2
|
|
expr = []
|
|
depth = 0
|
|
tern = [] # [in second branch, has operand]
|
|
complete = finished
|
|
boundary = False
|
|
while pos < n:
|
|
t = toks[pos]
|
|
if t["k"] == "nl":
|
|
if depth == 0 and not tern:
|
|
peek = pos + 1
|
|
while peek < n and toks[peek]["k"] == "nl":
|
|
peek += 1
|
|
if peek < n and is_p(toks[peek], "?"):
|
|
pos += 1
|
|
continue
|
|
boundary = True
|
|
if peek < n:
|
|
complete = True
|
|
break
|
|
pos += 1
|
|
continue
|
|
operand = False
|
|
if t["k"] == "p" and t["v"] in "([{":
|
|
depth += 1
|
|
elif t["k"] == "p" and t["v"] in ")]}":
|
|
if depth > 0:
|
|
depth -= 1
|
|
operand = depth == 0
|
|
elif is_p(t, "?") and depth == 0:
|
|
tern.append([False, False])
|
|
elif is_p(t, ":") and depth == 0:
|
|
if tern and not tern[-1][0]:
|
|
tern[-1] = [True, False]
|
|
elif t["k"] in VALUE_KINDS:
|
|
operand = depth == 0
|
|
if operand and tern:
|
|
tern[-1][1] = True
|
|
if tern[-1][0]:
|
|
tern.pop()
|
|
expr.append(t)
|
|
pos += 1
|
|
opened = (not boundary) and (depth > 0 or (bool(expr) and expr[-1]["open"]))
|
|
out.append({"name": name, "state": state, "line": line, "toks": expr, "complete": complete, "open": opened})
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.4 expressions
|
|
|
|
def close(toks):
|
|
out, stack = [], []
|
|
pairs = {"(": ")", "[": "]", "{": "}"}
|
|
for t in toks:
|
|
if t["k"] == "p" and t["v"] in pairs:
|
|
stack.append(pairs[t["v"]])
|
|
elif t["k"] == "p" and t["v"] in ")]}":
|
|
if stack and stack[-1] == t["v"]:
|
|
stack.pop()
|
|
else:
|
|
continue
|
|
out.append(t)
|
|
line = toks[-1]["line"] if toks else 0
|
|
while stack:
|
|
out.append({"k": "p", "v": stack.pop(), "line": line, "open": False})
|
|
return out
|
|
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, line):
|
|
self.line = line
|
|
|
|
|
|
class P:
|
|
def __init__(self, toks):
|
|
self.t, self.i = toks, 0
|
|
|
|
def peek(self):
|
|
return self.t[self.i] if self.i < len(self.t) else None
|
|
|
|
def at(self, v):
|
|
x = self.peek()
|
|
return x is not None and is_p(x, v)
|
|
|
|
def line(self):
|
|
x = self.peek()
|
|
if x is None:
|
|
x = self.t[-1] if self.t else None
|
|
return x["line"] if x else 0
|
|
|
|
def stop(self):
|
|
raise Stop(self.line())
|
|
|
|
def expect(self, v):
|
|
if not self.at(v):
|
|
self.stop()
|
|
self.i += 1
|
|
|
|
def ternary(self):
|
|
c = self.binary(0)
|
|
if self.at("?"):
|
|
self.i += 1
|
|
a = self.ternary()
|
|
self.expect(":")
|
|
b = self.ternary()
|
|
return {"e": "ternary", "line": c["line"]}
|
|
return c
|
|
|
|
LEVELS = [("||",), ("&&",), ("==", "!="), (">", "<", ">=", "<="), ("+", "-"), ("*", "/", "%")]
|
|
|
|
def binary(self, lvl):
|
|
if lvl == len(self.LEVELS):
|
|
return self.unary()
|
|
left = self.binary(lvl + 1)
|
|
while True:
|
|
x = self.peek()
|
|
if x is None or x["k"] != "p" or x["v"] not in self.LEVELS[lvl]:
|
|
return left
|
|
op = x["v"]
|
|
self.i += 1
|
|
self.binary(lvl + 1)
|
|
left = {"e": "binary", "op": op, "line": left["line"]}
|
|
|
|
def unary(self):
|
|
if self.at("!") or self.at("-"):
|
|
line = self.line()
|
|
self.i += 1
|
|
self.unary()
|
|
return {"e": "unary", "line": line}
|
|
return self.postfix()
|
|
|
|
def postfix(self):
|
|
e = self.primary()
|
|
while True:
|
|
if self.at("."):
|
|
self.i += 1
|
|
self.key()
|
|
e = {"e": "member", "line": e["line"]}
|
|
elif self.at("["):
|
|
self.i += 1
|
|
self.ternary()
|
|
self.expect("]")
|
|
e = {"e": "index", "line": e["line"]}
|
|
else:
|
|
return e
|
|
|
|
def key(self):
|
|
x = self.peek()
|
|
if x is None:
|
|
self.stop()
|
|
self.i += 1
|
|
k = x["k"]
|
|
if k in ("ident", "str"):
|
|
return x["v"]
|
|
if k == "state":
|
|
return x["v"][1:]
|
|
if k == "num":
|
|
return x["text"]
|
|
if k == "badnum":
|
|
return x["v"]
|
|
if k in ("true", "false", "null"):
|
|
return k
|
|
if k == "at":
|
|
return "@" + x["v"]
|
|
return "?"
|
|
|
|
def items(self, closer):
|
|
v = []
|
|
while True:
|
|
while self.at(","):
|
|
self.i += 1
|
|
if self.at(closer):
|
|
self.i += 1
|
|
return v
|
|
if self.peek() is None:
|
|
self.stop()
|
|
v.append(self.ternary())
|
|
|
|
def primary(self):
|
|
x = self.peek()
|
|
if x is None:
|
|
self.stop()
|
|
line = x["line"]
|
|
k = x["k"]
|
|
self.i += 1
|
|
if k == "str":
|
|
return {"e": "str", "v": x["v"], "line": line}
|
|
if k == "num":
|
|
return {"e": "num", "v": x["v"], "line": line}
|
|
if k == "badnum":
|
|
self.i -= 1
|
|
self.stop()
|
|
if k in ("true", "false"):
|
|
return {"e": "bool", "v": k == "true", "line": line}
|
|
if k == "null":
|
|
return {"e": "null", "line": line}
|
|
if k == "p" and x["v"] == "[":
|
|
return {"e": "list", "items": self.items("]"), "line": line}
|
|
if k == "p" and x["v"] == "{":
|
|
entries = []
|
|
while True:
|
|
while self.at(","):
|
|
self.i += 1
|
|
if self.at("}"):
|
|
self.i += 1
|
|
break
|
|
if self.peek() is None:
|
|
self.stop()
|
|
kk = self.key()
|
|
self.expect(":")
|
|
entries.append((kk, self.ternary()))
|
|
return {"e": "object", "entries": entries, "line": line}
|
|
if k == "p" and x["v"] == "(":
|
|
e = self.ternary()
|
|
self.expect(")")
|
|
return e
|
|
if k == "ident":
|
|
if self.at("(") and "A" <= x["v"][0] <= "Z":
|
|
self.i += 1
|
|
return {"e": "call", "at": False, "name": x["v"], "args": self.items(")"), "line": line}
|
|
return {"e": "ref", "name": x["v"], "line": line}
|
|
if k == "at":
|
|
self.expect("(")
|
|
return {"e": "call", "at": True, "name": x["v"], "args": self.items(")"), "line": line}
|
|
if k == "state":
|
|
if self.at("="):
|
|
self.i += 1
|
|
self.ternary()
|
|
return {"e": "assign", "line": line}
|
|
return {"e": "state", "name": x["v"], "line": line}
|
|
self.i -= 1
|
|
self.stop()
|
|
|
|
|
|
def parse_tokens(toks):
|
|
if not toks:
|
|
raise Stop(None)
|
|
p = P(toks)
|
|
e = p.ternary()
|
|
if p.i < len(toks):
|
|
p.stop()
|
|
return e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.5 into a joule-ui tree
|
|
|
|
BUILTINS = {"Count", "First", "Last", "Sum", "Avg", "Min", "Max", "Filter", "Sort", "Round", "Abs", "Floor", "Ceil", "Each"}
|
|
STATE_ACTIONS = {"Set", "Reset", "Run"}
|
|
|
|
|
|
def T(v):
|
|
return {"type": "text", "value": v}
|
|
|
|
|
|
class Ingest:
|
|
def __init__(self, lib, finished, lines):
|
|
self.reg = lib["widgets"]
|
|
self.lib = lib
|
|
self.finished = finished
|
|
self.lines = lines
|
|
self.errors = []
|
|
self.unresolved = set()
|
|
self.visited = set()
|
|
self.resolving = []
|
|
self.occ = {}
|
|
self.binds = {}
|
|
|
|
def src(self, line):
|
|
return self.lines[line] if 0 <= line < len(self.lines) else 0
|
|
|
|
def err(self, stmt, line, code, component=None, prop=None):
|
|
self.errors.append({"line": self.src(line), "statement": stmt, "code": code, "component": component, "prop": prop, "warning": False})
|
|
|
|
def reference(self, name):
|
|
b = self.binds.get(name)
|
|
if b is None or name in self.resolving:
|
|
self.unresolved.add(name)
|
|
return ("unresolved",)
|
|
if b["kind"] != "value":
|
|
return ("refused",)
|
|
self.visited.add(name)
|
|
self.occ[name] = self.occ.get(name, 0) + 1
|
|
key = name if self.occ[name] == 1 else "%s#%d" % (name, self.occ[name])
|
|
self.resolving.append(name)
|
|
v = self.eval(b["expr"], name, key)
|
|
self.resolving.pop()
|
|
return v
|
|
|
|
def eval(self, e, stmt, key):
|
|
k = e["e"]
|
|
if k == "str":
|
|
return ("data", T(e["v"]))
|
|
if k == "num":
|
|
return ("data", {"type": "number", "value": e["v"]})
|
|
if k == "bool":
|
|
return ("data", {"type": "bool", "value": e["v"]})
|
|
if k == "null":
|
|
return ("null",)
|
|
if k == "ref":
|
|
return self.reference(e["name"])
|
|
if k in ("state", "unary", "binary", "ternary", "member", "index", "assign"):
|
|
self.err(stmt, e["line"], "not-declarative")
|
|
return ("refused",)
|
|
if k == "list":
|
|
return ("list", [self.eval(it, stmt, "%s/%d" % (key, i)) for i, it in enumerate(e["items"])])
|
|
if k == "object":
|
|
return ("object", [(kk, self.eval(v, stmt, "%s/%s" % (key, kk))) for kk, v in e["entries"]])
|
|
# a call
|
|
name, args = e["name"], e["args"]
|
|
if not e["at"]:
|
|
if name in ("Query", "Mutation"):
|
|
self.err(stmt, e["line"], "inline-reserved")
|
|
return ("refused",)
|
|
if name == "Action":
|
|
return self.action_list(args, stmt, e["line"], key)
|
|
return self.build(name, args, e["line"], stmt, key)
|
|
if name in BUILTINS or name in STATE_ACTIONS:
|
|
self.err(stmt, e["line"], "not-declarative")
|
|
return ("refused",)
|
|
if name == "Action":
|
|
return self.action_list(args, stmt, e["line"], key)
|
|
items, not_yet, refused = [], False, False
|
|
for i, a in enumerate(args):
|
|
r = self.data(self.eval(a, stmt, "%s/%d" % (key, i)), stmt, a["line"], True)
|
|
if r == "not-yet":
|
|
not_yet = True
|
|
elif r == "refused":
|
|
refused = True
|
|
else:
|
|
items.append(r)
|
|
if refused:
|
|
return ("refused",)
|
|
if not_yet:
|
|
return ("held",)
|
|
payload = None if not items else items[0] if len(items) == 1 else {"type": "list", "value": items}
|
|
return ("action", {"type": "action", "value": {"id": name, "payload": payload}})
|
|
|
|
def action_list(self, args, stmt, line, key):
|
|
if len(args) == 1 and args[0]["e"] == "list" and len(args[0]["items"]) == 1:
|
|
one = args[0]["items"][0]
|
|
v = self.eval(one, stmt, key + "/0")
|
|
if v[0] == "action":
|
|
return v
|
|
if v[0] in ("unresolved", "held"):
|
|
return ("held",)
|
|
if v[0] == "refused":
|
|
return ("refused",)
|
|
self.err(stmt, one["line"], "type-mismatch")
|
|
return ("refused",)
|
|
self.err(stmt, line, "not-declarative")
|
|
return ("refused",)
|
|
|
|
def data(self, v, stmt, line, keep_null):
|
|
"""A value placed as data: a PropValue, "not-yet" or "refused"."""
|
|
k = v[0]
|
|
if k in ("data", "action"):
|
|
return v[1]
|
|
if k == "null":
|
|
return {"type": "null"} if keep_null else "not-yet"
|
|
if k == "list":
|
|
out = []
|
|
for it in v[1]:
|
|
r = self.data(it, stmt, line, True)
|
|
if isinstance(r, dict):
|
|
out.append(r)
|
|
return {"type": "list", "value": out}
|
|
if k == "object":
|
|
out = {}
|
|
for kk, it in v[1]:
|
|
r = self.data(it, stmt, line, True)
|
|
if isinstance(r, dict):
|
|
out[kk] = r
|
|
return {"type": "object", "value": out}
|
|
if k == "widget":
|
|
self.err(stmt, line, "type-mismatch", v[1]["name"])
|
|
return "refused"
|
|
if k in ("unresolved", "held"):
|
|
return "not-yet"
|
|
return "refused"
|
|
|
|
def build(self, name, args, line, stmt, key):
|
|
s = self.reg.get(name)
|
|
if s is None:
|
|
self.err(stmt, line, "unknown-component", name)
|
|
return ("refused",)
|
|
order = s["positional"]
|
|
if len(args) > len(order):
|
|
self.err(stmt, args[len(order)]["line"], "excess-args", name)
|
|
props, absent, children = {}, {}, []
|
|
for slot, arg in zip(order, args):
|
|
if slot == "children":
|
|
v = self.eval(arg, stmt, key)
|
|
if v[0] == "list":
|
|
for it in v[1]:
|
|
if it[0] == "widget":
|
|
children.append((it[1], it[2]))
|
|
elif it[0] in ("unresolved", "held", "null", "refused"):
|
|
pass
|
|
else:
|
|
self.err(stmt, arg["line"], "type-mismatch", name, "children")
|
|
elif v[0] in ("null", "unresolved", "held", "refused"):
|
|
pass
|
|
else:
|
|
self.err(stmt, arg["line"], "type-mismatch", name, "children")
|
|
continue
|
|
v = self.eval(arg, stmt, "%s/%s" % (key, slot))
|
|
if v[0] == "null":
|
|
absent[slot] = "null"
|
|
elif v[0] in ("unresolved", "held"):
|
|
absent[slot] = "not-yet"
|
|
elif v[0] == "refused":
|
|
absent[slot] = "refused"
|
|
elif v[0] == "widget":
|
|
self.err(stmt, arg["line"], "type-mismatch", name, slot)
|
|
absent[slot] = "refused"
|
|
else:
|
|
r = self.data(v, stmt, arg["line"], False)
|
|
if isinstance(r, dict):
|
|
props[slot] = r
|
|
else:
|
|
absent[slot] = r
|
|
w = {"name": name, "props": props}
|
|
if children:
|
|
w["children"] = [c[0] for c in children]
|
|
child_nodes = [c[1] for c in children]
|
|
while True:
|
|
errs = validate(w, self.reg)
|
|
if not errs:
|
|
break
|
|
if not self.finished and any(x["kind"] == "missing_prop" and x["path"] == "$" and absent.get(x["prop"]) == "not-yet" for x in errs):
|
|
return ("held",)
|
|
drop = False
|
|
drop_children = []
|
|
before = len(w["props"])
|
|
for x in errs:
|
|
kind, path = x["kind"], x["path"]
|
|
if kind == "missing_prop" and path == "$":
|
|
why = absent.get(x["prop"])
|
|
if why == "null":
|
|
self.err(stmt, line, "null-required", name, x["prop"])
|
|
elif why != "refused":
|
|
self.err(stmt, line, "missing-required", name, x["prop"])
|
|
drop = True
|
|
elif kind == "wrong_prop_type" and path == "$":
|
|
self.err(stmt, line, "type-mismatch", name, x["prop"])
|
|
if s["props"][x["prop"]].get("required"):
|
|
drop = True
|
|
else:
|
|
del w["props"][x["prop"]]
|
|
elif kind == "extra_prop" and path == "$":
|
|
self.err(stmt, line, "type-mismatch", name, x["prop"])
|
|
del w["props"][x["prop"]]
|
|
elif kind == "disallowed_child":
|
|
idx = child_index(path)
|
|
if idx is None:
|
|
drop = True
|
|
continue
|
|
first = child_nodes[idx][0] if idx < len(child_nodes) and child_nodes[idx] else None
|
|
self.errors.append({"line": first["line"] if first and first["line"] > 0 else self.src(line),
|
|
"statement": first["statement"] if first else stmt,
|
|
"code": "disallowed-child", "component": x["child"], "prop": None, "warning": False})
|
|
drop_children.append(idx)
|
|
else:
|
|
drop = True
|
|
if drop:
|
|
return ("refused",)
|
|
drop_children = sorted(set(drop_children))
|
|
changed = bool(drop_children) or len(w["props"]) != before
|
|
for i in reversed(drop_children):
|
|
del w["children"][i]
|
|
del child_nodes[i]
|
|
if "children" in w and not w["children"]:
|
|
del w["children"]
|
|
if not changed:
|
|
return ("refused",)
|
|
nodes = [{"key": key, "line": self.src(line), "statement": stmt}]
|
|
for cn in child_nodes:
|
|
nodes.extend(cn)
|
|
return ("widget", w, nodes)
|
|
|
|
|
|
def child_index(path):
|
|
pre = "$.children["
|
|
if not path.startswith(pre) or not path.endswith("]"):
|
|
return None
|
|
mid = path[len(pre):-1]
|
|
return int(mid) if mid.isdigit() else None
|
|
|
|
|
|
def snapshot(lib, text, finished):
|
|
body, lines = prepare(text)
|
|
toks = lex(body)
|
|
raws = split(toks, finished)
|
|
incomplete = bool(raws) and raws[-1]["open"]
|
|
ing = Ingest(lib, finished, lines)
|
|
src = ing.src
|
|
errors, statements, binds = [], [], {}
|
|
for order, r in enumerate(raws):
|
|
tk = close(r["toks"]) if r["open"] else r["toks"]
|
|
line = src(r["line"])
|
|
try:
|
|
e = parse_tokens(tk)
|
|
except Stop as stop:
|
|
if r["complete"]:
|
|
at = line if stop.line is None or not tk else src(stop.line)
|
|
errors.append({"line": at, "statement": r["name"], "code": "syntax", "component": None, "prop": None, "warning": False})
|
|
statements.append({"name": r["name"], "line": line, "kind": "syntax", "complete": True})
|
|
continue
|
|
if e["e"] == "call" and not e["at"] and e["name"] == "Query":
|
|
kind = "query"
|
|
elif e["e"] == "call" and not e["at"] and e["name"] == "Mutation":
|
|
kind = "mutation"
|
|
elif r["state"]:
|
|
kind = "state"
|
|
else:
|
|
kind = "value"
|
|
statements.append({"name": r["name"], "line": line, "kind": kind, "complete": r["complete"]})
|
|
if r["complete"] and kind != "value":
|
|
errors.append({"line": line, "statement": r["name"], "code": "not-declarative", "component": None, "prop": None, "warning": False})
|
|
prev = binds.get(r["name"])
|
|
if prev is not None and prev["complete"] and not r["complete"] and r["open"]:
|
|
continue
|
|
binds[r["name"]] = {"line": line, "complete": r["complete"], "kind": kind, "expr": e, "order": order}
|
|
ing.binds = binds
|
|
entry = None
|
|
if "root" in binds:
|
|
entry = ("root", False)
|
|
elif lib.get("root"):
|
|
vals = sorted((b["order"], n) for n, b in binds.items() if b["kind"] == "value")
|
|
for _, n in vals:
|
|
ex = binds[n]["expr"]
|
|
if ex["e"] == "call" and not ex["at"] and ex["name"] == lib["root"]:
|
|
entry = (n, True)
|
|
break
|
|
tree, nodes = None, []
|
|
if entry:
|
|
v = ing.reference(entry[0])
|
|
if v[0] == "widget":
|
|
tree, nodes = v[1], v[2]
|
|
errors.extend(ing.errors)
|
|
if finished:
|
|
if entry is None:
|
|
code = "parse-failed" if not statements and body != "" else "no-root"
|
|
errors.append({"line": 0, "statement": "", "code": code, "component": None, "prop": None, "warning": False})
|
|
else:
|
|
line = binds[entry[0]]["line"]
|
|
if tree is None:
|
|
errors.append({"line": line, "statement": entry[0], "code": "no-root", "component": None, "prop": None, "warning": False})
|
|
elif entry[1]:
|
|
errors.append({"line": line, "statement": entry[0], "code": "no-root", "component": None, "prop": None, "warning": True})
|
|
if not finished:
|
|
errors = [x for x in errors if x["statement"] not in binds or binds[x["statement"]]["complete"]]
|
|
|
|
def opt(v):
|
|
return (0, "") if v is None else (1, v)
|
|
|
|
def ekey(x):
|
|
return (x["line"], x["statement"], x["code"], opt(x["component"]), opt(x["prop"]), x["warning"])
|
|
|
|
errors.sort(key=ekey)
|
|
kept = []
|
|
for x in errors:
|
|
if not kept or ekey(kept[-1]) != ekey(x):
|
|
kept.append(x)
|
|
orphans = sorted(n for n, b in binds.items() if b["kind"] == "value" and n not in ing.visited) if entry else []
|
|
if tree is not None and validate(tree, ing.reg):
|
|
tree, nodes = None, []
|
|
return {"tree": tree, "nodes": nodes, "errors": kept, "unresolved": sorted(ing.unresolved), "orphans": orphans,
|
|
"incomplete": incomplete, "statements": statements}
|
|
|
|
|
|
def stated(s):
|
|
"""A snapshot as the pack states it (§11.9)."""
|
|
errs = []
|
|
for x in s["errors"]:
|
|
e = {"line": x["line"], "statement": x["statement"], "code": x["code"]}
|
|
if x["component"] is not None:
|
|
e["component"] = x["component"]
|
|
if x["prop"] is not None:
|
|
e["prop"] = x["prop"]
|
|
if x["warning"]:
|
|
e["warning"] = True
|
|
errs.append(e)
|
|
return {"tree": s["tree"], "nodes": s["nodes"], "errors": errs, "unresolved": s["unresolved"],
|
|
"orphans": s["orphans"], "incomplete": s["incomplete"]}
|
|
|
|
|
|
class Stream:
|
|
def __init__(self, lib):
|
|
self.lib, self.text = lib, ""
|
|
|
|
def push(self, chunk):
|
|
self.text += chunk
|
|
return stated(snapshot(self.lib, self.text, False))
|
|
|
|
def finish(self):
|
|
return stated(snapshot(self.lib, self.text, True))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §11.7 the canonical form
|
|
|
|
def is_name(s):
|
|
return bool(s) and is_start(s[0]) and all(is_part(c) for c in s)
|
|
|
|
|
|
def jstr(s):
|
|
out = ['"']
|
|
for ch in s:
|
|
o = ord(ch)
|
|
if ch == '"':
|
|
out.append('\\"')
|
|
elif ch == "\\":
|
|
out.append("\\\\")
|
|
elif ch == "\b":
|
|
out.append("\\b")
|
|
elif ch == "\f":
|
|
out.append("\\f")
|
|
elif ch == "\n":
|
|
out.append("\\n")
|
|
elif ch == "\r":
|
|
out.append("\\r")
|
|
elif ch == "\t":
|
|
out.append("\\t")
|
|
elif o < 0x20:
|
|
out.append("\\u%04x" % o)
|
|
else:
|
|
out.append(ch)
|
|
out.append('"')
|
|
return "".join(out)
|
|
|
|
|
|
def num_text(x):
|
|
"""The shortest decimal that reads back to x, positional, no exponent, no `.0`."""
|
|
if x != x or x in (float("inf"), float("-inf")):
|
|
raise ValueError("not finite")
|
|
t = format(Decimal(repr(x)), "f")
|
|
if "." in t:
|
|
t = t.rstrip("0").rstrip(".")
|
|
return t
|
|
|
|
|
|
RESERVED_IDS = {"Action"} | BUILTINS | STATE_ACTIONS
|
|
|
|
|
|
def val(v):
|
|
t = v["type"]
|
|
if t == "null":
|
|
return "null"
|
|
if t == "bool":
|
|
return "true" if v["value"] else "false"
|
|
if t == "number":
|
|
return num_text(v["value"])
|
|
if t == "text":
|
|
return jstr(v["value"])
|
|
if t == "list":
|
|
return "[" + ", ".join(val(i) for i in v["value"]) + "]"
|
|
if t == "object":
|
|
parts = []
|
|
for k in sorted(v["value"]):
|
|
kk = k if is_name(k) and k not in ("true", "false", "null") else jstr(k)
|
|
parts.append("%s: %s" % (kk, val(v["value"][k])))
|
|
return "{" + ", ".join(parts) + "}"
|
|
if t == "action":
|
|
aid, payload = v["value"]["id"], v["value"]["payload"]
|
|
if not is_name(aid) or aid in RESERVED_IDS:
|
|
raise ValueError("action id")
|
|
if payload is None:
|
|
inner = ""
|
|
elif payload["type"] == "list" and len(payload["value"]) >= 2:
|
|
inner = ", ".join(val(i) for i in payload["value"])
|
|
else:
|
|
inner = val(payload)
|
|
return "@%s(%s)" % (aid, inner)
|
|
raise ValueError(t)
|
|
|
|
|
|
def write(tree, lib):
|
|
reg = lib["widgets"]
|
|
if validate(tree, reg):
|
|
raise ValueError("invalid")
|
|
flat = []
|
|
|
|
def walk(w):
|
|
flat.append(w)
|
|
for c in w.get("children", []):
|
|
walk(c)
|
|
|
|
walk(tree)
|
|
names, counts = [], {}
|
|
for i, w in enumerate(flat):
|
|
if i == 0:
|
|
names.append("root")
|
|
continue
|
|
base = w["name"][0].lower() + w["name"][1:]
|
|
counts[base] = counts.get(base, 0) + 1
|
|
names.append("%s%d" % (base, counts[base]))
|
|
out = []
|
|
idx = [0]
|
|
|
|
def size(w):
|
|
return 1 + sum(size(c) for c in w.get("children", []))
|
|
|
|
def emit(w):
|
|
me = idx[0]
|
|
idx[0] += 1
|
|
if not ("A" <= w["name"][0] <= "Z") or w["name"] in ("Query", "Mutation", "Action") or not is_name(w["name"]):
|
|
raise ValueError("not callable")
|
|
s = reg[w["name"]]
|
|
order = s["positional"]
|
|
for p in w.get("props", {}):
|
|
if p not in order:
|
|
raise ValueError("not positional")
|
|
child_names, at = [], me + 1
|
|
for c in w.get("children", []):
|
|
child_names.append(names[at])
|
|
at += size(c)
|
|
args = []
|
|
for slot in order:
|
|
if slot == "children":
|
|
args.append("[" + ", ".join(child_names) + "]" if child_names else None)
|
|
continue
|
|
pv = w.get("props", {}).get(slot)
|
|
if pv is None:
|
|
args.append(None)
|
|
elif pv["type"] == "null":
|
|
raise ValueError("null prop")
|
|
else:
|
|
args.append(val(pv))
|
|
while args and args[-1] is None:
|
|
args.pop()
|
|
out.append("%s = %s(%s)\n" % (names[me], w["name"], ", ".join("null" if a is None else a for a in args)))
|
|
for c in w.get("children", []):
|
|
emit(c)
|
|
|
|
emit(tree)
|
|
return "".join(out)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the pack
|
|
|
|
def load(d):
|
|
files = {}
|
|
for n in sorted(os.listdir(d)):
|
|
with open(os.path.join(d, n), "rb") as f:
|
|
files[n] = f.read()
|
|
return files
|
|
|
|
|
|
def norm_tree(t):
|
|
"""A tree for comparison: JSON numbers as floats."""
|
|
return json.loads(json.dumps(t)) if t is not None else None
|
|
|
|
|
|
def check(files):
|
|
"""Every disagreement, as a list of lines; empty when the pack agrees."""
|
|
bad = []
|
|
man = json.loads(files["MANIFEST.json"])
|
|
for n, digest in man["files"].items():
|
|
if n not in files:
|
|
bad.append("MANIFEST names %s, which the pack lacks" % n)
|
|
elif "sha256:" + hashlib.sha256(files[n]).hexdigest() != digest:
|
|
bad.append("%s: its SHA-256 is not the manifest's" % n)
|
|
for n in files:
|
|
if n != "MANIFEST.json" and n not in man["files"]:
|
|
bad.append("%s is not in the manifest" % n)
|
|
if bad:
|
|
# A file the manifest does not name as it is: nothing in it is relied on.
|
|
return bad
|
|
lib = json.loads(files["library.json"])
|
|
|
|
def same(a, b):
|
|
return json.loads(json.dumps(a, sort_keys=True)) == json.loads(json.dumps(b, sort_keys=True))
|
|
|
|
for c in json.loads(files["l1-statements.json"])["cases"]:
|
|
s = snapshot(lib, c["input"], True)
|
|
if not same(s["statements"], c["statements"]) or s["incomplete"] != c["incomplete"]:
|
|
bad.append("L1 %s: got %s" % (c["name"], json.dumps(s["statements"])))
|
|
l2 = {}
|
|
for c in json.loads(files["l2-ingest.json"])["cases"]:
|
|
s = stated(snapshot(lib, c["input"], True))
|
|
l2[c["name"]] = s
|
|
if not same(s, c["expected"]):
|
|
bad.append("L2 %s: got %s" % (c["name"], json.dumps(s)[:600]))
|
|
for c in json.loads(files["l3-stream.json"])["cases"]:
|
|
st = Stream(lib)
|
|
for i, ch in enumerate(c["chunks"]):
|
|
s = st.push(ch)
|
|
if not same(s, c["after_each"][i]):
|
|
bad.append("L3 %s after chunk %d: got %s" % (c["name"], i, json.dumps(s)[:600]))
|
|
end = st.finish()
|
|
if not same(end, c["at_end"]):
|
|
bad.append("L3 %s at the end: got %s" % (c["name"], json.dumps(end)[:600]))
|
|
if not same(end, stated(snapshot(lib, "".join(c["chunks"]), True))):
|
|
bad.append("L3 %s: the end of the stream is not the batch parse" % c["name"])
|
|
for c in json.loads(files["l4-write.json"])["cases"]:
|
|
try:
|
|
text = write(c["tree"], lib)
|
|
except ValueError as e:
|
|
bad.append("L4 %s: not writable (%s)" % (c["name"], e))
|
|
continue
|
|
if text != c["text"]:
|
|
bad.append("L4 %s: wrote %r" % (c["name"], text))
|
|
back = snapshot(lib, c["text"], True)
|
|
if not same(back["tree"], c["tree"]) or back["errors"]:
|
|
bad.append("L4 %s: the text does not parse back to the tree" % c["name"])
|
|
for c in json.loads(files["l5-refusals.json"])["cases"]:
|
|
s = stated(snapshot(lib, c["input"], True))
|
|
if not any(e["code"] == c["code"] and e["line"] == c["line"] for e in s["errors"]):
|
|
bad.append("L5 %s: not refused %s at line %d" % (c["name"], c["code"], c["line"]))
|
|
if not same(s, c["expected"]):
|
|
bad.append("L5 %s: got %s" % (c["name"], json.dumps(s)[:600]))
|
|
return bad
|
|
|
|
|
|
def mutations(files):
|
|
"""Changed vectors, each of which the verifier must refuse."""
|
|
def edit(name, f):
|
|
doc = json.loads(files[name])
|
|
f(doc)
|
|
out = dict(files)
|
|
out[name] = (json.dumps(doc, indent=2, sort_keys=True) + "\n").encode()
|
|
# keep the manifest true, so that the content check is what refuses it
|
|
man = json.loads(out["MANIFEST.json"])
|
|
man["files"][name] = "sha256:" + hashlib.sha256(out[name]).hexdigest()
|
|
out["MANIFEST.json"] = (json.dumps(man, indent=2, sort_keys=True) + "\n").encode()
|
|
return out
|
|
|
|
def case(doc, n):
|
|
return next(c for c in doc["cases"] if c["name"] == n)
|
|
|
|
def first_text(t):
|
|
if t and "props" in t:
|
|
for k, v in t["props"].items():
|
|
if v.get("type") == "text":
|
|
return t["props"], k
|
|
for c in (t or {}).get("children", []):
|
|
r = first_text(c)
|
|
if r:
|
|
return r
|
|
return None
|
|
|
|
def change_text(doc, n, key="expected"):
|
|
props, k = first_text(case(doc, n)[key]["tree"])
|
|
props[k]["value"] += "!"
|
|
|
|
flipped = bytearray(files["l2-ingest.json"])
|
|
flipped[100] ^= 0x01
|
|
yield "a flipped byte under the manifest", dict(files, **{"l2-ingest.json": bytes(flipped)})
|
|
yield "L1: a statement's line moved", edit("l1-statements.json", lambda d: d["cases"][0]["statements"][1].__setitem__("line", 9))
|
|
yield "L1: complete flipped", edit("l1-statements.json", lambda d: d["cases"][0]["statements"][0].__setitem__("complete", False))
|
|
yield "L2: a text changed", edit("l2-ingest.json", lambda d: change_text(d, "hoisting"))
|
|
yield "L2: an unresolved name dropped", edit("l2-ingest.json", lambda d: case(d, "unresolved-left-out")["expected"]["unresolved"].clear())
|
|
yield "L2: an orphan dropped", edit("l2-ingest.json", lambda d: case(d, "orphans")["expected"]["orphans"].clear())
|
|
yield "L2: a key renamed", edit("l2-ingest.json", lambda d: case(d, "used-twice")["expected"]["nodes"][2].__setitem__("key", "t"))
|
|
yield "L3: a pending title already whole", edit("l3-stream.json", lambda d: change_text(d, "pending-title-grows", "at_end"))
|
|
yield "L3: an error not withheld", edit("l3-stream.json", lambda d: case(d, "errors-wait-for-completion")["after_each"][0]["errors"].append({"line": 3, "statement": "b", "code": "type-mismatch", "component": "Text", "prop": "text"}))
|
|
yield "L4: the canonical text changed", edit("l4-write.json", lambda d: d["cases"][0].__setitem__("text", d["cases"][0]["text"].replace("card1", "c1", 1)))
|
|
yield "L5: a refusal's line moved", edit("l5-refusals.json", lambda d: case(d, "unknown-component").__setitem__("line", 2))
|
|
yield "L5: a refused widget kept", edit("l5-refusals.json", lambda d: case(d, "unknown-component")["expected"]["tree"]["children"].append({"name": "Chartt", "props": {}}))
|
|
yield "L5: a code changed", edit("l5-refusals.json", lambda d: case(d, "disallowed-child")["expected"]["errors"][0].__setitem__("code", "type-mismatch"))
|
|
|
|
|
|
def main(argv):
|
|
mut = "--mutations" in argv
|
|
args = [a for a in argv if a != "--mutations"]
|
|
d = args[0] if args else DEFAULT_PACK
|
|
files = load(d)
|
|
if mut:
|
|
missed = 0
|
|
n = 0
|
|
for what, changed in mutations(files):
|
|
n += 1
|
|
if check(changed):
|
|
print("refused: %s" % what)
|
|
else:
|
|
print("NOT REFUSED: %s" % what)
|
|
missed += 1
|
|
if check(files):
|
|
print("the unchanged pack does not verify")
|
|
return 1
|
|
print("%d mutation(s), %d not refused" % (n, missed))
|
|
return 1 if missed else 0
|
|
bad = check(files)
|
|
for b in bad:
|
|
print(b)
|
|
counts = {n: len(json.loads(files[n])["cases"]) for n in files if n[:1] == "l" and n[1:2].isdigit()}
|
|
print("joule-ui-lang/v1: %s; %d disagreement(s)" % (", ".join("%s %d" % (k.split(".")[0], v) for k, v in sorted(counts.items())), len(bad)))
|
|
return 1 if bad else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|