Kamil Patryk Kozakowski
b158179559
Resolve property placeholders before call-chain linking, skip phantom routing states and lifecycle triggers from transition matching, dedupe exported states, and add golden JSON audit regression plus HTML lifecycle endpoint display. Co-authored-by: Cursor <cursoragent@cursor.com>
169 lines
6.8 KiB
Python
169 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Deep audit of golden JSON analysis exports."""
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
GOLDEN = Path(__file__).resolve().parent.parent / "src/test/resources/golden"
|
|
PH = re.compile(r"\$\{")
|
|
|
|
|
|
def fn_form(identifier):
|
|
if not identifier:
|
|
return identifier
|
|
last = identifier.rfind(".")
|
|
if last <= 0:
|
|
return identifier
|
|
prev = identifier.rfind(".", 0, last)
|
|
return identifier[prev + 1 :] if prev > 0 else identifier
|
|
|
|
|
|
def sid(st):
|
|
if isinstance(st, dict):
|
|
return st.get("fullIdentifier") or st.get("rawName")
|
|
return st
|
|
|
|
|
|
def eid(ev):
|
|
if isinstance(ev, dict):
|
|
return ev.get("fullIdentifier") or ev.get("rawName")
|
|
return ev
|
|
|
|
|
|
def audit_file(path, data):
|
|
name = data.get("name", path.stem)
|
|
issues = []
|
|
st_fqn = data.get("stateTypeFqn")
|
|
ev_fqn = data.get("eventTypeFqn")
|
|
transitions = data.get("transitions") or []
|
|
starts = list(data.get("startStates") or [])
|
|
ends = list(data.get("endStates") or [])
|
|
states = data.get("states") or []
|
|
meta = data.get("metadata") or {}
|
|
|
|
is_pkg_enum = st_fqn and "." in st_fqn and st_fqn not in ("String", "java.lang.String")
|
|
is_string = st_fqn in (None, "String", "java.lang.String") or (
|
|
not st_fqn and not ev_fqn
|
|
)
|
|
|
|
if PH.search(json.dumps(data)):
|
|
issues.append("BUG: unresolved ${placeholder} in export")
|
|
|
|
# Duplicate fullIdentifiers in states[]
|
|
by_full = defaultdict(list)
|
|
for s in states:
|
|
by_full[sid(s)].append(s.get("rawName"))
|
|
for fid, raws in by_full.items():
|
|
if fid and len(raws) > 1 and len(set(raws)) > 1:
|
|
issues.append(f"NOISE: duplicate state entry {fid!r} with rawNames {raws}")
|
|
|
|
# Collect transition state/event ids
|
|
t_events, t_sources, t_targets = set(), set(), set()
|
|
for t in transitions:
|
|
if eid(t.get("event")):
|
|
t_events.add(eid(t.get("event")))
|
|
for s in t.get("sourceStates") or []:
|
|
t_sources.add(sid(s))
|
|
for s in t.get("targetStates") or []:
|
|
t_targets.add(sid(s))
|
|
|
|
graph = t_sources | t_targets
|
|
|
|
# startStates canonical consistency
|
|
for ss in starts:
|
|
if is_string and ss and not ss.startswith("String."):
|
|
if any(x.startswith("String.") for x in graph):
|
|
issues.append(f"BUG: startStates {ss!r} bare but transitions use String.* form")
|
|
if ss not in graph and fn_form(ss) not in {fn_form(x) for x in graph}:
|
|
issues.append(f"WARN: startStates {ss!r} not in transition graph")
|
|
|
|
# endStates
|
|
for es in ends:
|
|
if es not in graph and fn_form(es) not in {fn_form(x) for x in graph}:
|
|
issues.append(f"WARN: endStates {es!r} not in transition graph")
|
|
|
|
# Enum canonical fullIdentifiers
|
|
if is_pkg_enum:
|
|
for i, t in enumerate(transitions):
|
|
for side in ("sourceStates", "targetStates"):
|
|
for s in t.get(side) or []:
|
|
fid = sid(s)
|
|
if fid and not fid.startswith(st_fqn + ".") and not fid.startswith('"'):
|
|
if "." in fid and not fid.startswith("<"):
|
|
issues.append(f"BUG: transitions[{i}].{side} {fid!r} not under {st_fqn}")
|
|
ev = eid(t.get("event"))
|
|
if ev and ev_fqn and "." in ev_fqn and not ev.startswith("<") and not ev.startswith(ev_fqn + "."):
|
|
if ev not in ("event",) and "." in ev and not ev.startswith(ev_fqn):
|
|
issues.append(f"BUG: transitions[{i}].event {ev!r} not under {ev_fqn}")
|
|
for i, tr in enumerate(meta.get("triggers") or []):
|
|
for field in ("event", "sourceState"):
|
|
val = tr.get(field)
|
|
if not val or val in ("event", "payload", "customMessage", "eventProvider"):
|
|
continue
|
|
type_fqn = ev_fqn if field == "event" else st_fqn
|
|
if type_fqn and "." in val and not val.startswith("<") and not val.startswith(type_fqn + "."):
|
|
if field == "sourceState" and val not in graph and fn_form(val) not in {fn_form(x) for x in graph}:
|
|
issues.append(f"BUG: triggers[{i}].sourceState {val!r} not a machine state")
|
|
elif field == "event" and type_fqn and val.isupper() is False and "." not in val and not val.startswith("<"):
|
|
if val not in ("event", "payload", "customMessage", "eventProvider", "true", "false"):
|
|
issues.append(f"INFO: triggers[{i}].event dynamic {val!r}")
|
|
|
|
# matchedTransitions
|
|
for i, chain in enumerate(meta.get("callChains") or []):
|
|
for j, mt in enumerate(chain.get("matchedTransitions") or []):
|
|
me, ms, mtgt = mt.get("event"), mt.get("sourceState"), mt.get("targetState")
|
|
ok = False
|
|
for t in transitions:
|
|
if fn_form(eid(t.get("event"))) != fn_form(me):
|
|
continue
|
|
srcs = [sid(s) for s in t.get("sourceStates") or []]
|
|
tgts = [sid(s) for s in t.get("targetStates") or []]
|
|
if (not ms or any(fn_form(x) == fn_form(ms) for x in srcs)) and (
|
|
not mtgt or any(fn_form(x) == fn_form(mtgt) for x in tgts)
|
|
):
|
|
ok = True
|
|
break
|
|
if not ok:
|
|
issues.append(
|
|
f"BUG: callChains[{i}].matchedTransitions[{j}] ({ms!r},{me!r},{mtgt!r}) no transition"
|
|
)
|
|
|
|
# String machine: mixed NEW vs String.NEW
|
|
if is_string or st_fqn == "String":
|
|
all_ids = set(graph) | set(starts) | set(ends) | {sid(s) for s in states}
|
|
for x in list(all_ids):
|
|
if not x:
|
|
continue
|
|
bare = x.strip('"')
|
|
if bare and "." not in bare and f"String.{bare}" in all_ids:
|
|
issues.append(f"BUG: mixed bare {x!r} and String.{bare}")
|
|
|
|
# stateTypeFqn missing on enum machine
|
|
if is_pkg_enum and not st_fqn:
|
|
issues.append("BUG: enum machine missing stateTypeFqn")
|
|
|
|
return name, issues
|
|
|
|
|
|
def main():
|
|
total = 0
|
|
for path in sorted(GOLDEN.rglob("*.json")):
|
|
data = json.loads(path.read_text())
|
|
machine, issues = audit_file(path, data)
|
|
rel = path.relative_to(GOLDEN)
|
|
bugs = [i for i in issues if i.startswith("BUG")]
|
|
warns = [i for i in issues if i.startswith("WARN")]
|
|
noise = [i for i in issues if i.startswith("NOISE")]
|
|
info = [i for i in issues if i.startswith("INFO")]
|
|
if bugs or warns:
|
|
print(f"\n{'='*70}\n{rel}\n machine: {machine}")
|
|
for i in bugs + warns + noise + info:
|
|
print(f" [{i.split(':')[0]}] {i.split(': ', 1)[-1]}")
|
|
total += len(bugs) + len(warns)
|
|
print(f"\nTotal BUG+WARN: {total}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|