595 lines
25 KiB
Python
595 lines
25 KiB
Python
"""HEL-633: the three image-input workflows the user supplied — Qwen Image 2.1
|
||
图片编辑, MiniMax H3 首帧图生视频 (04) and H3 参考图生视频 (05, 1–9 refs).
|
||
|
||
Slots come from real wiring, every picked image is uploaded to ComfyUI and
|
||
lands in the node input it belongs to (in order, nothing stale), and a
|
||
broken export is refused before anything is queued."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import threading
|
||
from http.server import ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from calliope.comfyui.lint import blocking_problem, workflow_warnings
|
||
from calliope.comfyui.parser import detect_untagged_inputs, parse_dynamic_inputs
|
||
from calliope.comfyui.patcher import patch_workflow
|
||
from calliope.comfyui.smart_fill import smart_fill_inputs
|
||
from calliope.comfyui.validate import validate_input_values
|
||
from calliope.config import settings
|
||
from tests.test_studio_generation import (
|
||
PNG_1PX,
|
||
_FakeComfyHandler,
|
||
_import_workflow,
|
||
_real_run,
|
||
_wait_done,
|
||
)
|
||
|
||
FIXTURES = Path(__file__).parent / "fixtures" / "comfy_native"
|
||
I2V = "h3_i2v_04_API.json" # 04_i2v_fl2v_4step_sla (1).json
|
||
REF = "h3_ref2va_05_API.json" # 05_ref2va_4step_sla (1).json
|
||
QWEN_EDIT = "qwen21_edit_sage_API.json" # qwen-image-2.1-图片编辑-sage加速.json
|
||
EXAMPLES = Path(__file__).resolve().parents[2] / "example_ComfyUI_workflows"
|
||
|
||
REF_KEY = "131@ref_images"
|
||
LAST_KEY = "131@last_frame"
|
||
|
||
|
||
def _load(name: str) -> dict:
|
||
return json.loads((FIXTURES / name).read_text(encoding="utf-8"))
|
||
|
||
|
||
def _by_node(inputs: list[dict]) -> dict[str, dict]:
|
||
return {i["nodeId"]: i for i in inputs}
|
||
|
||
|
||
def _ref_links(wf: dict, node_id: str = "131") -> list[tuple[str, str]]:
|
||
"""[(input name, LoadImage image value), …] in slot order."""
|
||
inputs = wf[node_id]["inputs"]
|
||
names = sorted(
|
||
(k for k in inputs if k.startswith("ref_images.ref_image_")),
|
||
key=lambda k: int(k.rsplit("_", 1)[1]),
|
||
)
|
||
return [(k, wf[inputs[k][0]]["inputs"]["image"]) for k in names]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Slot detection
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_i2v_exposes_required_first_frame_and_optional_last_frame():
|
||
"""HEL-671: the 04 export is H3 FL2VA — `last_frame` is unwired but real."""
|
||
inputs = parse_dynamic_inputs(_load(I2V))
|
||
media = [i for i in inputs if i["kind"] in ("image", "image_list", "video", "audio")]
|
||
assert media == [
|
||
{
|
||
"nodeId": "901",
|
||
"label": "首帧",
|
||
"role": "first_frame",
|
||
"kind": "image",
|
||
"defaultValue": None,
|
||
"required": True,
|
||
"auto": True,
|
||
},
|
||
{
|
||
"nodeId": LAST_KEY,
|
||
"label": "尾帧",
|
||
"role": "last_frame",
|
||
"kind": "image",
|
||
"defaultValue": None,
|
||
"required": False,
|
||
"auto": True,
|
||
},
|
||
]
|
||
|
||
|
||
def test_wired_last_frame_is_its_own_slot_not_a_second_virtual_one():
|
||
wf = _load(I2V)
|
||
wf["950"] = {"class_type": "LoadImage", "inputs": {"image": "last.png"}}
|
||
wf["131"]["inputs"]["last_frame"] = ["950", 0]
|
||
media = [i for i in parse_dynamic_inputs(wf) if i["kind"] == "image"]
|
||
assert [(i["nodeId"], i["role"], i["required"]) for i in media] == [
|
||
("901", "first_frame", True),
|
||
("950", "last_frame", True),
|
||
]
|
||
|
||
|
||
def test_first_frame_through_a_custom_loader_gets_no_last_frame_slot():
|
||
# Repo example: VHS_LoadImagePath → resize → first_frame; no slot pair to mirror.
|
||
wf = json.loads(
|
||
(EXAMPLES / "minimax_h3-Turbo_Image2Video_20260813_API.json").read_text(encoding="utf-8")
|
||
)
|
||
assert not any(i.get("role") == "last_frame" for i in parse_dynamic_inputs(wf))
|
||
|
||
|
||
def test_ref2v_exposes_one_group_of_one_to_nine_reference_images():
|
||
inputs = parse_dynamic_inputs(_load(REF))
|
||
media = [i for i in inputs if i["kind"] in ("image", "image_list", "video", "audio")]
|
||
assert len(media) == 1
|
||
group = media[0]
|
||
assert group["nodeId"] == REF_KEY
|
||
assert group["kind"] == "image_list"
|
||
assert group["role"] == "reference_image"
|
||
assert group["label"] == "参考图"
|
||
assert (group["min"], group["max"], group["required"]) == (1, 9, True)
|
||
# the parameters HEL-627 wired stay exposed
|
||
by_node = _by_node(inputs)
|
||
for key in ("1", "115@aspect_ratio", "115@megapixels", "124@steps", "124@denoise",
|
||
"129@noise_seed", "133"):
|
||
assert key in by_node, key
|
||
assert not any(i["kind"] == "video" for i in inputs)
|
||
|
||
|
||
def test_example_five_ref_workflow_folds_into_the_same_group():
|
||
# The repo example is (Input:…)-tagged and keeps its tagged slots; stripped
|
||
# of tags (as a plain ComfyUI export) its five loaders become one group.
|
||
wf = json.loads((EXAMPLES / "video_minimax_h3_r2v_5ref_API.json").read_text(encoding="utf-8"))
|
||
inputs = detect_untagged_inputs(wf)
|
||
groups = [i for i in inputs if i["kind"] == "image_list"]
|
||
assert [g["nodeId"] for g in groups] == ["178@ref_images"]
|
||
# its five LoadImages are the group, not five separate slots
|
||
assert not any(i["kind"] == "image" for i in inputs if i.get("role") != "mask")
|
||
|
||
|
||
def test_qwen_edit_exposes_input_image_real_resolution_and_hides_orphan_latent():
|
||
inputs = parse_dynamic_inputs(_load(QWEN_EDIT))
|
||
by_node = _by_node(inputs)
|
||
img = by_node["10"]
|
||
assert (img["kind"], img["role"], img["label"]) == ("image", "image", "输入图")
|
||
assert img["required"] is True
|
||
res = by_node["4@resolution"]
|
||
assert (res["role"], res["defaultValue"], res["min"], res["max"], res["step"]) == (
|
||
"resolution", 1024, 0, 4096, 32,
|
||
)
|
||
assert by_node["4@prompt"]["role"] == "prompt"
|
||
assert by_node["4@negative_prompt"]["role"] == "negative"
|
||
for key in ("6@seed", "6@steps", "6@cfg", "6@denoise"):
|
||
assert key in by_node
|
||
# EmptyLatentImage #5 feeds nothing — its width/height would do nothing
|
||
assert not any(i["nodeId"].startswith("5@") for i in inputs)
|
||
|
||
|
||
def test_unwired_load_image_gets_no_slot():
|
||
wf = _load(I2V)
|
||
wf["950"] = {
|
||
"class_type": "LoadImage",
|
||
"inputs": {"image": "last.png"},
|
||
"_meta": {"title": "LAST"},
|
||
}
|
||
assert "950" not in _by_node(parse_dynamic_inputs(wf))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Broken exports
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_broken_qwen_export_is_flagged_and_blocked():
|
||
warnings = workflow_warnings(_load(QWEN_EDIT))
|
||
errors = {(w["code"], w["node_id"]) for w in warnings if w["level"] == "error"}
|
||
assert errors == {
|
||
("missing_class_type", "1"),
|
||
("missing_class_type", "9"),
|
||
("missing_model_link", "6"),
|
||
}
|
||
assert ("orphan_latent", "5") in {(w["code"], w["node_id"]) for w in warnings}
|
||
msg = blocking_problem(_load(QWEN_EDIT))
|
||
assert msg and "Unet Loader (GGUF)" in msg and "重新导出" in msg
|
||
|
||
|
||
def test_h3_workflows_are_clean():
|
||
assert workflow_warnings(_load(I2V)) == []
|
||
assert workflow_warnings(_load(REF)) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Patching 1 / 4 / 9 reference images
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("count", [1, 4, 9])
|
||
def test_patch_wires_each_reference_image_in_order(count):
|
||
base = _load(REF)
|
||
paths = [f"/assets/uploads/ref-{i}.png" for i in range(1, count + 1)]
|
||
wf = patch_workflow(base, {REF_KEY: paths})
|
||
links = _ref_links(wf)
|
||
assert links == [(f"ref_images.ref_image_{i}", p) for i, p in enumerate(paths)]
|
||
# every loader is a real, distinct LoadImage feeding node 131
|
||
loader_ids = [wf["131"]["inputs"][name][0] for name, _ in links]
|
||
assert len(set(loader_ids)) == count
|
||
assert all(wf[n]["class_type"] == "LoadImage" for n in loader_ids)
|
||
assert loader_ids[0] == "901"
|
||
# the base workflow is untouched
|
||
assert _ref_links(base) == [("ref_images.ref_image_0", "reference_1.png")]
|
||
|
||
|
||
def test_patch_shrinking_removes_stale_slots_and_loaders():
|
||
nine = patch_workflow(_load(REF), {REF_KEY: [f"/u/{i}.png" for i in range(9)]})
|
||
two = patch_workflow(nine, {REF_KEY: ["/u/b.png", "/u/a.png"]})
|
||
assert _ref_links(two) == [
|
||
("ref_images.ref_image_0", "/u/b.png"),
|
||
("ref_images.ref_image_1", "/u/a.png"),
|
||
]
|
||
load_images = [n for n, v in two.items() if v.get("class_type") == "LoadImage"]
|
||
assert len(load_images) == 2
|
||
assert not any("/u/8.png" in json.dumps(v) for v in two.values())
|
||
|
||
|
||
def test_single_slot_values_patch_first_frame_and_qwen_input():
|
||
i2v = patch_workflow(_load(I2V), {"901": "/assets/uploads/first.png"})
|
||
assert i2v["901"]["inputs"]["image"] == "/assets/uploads/first.png"
|
||
assert i2v["131"]["inputs"]["first_frame"] == ["901", 0]
|
||
qwen = patch_workflow(_load(QWEN_EDIT), {"10": "/assets/uploads/in.png", "4@resolution": 1536})
|
||
assert qwen["10"]["inputs"]["image"] == "/assets/uploads/in.png"
|
||
assert qwen["4"]["inputs"]["resolution"] == 1536
|
||
|
||
|
||
def test_patch_last_frame_adds_its_own_load_image():
|
||
base = _load(I2V)
|
||
wf = patch_workflow(base, {"901": "/u/first.png", LAST_KEY: "/u/last.png"})
|
||
node = wf["131"]["inputs"]
|
||
assert node["first_frame"] == ["901", 0]
|
||
last_id = node["last_frame"][0]
|
||
assert last_id not in base and node["last_frame"] == [last_id, 0]
|
||
assert wf[last_id]["class_type"] == "LoadImage"
|
||
assert wf[last_id]["inputs"] == {"image": "/u/last.png"}
|
||
assert wf["901"]["inputs"]["image"] == "/u/first.png"
|
||
# base untouched; patching the result again rewires nothing
|
||
assert "last_frame" not in base["131"]["inputs"]
|
||
again = patch_workflow(wf, {LAST_KEY: "/u/other.png"})
|
||
assert again["131"]["inputs"]["last_frame"] == [last_id, 0]
|
||
|
||
|
||
def test_patch_without_last_frame_leaves_first_frame_only_graph():
|
||
for values in ({"901": "/u/first.png"}, {"901": "/u/first.png", LAST_KEY: ""}):
|
||
wf = patch_workflow(_load(I2V), values)
|
||
assert "last_frame" not in wf["131"]["inputs"]
|
||
assert sorted(wf) == sorted(_load(I2V))
|
||
|
||
|
||
def test_legacy_job_values_keyed_by_load_image_still_patch():
|
||
"""Jobs queued before HEL-633 carry the LoadImage id, not the group key."""
|
||
wf = patch_workflow(_load(REF), {"901": "/assets/uploads/old.png"})
|
||
assert _ref_links(wf) == [("ref_images.ref_image_0", "/assets/uploads/old.png")]
|
||
|
||
|
||
def test_smart_fill_hands_reference_list_to_the_group():
|
||
inputs = parse_dynamic_inputs(_load(REF))
|
||
refs = [f"/r/{i}.png" for i in range(12)]
|
||
values = smart_fill_inputs(inputs, prompt="p", ref_images=refs)
|
||
assert values[REF_KEY] == refs[:9]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Validation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def uploads_dir(client):
|
||
d = settings.assets_dir / "uploads"
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
return d
|
||
|
||
|
||
def _pngs(d: Path, n: int) -> list[str]:
|
||
out = []
|
||
for i in range(n):
|
||
p = d / f"ref{i + 1}.png"
|
||
p.write_bytes(PNG_1PX)
|
||
out.append(str(p))
|
||
return out
|
||
|
||
|
||
def test_validate_reference_count_limits(uploads_dir):
|
||
schema = parse_dynamic_inputs(_load(REF))
|
||
base = {"1": "prompt"}
|
||
for n in (1, 4, 9):
|
||
clean, errs = validate_input_values(schema, {**base, REF_KEY: _pngs(uploads_dir, n)})
|
||
assert errs == [] and len(clean[REF_KEY]) == n
|
||
_, errs = validate_input_values(schema, {**base, REF_KEY: _pngs(uploads_dir, 10)})
|
||
assert errs == ["参考图最多 9 张（当前 10 张），请删除多余的图片"]
|
||
_, errs = validate_input_values(schema, base)
|
||
assert errs == ["需要至少 1 张参考图"]
|
||
_, errs = validate_input_values(schema, {**base, REF_KEY: []})
|
||
assert errs == ["需要至少 1 张参考图"]
|
||
|
||
|
||
def test_validate_rejects_bad_reference_files(uploads_dir):
|
||
schema = parse_dynamic_inputs(_load(REF))
|
||
gif = uploads_dir / "anim.gif"
|
||
gif.write_bytes(b"GIF89a")
|
||
ok = _pngs(uploads_dir, 1)[0]
|
||
_, errs = validate_input_values(schema, {"1": "p", REF_KEY: [ok, str(gif)]})
|
||
assert errs == ["参考图 2：格式不支持，请使用 PNG / JPG / WebP"]
|
||
_, errs = validate_input_values(schema, {"1": "p", REF_KEY: ["/etc/passwd.png"]})
|
||
assert errs == ["参考图 1：只能使用在本页上传或从生成结果选入的图片"]
|
||
_, errs = validate_input_values(schema, {"1": "p", REF_KEY: [str(uploads_dir / "gone.png")]})
|
||
assert errs == ["参考图 1：文件已不存在，请重新上传"]
|
||
|
||
|
||
def test_validate_requires_first_frame_and_input_image():
|
||
_, errs = validate_input_values(parse_dynamic_inputs(_load(I2V)), {"1": "p"})
|
||
assert errs == ["请上传首帧"]
|
||
_, errs = validate_input_values(
|
||
parse_dynamic_inputs(_load(I2V)), {"1": "p", LAST_KEY: "/u/l.png"}
|
||
)
|
||
assert errs == ["请上传首帧"]
|
||
clean, errs = validate_input_values(
|
||
parse_dynamic_inputs(_load(I2V)), {"1": "p", "901": "/u/f.png"}
|
||
)
|
||
assert errs == [] and LAST_KEY not in clean
|
||
_, errs = validate_input_values(parse_dynamic_inputs(_load(QWEN_EDIT)), {"4@prompt": "p"})
|
||
assert errs == ["请上传输入图"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Full chain against a fake ComfyUI
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _NamedUploadHandler(_FakeComfyHandler):
|
||
"""/upload/image answers with the uploaded file's own name, like ComfyUI."""
|
||
|
||
def do_POST(self): # noqa: N802 — BaseHTTPRequestHandler hook name
|
||
if self.path != "/upload/image":
|
||
super().do_POST()
|
||
return
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
body = self.rfile.read(length) if length else b""
|
||
self.server.requests.append(self.path)
|
||
self.server.uploads.append(body)
|
||
m = re.search(rb'filename="([^"]+)"', body)
|
||
name = m.group(1).decode() if m else "x.png"
|
||
self.server.upload_names.append(name)
|
||
self._json(200, {"name": name, "subfolder": "calliope", "type": "input"})
|
||
|
||
|
||
@pytest.fixture
|
||
def named_comfy():
|
||
server = ThreadingHTTPServer(("127.0.0.1", 0), _NamedUploadHandler)
|
||
server.requests = []
|
||
server.prompts = []
|
||
server.uploads = []
|
||
server.upload_names = []
|
||
server.history = {}
|
||
server.fail_prompt = False
|
||
server.out_name = "ref2v_00001_.mp4"
|
||
server.outputs_key = "images"
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
yield server
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
|
||
|
||
def _upload(client, name: str) -> str:
|
||
r = client.post("/api/playground/uploads", files={"file": (name, PNG_1PX, "image/png")})
|
||
assert r.status_code == 200, r.text
|
||
return r.json()["path"]
|
||
|
||
|
||
def _generate(client, wf_id: int, values: dict, source: str = "studio-video"):
|
||
return client.post(
|
||
"/api/playground/generate",
|
||
json={
|
||
"workflow_id": wf_id,
|
||
"input_values": values,
|
||
"source": source,
|
||
"endpoint_id": settings.comfyui_active_id,
|
||
},
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("count", [1, 4, 9])
|
||
def test_ref2v_chain_uploads_every_image_and_wires_it_in_order(client, named_comfy, count):
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "H3 参考生视频", "video", _load(REF))
|
||
paths = [_upload(client, f"角色{i}.png") for i in range(1, count + 1)]
|
||
r = _generate(
|
||
client,
|
||
wf_id,
|
||
{"1": "<Subject 1> 走进雨夜码头", REF_KEY: paths, "124@steps": 4, "133": 2},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
job = _wait_done(client, r.json()["job"]["id"])
|
||
assert job["status"] == "done", job.get("error")
|
||
assert job["payload"]["input_values"][REF_KEY] == paths
|
||
|
||
assert len(named_comfy.prompts) == 1
|
||
prompt = named_comfy.prompts[0]["prompt"]
|
||
expected = [f"calliope/{Path(p).name}" for p in paths]
|
||
assert [img for _k, img in _ref_links(prompt)] == expected
|
||
assert named_comfy.upload_names == [Path(p).name for p in paths]
|
||
assert prompt["1"]["inputs"]["value"] == "<Subject 1> 走进雨夜码头"
|
||
assert prompt["124"]["inputs"]["steps"] == 4
|
||
assert prompt["133"]["inputs"]["value"] == 2
|
||
# no leftover default reference and no unconsumed LoadImage
|
||
assert "reference_1.png" not in json.dumps(prompt)
|
||
loaders = {n for n, v in prompt.items() if v["class_type"] == "LoadImage"}
|
||
wired = {v[0] for k, v in prompt["131"]["inputs"].items() if k.startswith("ref_images.")}
|
||
assert loaders == wired
|
||
|
||
|
||
def test_ref2v_replace_and_delete_leave_no_stale_image(client, named_comfy):
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "H3 参考生视频", "video", _load(REF))
|
||
a, b, c = (_upload(client, n) for n in ("a.png", "b.png", "c.png"))
|
||
r = _generate(client, wf_id, {"1": "x", REF_KEY: [a, b, c]})
|
||
assert _wait_done(client, r.json()["job"]["id"])["status"] == "done"
|
||
# user replaced #2 with d and deleted #3
|
||
d = _upload(client, "d.png")
|
||
r = _generate(client, wf_id, {"1": "x", REF_KEY: [a, d]})
|
||
assert _wait_done(client, r.json()["job"]["id"])["status"] == "done"
|
||
second = named_comfy.prompts[1]["prompt"]
|
||
assert [img for _k, img in _ref_links(second)] == [
|
||
f"calliope/{Path(a).name}",
|
||
f"calliope/{Path(d).name}",
|
||
]
|
||
dumped = json.dumps(second)
|
||
assert Path(b).name not in dumped and Path(c).name not in dumped
|
||
|
||
|
||
def test_ref2v_ten_images_rejected_before_queue(client, named_comfy):
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "H3 参考生视频", "video", _load(REF))
|
||
paths = [_upload(client, f"r{i}.png") for i in range(10)]
|
||
r = _generate(client, wf_id, {"1": "x", REF_KEY: paths})
|
||
assert r.status_code == 400
|
||
assert "参考图最多 9 张" in r.json()["detail"]
|
||
r = _generate(client, wf_id, {"1": "x"})
|
||
assert r.status_code == 400
|
||
assert "需要至少 1 张参考图" in r.json()["detail"]
|
||
assert named_comfy.prompts == [] and named_comfy.upload_names == []
|
||
|
||
|
||
def test_i2v_chain_uploads_first_frame(client, named_comfy):
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "H3 图生视频", "video", _load(I2V))
|
||
r = _generate(client, wf_id, {"1": "镜头缓慢推近"})
|
||
assert r.status_code == 400 and "请上传首帧" in r.json()["detail"]
|
||
first = _upload(client, "首帧.png")
|
||
r = _generate(client, wf_id, {"1": "镜头缓慢推近", "901": first, "124@denoise": 0.8})
|
||
assert r.status_code == 200, r.text
|
||
assert _wait_done(client, r.json()["job"]["id"])["status"] == "done"
|
||
prompt = named_comfy.prompts[0]["prompt"]
|
||
assert prompt["901"]["inputs"]["image"] == f"calliope/{Path(first).name}"
|
||
assert prompt["131"]["inputs"]["first_frame"] == ["901", 0]
|
||
assert "last_frame" not in prompt["131"]["inputs"]
|
||
assert prompt["124"]["inputs"]["denoise"] == 0.8
|
||
|
||
|
||
def test_fl2v_chain_uploads_first_and_last_frame_into_their_own_inputs(client, named_comfy):
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "H3 首尾帧生视频", "video", _load(I2V))
|
||
listed = next(w for w in client.get("/api/workflows").json() if w["id"] == wf_id)
|
||
assert listed["usage"] == "image2video"
|
||
slots = {i["nodeId"]: i for i in listed["input_schema"] if i["kind"] == "image"}
|
||
assert slots["901"]["role"] == "first_frame" and slots[LAST_KEY]["role"] == "last_frame"
|
||
last = _upload(client, "尾帧.png")
|
||
r = _generate(client, wf_id, {"1": "从白天过渡到夜晚", LAST_KEY: last})
|
||
assert r.status_code == 400 and "请上传首帧" in r.json()["detail"]
|
||
first = _upload(client, "首帧.png")
|
||
r = _generate(client, wf_id, {"1": "从白天过渡到夜晚", "901": first, LAST_KEY: last})
|
||
assert r.status_code == 200, r.text
|
||
assert _wait_done(client, r.json()["job"]["id"])["status"] == "done"
|
||
prompt = named_comfy.prompts[0]["prompt"]
|
||
node = prompt["131"]["inputs"]
|
||
assert prompt[node["first_frame"][0]]["inputs"]["image"] == f"calliope/{Path(first).name}"
|
||
assert prompt[node["last_frame"][0]]["inputs"]["image"] == f"calliope/{Path(last).name}"
|
||
assert node["first_frame"][0] != node["last_frame"][0]
|
||
assert sorted(named_comfy.upload_names) == sorted([Path(first).name, Path(last).name])
|
||
|
||
|
||
def test_migration_renames_the_imported_first_frame_workflow(tmp_path):
|
||
import asyncio
|
||
import sqlite3
|
||
|
||
from calliope.db import migrate_db
|
||
|
||
db = tmp_path / "c.db"
|
||
asyncio.run(migrate_db(db))
|
||
conn = sqlite3.connect(db)
|
||
i2v, ref = json.dumps(_load(I2V)), json.dumps(_load(REF))
|
||
conn.executemany(
|
||
"INSERT INTO workflows (name, kind, workflow_json, description) VALUES (?, 'video', ?, ?)",
|
||
[
|
||
("H3 图生视频·首帧（04 用户提供）", i2v, None),
|
||
("H3图生视频·首帧", i2v, "我自己写的说明"),
|
||
("图生视频·首帧模板", ref, None), # not the FL2VA graph — left alone
|
||
("我的视频工作流", i2v, None), # user already renamed — left alone
|
||
],
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
asyncio.run(migrate_db(db))
|
||
asyncio.run(migrate_db(db)) # idempotent
|
||
conn = sqlite3.connect(db)
|
||
rows = conn.execute("SELECT name, description FROM workflows ORDER BY id").fetchall()
|
||
conn.close()
|
||
assert rows[0] == (
|
||
"H3 首尾帧生视频（04 用户提供）",
|
||
"首帧必填，尾帧可选：只传首帧为图生视频，同时传尾帧为首尾帧生视频",
|
||
)
|
||
assert rows[1] == ("H3首尾帧生视频", "我自己写的说明") # suffix kept, description kept
|
||
assert rows[2] == ("图生视频·首帧模板", None)
|
||
assert rows[3] == ("我的视频工作流", None)
|
||
|
||
|
||
def test_broken_qwen_import_warns_and_generation_is_refused(client, named_comfy):
|
||
with _real_run(client, named_comfy):
|
||
analyzed = client.post("/api/workflows/analyze", json={"workflow_json": _load(QWEN_EDIT)})
|
||
assert analyzed.status_code == 200
|
||
codes = {w["code"] for w in analyzed.json()["warnings"]}
|
||
assert {"missing_class_type", "missing_model_link", "orphan_latent"} <= codes
|
||
# importing is allowed (the user may fix it later) …
|
||
wf_id = _import_workflow(client, "Qwen 图片编辑", "image", _load(QWEN_EDIT))
|
||
listed = next(w for w in client.get("/api/workflows").json() if w["id"] == wf_id)
|
||
assert any(w["level"] == "error" for w in listed["warnings"])
|
||
# … but it never reaches ComfyUI and never shows as done
|
||
img = _upload(client, "门面.png")
|
||
r = _generate(client, wf_id, {"4@prompt": "换招牌", "10": img}, source="studio-image")
|
||
assert r.status_code == 400
|
||
assert "重新导出" in r.json()["detail"]
|
||
assert client.get("/api/playground/jobs").json() == []
|
||
assert named_comfy.prompts == [] and named_comfy.upload_names == []
|
||
|
||
|
||
def test_repaired_qwen_export_runs_with_its_input_image(client, named_comfy):
|
||
wf = _load(QWEN_EDIT)
|
||
wf["1"] = {
|
||
"class_type": "UNETLoader",
|
||
"inputs": {
|
||
"unet_name": "qwen_image_2.1_int8_convrot.safetensors",
|
||
"weight_dtype": "default",
|
||
},
|
||
}
|
||
del wf["9"]
|
||
wf["6"]["inputs"]["model"] = ["1", 0]
|
||
with _real_run(client, named_comfy):
|
||
named_comfy.out_name = "qwen_image_00001_.png"
|
||
wf_id = _import_workflow(client, "Qwen 图片编辑（修复版）", "image", wf)
|
||
img = _upload(client, "门面.png")
|
||
r = _generate(
|
||
client, wf_id, {"4@prompt": "换招牌", "10": img, "4@resolution": 1536, "6@steps": 20},
|
||
source="studio-image",
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
job = _wait_done(client, r.json()["job"]["id"])
|
||
assert job["status"] == "done" and job["output_paths"]
|
||
prompt = named_comfy.prompts[0]["prompt"]
|
||
assert prompt["10"]["inputs"]["image"] == f"calliope/{Path(img).name}"
|
||
assert prompt["4"]["inputs"]["images.image_1"] == ["10", 0]
|
||
assert prompt["4"]["inputs"]["resolution"] == 1536
|
||
assert prompt["6"]["inputs"]["steps"] == 20
|
||
|
||
|
||
def test_worker_refuses_broken_workflow_from_other_callers(client, named_comfy):
|
||
"""Workbench/agent jobs skip the playground check — the worker still stops them."""
|
||
from calliope.db import get_db
|
||
from calliope.queue.manager import queue_manager
|
||
from calliope.routers.playground import ensure_playground_project
|
||
|
||
with _real_run(client, named_comfy):
|
||
wf_id = _import_workflow(client, "Qwen 图片编辑", "image", _load(QWEN_EDIT))
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
pid = ensure_playground_project(conn)
|
||
finally:
|
||
conn.close()
|
||
job = queue_manager.enqueue(
|
||
project_id=pid, kind="image", workflow_id=wf_id, scene_id=None,
|
||
payload={"input_values": {"4@prompt": "x"}, "source": "playground"},
|
||
)
|
||
done = _wait_done(client, job["id"])
|
||
assert done["status"] == "failed"
|
||
assert "重新导出" in (done["error"] or "")
|
||
assert named_comfy.prompts == []
|