312 lines
12 KiB
Python
312 lines
12 KiB
Python
"""HEL-636 A1–A6: the five-usage classifier judges from graph evidence only.
|
||
|
||
Every case feeds real (or minimally mutated) ComfyUI API exports — never a
|
||
file name, model name or fixed node id — and asserts the usage, the media
|
||
slots it implies, and the failure paths for ambiguous / unknown graphs.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
from calliope.comfyui.lint import blocking_problem, workflow_warnings
|
||
from calliope.comfyui.parser import (
|
||
classify_usage,
|
||
detect_untagged_inputs,
|
||
parse_dynamic_inputs,
|
||
parse_dynamic_outputs,
|
||
)
|
||
from calliope.comfyui.patcher import patch_workflow
|
||
|
||
FIXTURES = Path(__file__).parent / "fixtures" / "comfy_native"
|
||
EXAMPLES = Path(__file__).resolve().parents[2] / "example_ComfyUI_workflows"
|
||
|
||
|
||
def _load(path: Path) -> dict:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _rename_everything(wf: dict, id_shift: int = 7) -> dict:
|
||
"""Change every node id, title and widget value the classifier must not
|
||
depend on (model file names / prompt texts)."""
|
||
def new_id(nid: str) -> str:
|
||
return f"x{int(nid) * id_shift + 3}" if nid.isdigit() else "x" + nid.replace(":", "_")
|
||
|
||
out: dict = {}
|
||
for nid, node in wf.items():
|
||
out[new_id(nid)] = json.loads(json.dumps(node))
|
||
for node in out.values():
|
||
for val in (node.get("inputs") or {}).values():
|
||
if isinstance(val, list) and len(val) == 2 and str(val[0]).isdigit():
|
||
val[0] = new_id(str(val[0]))
|
||
if node.get("_meta", {}).get("title"):
|
||
node["_meta"]["title"] = f"改·{node['_meta']['title']}"
|
||
for key, val in list((node.get("inputs") or {}).items()):
|
||
if isinstance(val, str) and val.endswith((".safetensors", ".gguf")):
|
||
node["inputs"][key] = "另一个模型_" + val
|
||
return out
|
||
|
||
|
||
def _media(inputs: list[dict]) -> list[dict]:
|
||
return [
|
||
i
|
||
for i in inputs
|
||
if i["kind"] in ("image", "image_list", "video", "audio")
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# A1 / A2 / A3: the three real shapes, plus id/title/model renames
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_a1_native_t2i_is_text2image_and_form_has_no_image_slot():
|
||
wf = _load(FIXTURES / "qwen_image_t2i_API.json")
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "text2image"
|
||
assert c["ambiguous"] is False and c["needs_mapping"] is False
|
||
assert _media(parse_dynamic_inputs(wf)) == []
|
||
outs = parse_dynamic_outputs(wf)
|
||
assert [(o["nodeId"], o["kind"]) for o in outs] == [("60", "image")]
|
||
|
||
|
||
def test_a2_qwen_edit_variant_with_new_ids_titles_and_model_stays_image2image():
|
||
wf = _rename_everything(_load(FIXTURES / "qwen_image_edit_mask_API.json"))
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "image2image"
|
||
inputs = parse_dynamic_inputs(wf)
|
||
img = [i for i in _media(inputs) if i.get("role") == "image"]
|
||
assert len(img) == 1 and img[0]["required"] is True
|
||
mask = [i for i in _media(inputs) if i.get("role") == "mask"]
|
||
assert len(mask) == 1 # mask never counts as an input image
|
||
|
||
|
||
def test_a3_i2v_variant_renamed_still_image2video_with_optional_last_frame():
|
||
wf = _rename_everything(_load(FIXTURES / "h3_i2v_04_API.json"))
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "image2video"
|
||
media = _media(parse_dynamic_inputs(wf))
|
||
# HEL-671: the FL2VA node's unwired last_frame is offered, but optional
|
||
assert [(i.get("role"), i["required"]) for i in media] == [
|
||
("first_frame", True),
|
||
("last_frame", False),
|
||
]
|
||
assert blocking_problem(wf) is None
|
||
|
||
|
||
def test_a4_ref2va_variant_with_renamed_class_and_prefix_still_groups():
|
||
wf = _rename_everything(_load(FIXTURES / "h3_ref2va_05_API.json"))
|
||
for node in wf.values():
|
||
if node.get("class_type") == "MiniMaxH3ReferenceToVideo":
|
||
node["class_type"] = "MyFancyRefToVideo"
|
||
inputs = node["inputs"]
|
||
for k in list(inputs):
|
||
if k.startswith("ref_images.ref_image_"):
|
||
inputs[k.replace("ref_images.ref_image_", "refs.pic_")] = inputs.pop(k)
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "ref2video"
|
||
groups = [i for i in detect_untagged_inputs(wf) if i["kind"] == "image_list"]
|
||
assert len(groups) == 1
|
||
g = groups[0]
|
||
assert g["role"] == "reference_image" and (g["min"], g["max"]) == (1, 9)
|
||
# patch rewires the renamed group in order and refuses a 10th image
|
||
key = g["nodeId"]
|
||
node_id = key.split("@")[0]
|
||
patched = patch_workflow(wf, {key: [f"/u/{i}.png" for i in range(4)]})
|
||
names = sorted(
|
||
(k for k in patched[node_id]["inputs"] if k.startswith("refs.pic_")),
|
||
key=lambda k: int(k.rsplit("_", 1)[-1]) if k.rsplit("_", 1)[-1].isdigit() else int(k[-1]),
|
||
)
|
||
assert names == [f"refs.pic_{i}" for i in range(4)]
|
||
from calliope.comfyui.validate import validate_input_values
|
||
|
||
_c, errs = validate_input_values(
|
||
detect_untagged_inputs(wf), {key: [f"/u/{i}.png" for i in range(10)]}
|
||
)
|
||
assert any("最多 9 张" in e for e in errs)
|
||
|
||
|
||
def test_qwen_single_numbered_encoder_input_is_not_mistaken_for_a_group():
|
||
"""TextEncodeQwenImage21's ``images.image_1`` stays an ordinary input slot."""
|
||
inputs = parse_dynamic_inputs(_load(FIXTURES / "qwen21_edit_sage_API.json"))
|
||
assert not any(i["kind"] == "image_list" for i in inputs)
|
||
assert any(i.get("role") == "image" for i in inputs)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Judgment table rows from the real example library
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_all_real_exports_classify():
|
||
expect = {
|
||
"qwen_image_t2i_API.json": "text2image",
|
||
"qwen_image_edit_mask_API.json": "image2image",
|
||
"qwen21_edit_sage_API.json": "image2image",
|
||
"h3_i2v_04_API.json": "image2video",
|
||
"h3_ref2va_05_API.json": "ref2video",
|
||
"minimax_h3_t2v_untagged_API.json": "text2video",
|
||
"wen_sheng_tu_API.json": "text2image",
|
||
"wen_sheng_shipin_API.json": "text2video",
|
||
}
|
||
for name, usage in expect.items():
|
||
assert classify_usage(_load(FIXTURES / name))["usage"] == usage, name
|
||
examples = {
|
||
"Krea2_t2i_20260818_API.json": "text2image",
|
||
"krea2_T2I_CharSheetV1_API.json": "text2image",
|
||
"minimax_h3-Turbo_Text2Video_20260813_API.json": "text2video",
|
||
"minimax_h3-Turbo_Image2Video_20260813_API.json": "text2video",
|
||
"MiniMax-H3-Turbo-R2V-Extend-2Refs1Vid_API.json": "ref2video",
|
||
"video_minimax_h3_r2v_2ref_API.json": "ref2video",
|
||
"video_minimax_h3_r2v_3ref_API.json": "ref2video",
|
||
"video_minimax_h3_r2v_4ref_API.json": "ref2video",
|
||
"video_minimax_h3_r2v_5ref_API.json": "ref2video",
|
||
}
|
||
for name, usage in examples.items():
|
||
assert classify_usage(_load(EXAMPLES / name))["usage"] == usage, name
|
||
|
||
|
||
def test_r2v_extend_reports_its_video_input_slot():
|
||
wf = _load(EXAMPLES / "MiniMax-H3-Turbo-R2V-Extend-2Refs1Vid_API.json")
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "ref2video"
|
||
assert c["evidence"]["video_slots"], "the reference-video slot must be reported"
|
||
assert any("视频输入槽" in line for line in c["evidence"]["summary"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# A5: ambiguity — single loose image into a video graph; multi-output graph
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _load_image(nid: str) -> dict:
|
||
return {
|
||
"class_type": "LoadImage",
|
||
"inputs": {"image": "x.png"},
|
||
"_meta": {"title": f"图 {nid}"},
|
||
}
|
||
|
||
|
||
def test_single_loose_image_with_video_output_is_ambiguous():
|
||
wf = {
|
||
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
|
||
"5": {"class_type": "LoadImage", "inputs": {"image": "x.png"}},
|
||
"7": {
|
||
"class_type": "SomeVideoSampler",
|
||
"inputs": {"prompt": ["1", 0], "image": ["5", 0]},
|
||
},
|
||
"9": {"class_type": "SaveVideo", "inputs": {"images": ["7", 0]}},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["usage"] is None and c["ambiguous"] is True
|
||
assert "首帧还是参考图" in c["problem"]
|
||
|
||
|
||
def test_first_frame_role_decides_image2video_without_ambiguity():
|
||
wf = {
|
||
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
|
||
"5": {"class_type": "LoadImage", "inputs": {"image": "x.png"}},
|
||
"7": {
|
||
"class_type": "SomeVideoSampler",
|
||
"inputs": {"prompt": ["1", 0], "first_frame": ["5", 0]},
|
||
},
|
||
"9": {"class_type": "SaveVideo", "inputs": {"images": ["7", 0]}},
|
||
}
|
||
assert classify_usage(wf)["usage"] == "image2video"
|
||
|
||
|
||
def test_same_graph_both_image_and_video_outputs_is_ambiguous():
|
||
wf = {
|
||
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
|
||
"5": {"class_type": "LoadImage", "inputs": {"image": "x.png"}},
|
||
"7": {
|
||
"class_type": "SomeVideoSampler",
|
||
"inputs": {"prompt": ["1", 0], "image": ["5", 0]},
|
||
},
|
||
"9": {"class_type": "SaveVideo", "inputs": {"images": ["7", 0]}},
|
||
"10": {"class_type": "SaveImage", "inputs": {"images": ["7", 1]}},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["usage"] is None and c["ambiguous"] is True
|
||
assert "同时输出图片和视频" in c["problem"]
|
||
|
||
|
||
def test_two_reference_images_into_a_custom_node_become_a_group_ref2video():
|
||
wf = {
|
||
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
|
||
"5": {"class_type": "LoadImage", "inputs": {"image": "a.png"}},
|
||
"6": {"class_type": "LoadImage", "inputs": {"image": "b.png"}},
|
||
"7": {
|
||
"class_type": "GenericRefToVideo",
|
||
"inputs": {
|
||
"prompt": ["1", 0],
|
||
"refs.pic_1": ["5", 0],
|
||
"refs.pic_2": ["6", 0],
|
||
},
|
||
},
|
||
"9": {"class_type": "SaveVideo", "inputs": {"images": ["7", 0]}},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "ref2video"
|
||
groups = [i for i in detect_untagged_inputs(wf) if i["kind"] == "image_list"]
|
||
assert [g["nodeId"] for g in groups] == ["7@refs"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# A6: failure paths — no output, unknown custom output nodes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_no_output_graph_reports_chinese_error_and_blocks():
|
||
wf = {
|
||
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "p"}},
|
||
"2": {"class_type": "VAEDecode", "inputs": {"samples": ["1", 0]}},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["usage"] is None and c["ambiguous"] is False
|
||
assert "未在图中找到输出节点" in c["problem"]
|
||
assert blocking_problem(wf) and "未在图中找到输出节点" in blocking_problem(wf)
|
||
|
||
|
||
def test_unknown_custom_save_node_needs_mapping_and_keeps_a_runnable_path():
|
||
wf = {
|
||
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
|
||
"7": {
|
||
"class_type": "MyCloudExporter",
|
||
"inputs": {"filename_prefix": "out", "render": ["8", 0]},
|
||
},
|
||
"8": {"class_type": "VAEDecode", "inputs": {"samples": ["1", 0]}},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["needs_mapping"] is True
|
||
assert c["usage"] is None and "无法识别" in (c["problem"] or "")
|
||
# unknown savers are a warning (user maps them at import), not a block —
|
||
# once the user confirms the usage the job can still run and results are
|
||
# pulled from ComfyUI history's images/gifs/videos keys.
|
||
codes = {(w["level"], w["code"]) for w in workflow_warnings(wf)}
|
||
assert ("warning", "unknown_output_node") in codes
|
||
assert ("error", "no_output") not in codes
|
||
|
||
|
||
def test_utility_sink_tail_is_not_an_output_and_does_not_break_detection():
|
||
"""VideoCombine feeding a cache-cleanup tail still counts as the output."""
|
||
wf = _load(FIXTURES / "minimax_h3_t2v_untagged_API.json")
|
||
wf["9999"] = {
|
||
"class_type": "easy cleanGpuUsed",
|
||
"inputs": {"anything": ["5978", 0]},
|
||
}
|
||
c = classify_usage(wf)
|
||
assert c["usage"] == "text2video"
|
||
assert c["needs_mapping"] is False
|
||
assert workflow_warnings(wf) == []
|
||
|
||
|
||
def test_evidence_lists_output_nodes_and_slots():
|
||
wf = _load(FIXTURES / "h3_ref2va_05_API.json")
|
||
c = classify_usage(wf)
|
||
ev = c["evidence"]
|
||
assert any(o["nodeId"] == "92" and o["kind"] == "video" for o in ev["outputs"])
|
||
assert any(s["kind"] == "image_list" for s in ev["image_slots"])
|
||
assert any("输出" in line for line in ev["summary"])
|