| |
| """MAI Video Compare node side: the viewer extras, and the promise that adding |
| them changed nothing for a workflow that predates them. |
| |
| python tests/test_video_compare.py |
| Synthetic, no GPU, no models, no ComfyUI running. folder_paths and |
| comfy_api are stubbed; comfy.nested_tensor is stubbed for motion.py, |
| which window_expand imports for the burst arithmetic. |
| |
| What it checks: |
| |
| 1. HOLD MAP -> SPANS on a real map: the pier graph's H3ManualHoldMap |
| (length 243, ranges "132-150", hold 4, ramp on, bridge 8) run through |
| the node's converter, which must agree frame for frame with |
| window_expand._bursts (the arithmetic is imported, so a drift here |
| means someone copied it). |
| 2. THE MANIFEST carries `spans` and `curves` when they are wired, and |
| carries neither key when they are not. |
| 3. BYTE COMPATIBILITY: a two-video call with no hold_map and no curves |
| emits exactly the manifest string the pre-change node emitted, and |
| `winner_video` / `winner_index` are untouched. |
| 4. WIDGET ORDER: INPUT_TYPES before vs after for both classes is |
| append-only, including MAISeedHunter, whose seeds must stay ahead of |
| the new inputs. |
| |
| Exit code 0 = pass. |
| """ |
| import importlib |
| import json |
| import os |
| import sys |
| import tempfile |
| import types |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| PACK = os.path.dirname(HERE) |
| sys.path.insert(0, os.path.dirname(PACK)) |
| sys.path.insert(0, os.environ.get("COMFYUI_DIR", "/mnt/work/ai/apps/ComfyUI")) |
|
|
| |
| |
| _pkg = types.ModuleType("mainodes_pack") |
| _pkg.__path__ = [PACK] |
| sys.modules["mainodes_pack"] = _pkg |
|
|
| if "comfy.nested_tensor" not in sys.modules: |
| class _StubNested: |
| def __init__(self, tensors, *a, **k): |
| self.tensors = tensors |
|
|
| def unbind(self): |
| return self.tensors |
| _c = sys.modules.setdefault("comfy", types.ModuleType("comfy")) |
| _m = types.ModuleType("comfy.nested_tensor") |
| _m.NestedTensor = _StubNested |
| _c.nested_tensor = _m |
| sys.modules["comfy.nested_tensor"] = _m |
|
|
| vc = importlib.import_module("mainodes_pack.video_compare") |
| we = importlib.import_module("mainodes_pack.window_expand") |
| motion = importlib.import_module("mainodes_pack.motion") |
|
|
| FAILS = [] |
|
|
|
|
| def check(name, ok, detail=""): |
| print((" ok " if ok else " FAIL ") + name + ((" " + detail) if detail else "")) |
| if not ok: |
| FAILS.append(name) |
|
|
|
|
| |
| |
| |
| print("1. hold_map -> spans (pier: length 243, ranges '132-150', hold 4, ramp, bridge 8)") |
| PIER = motion.H3ManualHoldMap().build(length=243, fps=24, ranges="132-150", |
| hold=4, ramp=True, bridge=8)[0] |
| holds = json.loads(PIER)["holds"] |
| spans = vc._spans_from_hold_map(PIER) |
| runs = [] |
| for i, h in enumerate(holds): |
| if not runs or runs[-1][0] != h: |
| runs.append([h, i, i]) |
| else: |
| runs[-1][2] = i |
| print(" holds: " + ", ".join(f"{r[0]}x{r[2] - r[1] + 1}f (f{r[1]}-f{r[2]})" for r in runs)) |
| print(" spans: " + json.dumps(spans)) |
| check("world length preserved", len(holds) == 243, f"{len(holds)}") |
| check("one regenerated window", len(spans) == 1, json.dumps(spans)) |
| check("span agrees with window_expand._bursts", |
| spans == [[a, b] for a, b in we._bursts(holds)], |
| json.dumps([list(x) for x in we._bursts(holds)])) |
| check("span covers every dilated frame and no real-time frame", |
| all(holds[f] > 1 for f in range(spans[0][0], spans[0][1] + 1)) |
| and (spans[0][0] == 0 or holds[spans[0][0] - 1] == 1) |
| and (spans[0][1] == 242 or holds[spans[0][1] + 1] == 1)) |
| check("the typed range 132-150 lies inside the span", |
| spans[0][0] <= 132 and spans[0][1] >= 150, |
| f"typed 132-150, snapped {spans[0][0]}-{spans[0][1]}") |
|
|
| TWO = json.dumps({"holds": [1] * 4 + [3] * 6 + [1] * 5 + [2] * 3 + [1] * 2, "world_len": 20}) |
| check("two bursts -> two spans", vc._spans_from_hold_map(TWO) == [[4, 9], [15, 17]], |
| json.dumps(vc._spans_from_hold_map(TWO))) |
| check("a flat map has no span", |
| vc._spans_from_hold_map(json.dumps({"holds": [1] * 39, "world_len": 39})) == []) |
|
|
| |
| |
| |
| print("2. manifest extras / 3. byte compatibility") |
|
|
|
|
| class _Img: |
| def __init__(self, n): |
| self.shape = (n, 480, 864, 3) |
|
|
|
|
| class _Comp: |
| def __init__(self, n, fps, audio): |
| self.images = _Img(n) |
| self.frame_rate = fps |
| self.audio = audio |
|
|
|
|
| class FakeVideo: |
| """Enough VIDEO to drive the node: dimensions, a file write, components.""" |
|
|
| def __init__(self, n=243, fps=24.0, w=864, h=480, audio=None): |
| self.n, self.fps, self.w, self.h, self.audio = n, fps, w, h, audio |
|
|
| def get_dimensions(self): |
| return self.w, self.h |
|
|
| def save_to(self, path, **kw): |
| with open(path, "wb") as f: |
| f.write(b"\x00") |
|
|
| def get_components(self): |
| return _Comp(self.n, self.fps, self.audio) |
|
|
|
|
| TMP = tempfile.mkdtemp(prefix="mai_compare_test_") |
| _fp = types.ModuleType("folder_paths") |
| _fp.get_temp_directory = lambda: TMP |
| sys.modules["folder_paths"] = _fp |
| _api = types.ModuleType("comfy_api") |
| _latest = types.ModuleType("comfy_api.latest") |
| _latest.Types = types.SimpleNamespace(VideoContainer=lambda s: s, VideoCodec=lambda s: s) |
| _api.latest = _latest |
| sys.modules["comfy_api"] = _api |
| sys.modules["comfy_api.latest"] = _latest |
| vc.time.strftime = lambda f: "121212" |
|
|
| A, B = FakeVideo(audio=object()), FakeVideo(audio=None) |
| old = vc.MAIVideoCompare().compare(winner=2, preview_crf=23, unique_id="7", |
| video_1=A, video_2=B, label_1="ctrl", label_2="warm") |
| EXPECT = json.dumps({"items": [ |
| {"index": 1, "label": "ctrl", "filename": "cmp_7_121212_1.mp4", "subfolder": "mai_compare", |
| "type": "temp", "frames": 243, "fps": 24.0, "width": 864, "height": 480, "audio": True}, |
| {"index": 2, "label": "warm", "filename": "cmp_7_121212_2.mp4", "subfolder": "mai_compare", |
| "type": "temp", "frames": 243, "fps": 24.0, "width": 864, "height": 480, "audio": False}], |
| "winner": 2}) |
| check("old two-video manifest is byte-identical", old["result"][2] == EXPECT, |
| "got " + old["result"][2][:70] + "...") |
| check("old manifest has no new keys", list(json.loads(old["result"][2])) == ["items", "winner"]) |
| check("winner passthrough unchanged", old["result"][0] is B and old["result"][1] == 2) |
| empty = vc.MAIVideoCompare().compare(winner=2, preview_crf=23, unique_id="7", hold_map="", |
| curves="", video_1=A, video_2=B, label_1="ctrl", label_2="warm") |
| check("wiring the new inputs empty changes nothing", empty["result"][2] == EXPECT) |
|
|
| CURVES = json.dumps({"A": [0.0, 0.5, 1.0, 0.25], "B": [1.0, 0.5, 0.0, 0.75]}) |
| wired = vc.MAIVideoCompare().compare(winner=1, preview_crf=23, unique_id="7", hold_map=PIER, |
| curves=CURVES, video_1=A, video_2=B) |
| m = json.loads(wired["result"][2]) |
| check("manifest carries spans", m.get("spans") == spans, json.dumps(m.get("spans"))) |
| check("manifest carries curves", m.get("curves") == {"A": [0.0, 0.5, 1.0, 0.25], "B": [1.0, 0.5, 0.0, 0.75]}) |
| check("ui payload is the same dict", wired["ui"]["mai_compare"][0]["spans"] == spans) |
| junk = vc.MAIVideoCompare().compare(winner=1, preview_crf=23, unique_id="7", |
| hold_map="{not json", curves="[1,2]", video_1=A, video_2=B) |
| check("garbage in the new inputs never fails the render", |
| list(json.loads(junk["result"][2])) == ["items", "winner"]) |
|
|
| sh = vc.MAISeedHunter().compare(winner=2, preview_crf=23, unique_id="9", video_1=A, video_2=B, |
| seed_1=11, seed_2=22, hold_map=PIER, curves=CURVES) |
| msh = json.loads(sh["result"][3]) |
| check("seed hunter keeps the extras and its seeds", |
| msh.get("spans") == spans and [it["seed"] for it in msh["items"]] == [11, 22] |
| and sh["result"][1] == 22) |
|
|
| |
| |
| |
| print("4. widget order") |
| OLD_REQ = ["winner", "preview_crf"] |
| OLD_OPT = [f"video_{i}" for i in range(1, 7)] + [f"label_{i}" for i in range(1, 7)] |
| OLD_OPT_SH = OLD_OPT + [f"seed_{i}" for i in range(1, 7)] |
| for cls, old_req, old_opt in ((vc.MAIVideoCompare, OLD_REQ, OLD_OPT), |
| (vc.MAISeedHunter, OLD_REQ, OLD_OPT_SH)): |
| t = cls.INPUT_TYPES() |
| req, opt = list(t["required"]), list(t["optional"]) |
| print(f" {cls.__name__}") |
| print(f" required before {old_req}") |
| print(f" required after {req}") |
| print(f" optional before {old_opt}") |
| print(f" optional after {opt}") |
| check(f"{cls.__name__}: required untouched", req == old_req) |
| check(f"{cls.__name__}: optional is the old order plus a tail", |
| opt[:len(old_opt)] == old_opt, f"prefix {opt[:len(old_opt)] == old_opt}") |
| check(f"{cls.__name__}: the tail is exactly the two new inputs", |
| opt[len(old_opt):] == ["hold_map", "curves"], json.dumps(opt[len(old_opt):])) |
| for k in ("hold_map", "curves"): |
| spec = t["optional"][k] |
| check(f"{cls.__name__}: {k} is an optional forced input with an empty default", |
| spec[0] == "STRING" and spec[1].get("forceInput") is True and spec[1].get("default") == "") |
|
|
| print() |
| print("FAILURES: " + (", ".join(FAILS) if FAILS else "none")) |
| sys.exit(1 if FAILS else 0) |
|
|
