Fast_deploy / ComfyUI-MiniMaxH3-Context-Loop /ComfyUI-MAINodes /tests /test_widget_order_vs_release.py
| #!/usr/bin/env python3 | |
| """Widget order is a compatibility surface: every node's INPUT_TYPES vs the | |
| last published release. | |
| python tests/test_widget_order_vs_release.py | |
| Synthetic, no GPU, no models, no ComfyUI running (only a ComfyUI | |
| checkout on the path: COMFYUI_DIR, default /mnt/work/ai/apps/ComfyUI). | |
| python tests/test_widget_order_vs_release.py --regen | |
| REGENERATE THE BASELINE FIXTURE from the published tag. Extracts the | |
| tag's tree with `git show published-1.0.6:<file>` into a temp dir | |
| (the tag is NEVER checked out into a live tree), imports that pack, | |
| and rewrites tests/fixtures/widget_order_1.0.6.json. Do this only | |
| when a NEW release tag becomes the baseline: pass --tag published-X.Y.Z | |
| and commit the new fixture with the release. | |
| WHY: a saved workflow stores widget VALUES positionally, in required-then- | |
| optional declaration order. Insert an input anywhere but the end and every | |
| older workflow silently shifts one slot - the values land on the wrong | |
| widgets and the render is wrong without an error. So the pack's rule is | |
| append-only. | |
| What it checks, for every node id present in BOTH the baseline and the | |
| current tree: | |
| 1. REQUIRED is unchanged and in order. A new required key is a violation | |
| too, not just a reorder: required widgets precede optional ones, so | |
| appending one shifts every optional widget of that node. | |
| 2. OPTIONAL is the baseline list as a prefix, plus appended keys only. | |
| 3. A node that is in the baseline and GONE from the current tree FAILS | |
| loudly. Removing a published node is a release decision, never a | |
| silent default (it also fires if a module failed to import - the | |
| loader keeps the pack alive when an alpha module raises). | |
| 4. A brand-new node id passes and is listed as a notice. | |
| Type changes on a shared key and hidden-input changes are NOTICES, not | |
| failures: they do not move widget slots. | |
| Exit code 0 = pass. | |
| """ | |
| import argparse | |
| import importlib.util | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| HERE = os.path.dirname(os.path.abspath(__file__)) | |
| PACK = os.path.dirname(HERE) | |
| FIXTURE_DIR = os.path.join(HERE, "fixtures") | |
| DEFAULT_TAG = "published-1.0.6" | |
| COMFY = os.environ.get("COMFYUI_DIR", "/mnt/work/ai/apps/ComfyUI") | |
| # What a pack needs to import. Media, docs and the web assets are not code | |
| # and would only slow the extraction down. | |
| EXTRACT_SUFFIXES = (".py", ".json", ".toml") | |
| EXTRACT_SKIP_DIRS = ("assets/", "examples/", "docs/", "web/", "tests/") | |
| FAILS = [] | |
| NOTICES = [] | |
| def check(name, ok, detail=""): | |
| print((" PASS " if ok else " FAIL ") + name + (" " + detail if detail else "")) | |
| if not ok: | |
| FAILS.append(name) | |
| def _type_name(spec): | |
| """A stable, printable name for an input's declared type.""" | |
| if not isinstance(spec, (tuple, list)) or not spec: | |
| return repr(spec)[:32] | |
| t = spec[0] | |
| if isinstance(t, str): | |
| return t | |
| if isinstance(t, (list, tuple)): | |
| return "COMBO" # choices are runtime file lists; not stable | |
| return type(t).__name__ | |
| def load_pack(root, modname): | |
| """Import a pack directory AS a package and return its NODE_CLASS_MAPPINGS. | |
| The directory name is not an importable identifier and the modules use | |
| relative imports, so bind the root to a name of our choosing. | |
| """ | |
| if COMFY not in sys.path: | |
| sys.path.insert(0, COMFY) | |
| spec = importlib.util.spec_from_file_location( | |
| modname, os.path.join(root, "__init__.py"), | |
| submodule_search_locations=[root]) | |
| mod = importlib.util.module_from_spec(spec) | |
| sys.modules[modname] = mod | |
| spec.loader.exec_module(mod) | |
| return mod.NODE_CLASS_MAPPINGS | |
| def snapshot(mappings): | |
| """node id -> its INPUT_TYPES key order, per section.""" | |
| out = {} | |
| for node_id, cls in mappings.items(): | |
| t = cls.INPUT_TYPES() | |
| rec = {"class": f"{cls.__module__.split('.', 1)[-1]}.{cls.__name__}"} | |
| for section in ("required", "optional", "hidden"): | |
| d = t.get(section) or {} | |
| rec[section] = list(d) | |
| if section != "hidden": | |
| rec[section + "_types"] = {k: _type_name(v) for k, v in d.items()} | |
| out[node_id] = rec | |
| return out | |
| def extract_tag(tag, dest): | |
| """Materialize the tag's code with `git show`, without touching a tree.""" | |
| names = subprocess.run(["git", "-C", PACK, "ls-tree", "-r", "--name-only", tag], | |
| capture_output=True, text=True, check=True).stdout.split() | |
| n = 0 | |
| for name in names: | |
| if not name.endswith(EXTRACT_SUFFIXES): | |
| continue | |
| if any(name.startswith(d) for d in EXTRACT_SKIP_DIRS): | |
| continue | |
| blob = subprocess.run(["git", "-C", PACK, "show", f"{tag}:{name}"], | |
| capture_output=True, check=True).stdout | |
| path = os.path.join(dest, name) | |
| os.makedirs(os.path.dirname(path), exist_ok=True) | |
| with open(path, "wb") as fh: | |
| fh.write(blob) | |
| n += 1 | |
| return n | |
| def regen(tag, fixture): | |
| sha = subprocess.run(["git", "-C", PACK, "rev-list", "-n1", tag], | |
| capture_output=True, text=True, check=True).stdout.strip() | |
| with tempfile.TemporaryDirectory(prefix="widget-order-") as tmp: | |
| n = extract_tag(tag, tmp) | |
| print(f"extracted {n} files of {tag} ({sha[:12]}) to {tmp}") | |
| nodes = snapshot(load_pack(tmp, "mainodes_baseline")) | |
| doc = { | |
| "_comment": ("Baseline INPUT_TYPES key order per node at the published " | |
| "tag. Regenerate ONLY when a new release becomes the " | |
| "baseline: python tests/test_widget_order_vs_release.py " | |
| "--regen --tag <published-X.Y.Z>"), | |
| "tag": tag, | |
| "commit": sha, | |
| "generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), | |
| "node_count": len(nodes), | |
| "nodes": dict(sorted(nodes.items())), | |
| } | |
| os.makedirs(FIXTURE_DIR, exist_ok=True) | |
| with open(fixture, "w") as fh: | |
| json.dump(doc, fh, indent=1, sort_keys=False) | |
| fh.write("\n") | |
| print(f"wrote {fixture}: {len(nodes)} nodes at {tag}") | |
| def audit(fixture): | |
| with open(fixture) as fh: | |
| base = json.load(fh) | |
| baseline = base["nodes"] | |
| print(f"baseline: {base['tag']} ({base['commit'][:12]}), " | |
| f"{len(baseline)} nodes, generated {base['generated_utc']}") | |
| current = snapshot(load_pack(PACK, "mainodes_current")) | |
| print(f"current : {PACK}, {len(current)} nodes") | |
| print() | |
| shared = sorted(set(baseline) & set(current)) | |
| gone = sorted(set(baseline) - set(current)) | |
| new = sorted(set(current) - set(baseline)) | |
| rows = [] | |
| for node_id in shared: | |
| b, c = baseline[node_id], current[node_id] | |
| req_ok = b["required"] == c["required"] | |
| req_prefix = c["required"][:len(b["required"])] == b["required"] | |
| added_req = c["required"][len(b["required"]):] if req_prefix else [] | |
| opt_prefix = c["optional"][:len(b["optional"])] == b["optional"] | |
| added_opt = c["optional"][len(b["optional"]):] if opt_prefix else [] | |
| ok = req_ok and opt_prefix | |
| status = ("OK" if not (added_opt or added_req) else "OK +append") if ok else "VIOLATION" | |
| rows.append((node_id, status, len(b["required"]), len(c["required"]), | |
| len(b["optional"]), len(c["optional"]), | |
| added_req, added_opt, req_ok, req_prefix, opt_prefix, b, c)) | |
| if b["hidden"] != c["hidden"]: | |
| NOTICES.append(f"{node_id}: hidden {b['hidden']} -> {c['hidden']}") | |
| for section in ("required", "optional"): | |
| bt, ct = b.get(section + "_types", {}), c.get(section + "_types", {}) | |
| for k in b[section]: | |
| if k in ct and bt.get(k) != ct[k]: | |
| NOTICES.append(f"{node_id}: {section} '{k}' type " | |
| f"{bt.get(k)} -> {ct[k]}") | |
| print(f"PER-NODE TABLE ({len(shared)} shared, {len(new)} new, {len(gone)} gone)") | |
| head = (f"{'node id':<26} {'status':<11} {'req':>7} {'opt':>7} appended") | |
| print(head) | |
| print("-" * len(head)) | |
| for (node_id, status, br, cr, bo, co, added_req, added_opt, *_rest) in rows: | |
| tail = [] | |
| if added_req: | |
| tail.append("required+" + ",".join(added_req)) | |
| if added_opt: | |
| tail.append("optional+" + ",".join(added_opt)) | |
| print(f"{node_id:<26} {status:<11} {br:>3}->{cr:<3} {bo:>3}->{co:<3} " | |
| + ("; ".join(tail) if tail else "-")) | |
| for node_id in new: | |
| c = current[node_id] | |
| print(f"{node_id:<26} {'NEW':<11} {'-':>3}->{len(c['required']):<3} " | |
| f"{'-':>3}->{len(c['optional']):<3} brand new since {base['tag']}") | |
| for node_id in gone: | |
| b = baseline[node_id] | |
| print(f"{node_id:<26} {'GONE':<11} {len(b['required']):>3}->- " | |
| f"{len(b['optional']):>3}->- was in {base['tag']}") | |
| print() | |
| print("1. REQUIRED unchanged and in order") | |
| for row in rows: | |
| node_id, req_ok, req_prefix, b, c = row[0], row[8], row[9], row[11], row[12] | |
| if req_ok: | |
| continue | |
| kind = ("APPENDED required keys (they shift every optional widget)" | |
| if req_prefix else "REORDERED or REMOVED required keys") | |
| check(f"{node_id}: required keys unchanged and in order", False, | |
| f"{kind}: {b['required']} -> {c['required']}") | |
| if all(r[8] for r in rows): | |
| print(f" PASS all {len(rows)} shared nodes: required identical") | |
| print("\n2. OPTIONAL is baseline prefix + appends only") | |
| for row in rows: | |
| node_id, b, c = row[0], row[11], row[12] | |
| if row[10]: | |
| continue | |
| check(f"{node_id}: optional is the baseline order plus a tail", | |
| False, f"{b['optional']} -> {c['optional']}") | |
| if all(r[10] for r in rows): | |
| print(f" PASS all {len(rows)} shared nodes: optional prefix intact") | |
| appended = [(r[0], r[7]) for r in rows if r[7]] | |
| for node_id, keys in appended: | |
| print(f" {node_id} appended {keys}") | |
| print("\n3. NO PUBLISHED NODE DISAPPEARED") | |
| check(f"every node id in {base['tag']} is still registered", | |
| not gone, ", ".join(gone) if gone else f"{len(baseline)} ids") | |
| print("\n4. BRAND-NEW NODES (notice only)") | |
| if new: | |
| for node_id in new: | |
| print(f" {node_id} ({current[node_id]['class']}, " | |
| f"{len(current[node_id]['required'])} required / " | |
| f"{len(current[node_id]['optional'])} optional)") | |
| else: | |
| print(" none") | |
| print("\nNOTICES (type / hidden changes; no widget slot moves)") | |
| for n in NOTICES or [" none"]: | |
| print(f" {n}" if n.strip() != "none" else n) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regen", action="store_true", | |
| help="rewrite the baseline fixture from the tag") | |
| ap.add_argument("--tag", default=DEFAULT_TAG) | |
| args = ap.parse_args() | |
| fixture = os.path.join(FIXTURE_DIR, | |
| "widget_order_%s.json" % args.tag.replace("published-", "")) | |
| if args.regen: | |
| regen(args.tag, fixture) | |
| return 0 | |
| if not os.path.exists(fixture): | |
| print(f"MISSING FIXTURE {fixture} - regenerate with --regen --tag {args.tag}") | |
| return 1 | |
| audit(fixture) | |
| print() | |
| if FAILS: | |
| print(f"FAILED: {len(FAILS)} check(s): " + ", ".join(FAILS)) | |
| return 1 | |
| print("all checks passed") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |