#!/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 [] verify the pack python3 -I lang_verify.py --mutations [] 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:]))