digital-garbageandClaude Sonnet 5.5
30d1aa14a9
A pipeline with no length or rate input hides those controls in the Project and Scene panels. A preset can carry project numbers it goes with. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
608 lines
29 KiB
Python
608 lines
29 KiB
Python
"""The pipeline, over HTTP.
|
|
|
|
core/graph.py was well tested and called from nowhere -- proven correct in a
|
|
vacuum while the claim it backs ("a built-in node can be replaced or removed")
|
|
was not something the running server could do. These tests are about
|
|
reachability as much as behaviour.
|
|
"""
|
|
|
|
import json
|
|
import socket
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("aiohttp")
|
|
|
|
from aiohttp import web # noqa: E402
|
|
|
|
from core import routes # noqa: E402
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def model_files(comfyui, tmp_path_factory):
|
|
"""One empty file in each folder a default loader reads, so the loaders' dropdowns have something to offer.
|
|
|
|
A machine without models (a cloud test run) would otherwise report "nothing fills it" for every loader, and a
|
|
test that deletes ONE file could not tell its own problem from that. The nodes read their options when they
|
|
register, so this comes before `registered`."""
|
|
import folder_paths
|
|
root = tmp_path_factory.mktemp("models")
|
|
for kind in ("diffusion_models", "vae", "text_encoders", "clip", "loras"):
|
|
(root / kind).mkdir()
|
|
(root / kind / "stand-in.safetensors").write_bytes(b"\0")
|
|
folder_paths.add_model_folder_path(kind, str(root / kind))
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def registered(comfyui, model_files):
|
|
"""FunUI's nodes registered the way ComfyUI registers them.
|
|
|
|
Without this the pipeline's slots all look like missing nodes, because the
|
|
schema lookup reads ComfyUI's registry and nothing had put anything in it.
|
|
"""
|
|
import asyncio
|
|
import nodes as comfy_nodes
|
|
from pathlib import Path
|
|
|
|
async def load():
|
|
await comfy_nodes.init_extra_nodes(init_custom_nodes=False)
|
|
await comfy_nodes.load_custom_node(
|
|
str(Path(__file__).resolve().parents[2]), module_parent="custom_nodes")
|
|
|
|
asyncio.run(load())
|
|
return comfy_nodes
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def server(comfyui, registered):
|
|
"""The real route table on a throwaway app, on a real socket."""
|
|
from aiohttp import web as aioweb
|
|
|
|
app = aioweb.Application()
|
|
table = aioweb.RouteTableDef()
|
|
routes.register(table, prefix="/fun")
|
|
app.add_routes(table)
|
|
|
|
holder = {}
|
|
|
|
def run():
|
|
import asyncio
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
runner = aioweb.AppRunner(app)
|
|
loop.run_until_complete(runner.setup())
|
|
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
|
|
loop.run_until_complete(site.start())
|
|
holder["port"] = site._server.sockets[0].getsockname()[1]
|
|
holder["loop"] = loop
|
|
holder["ready"].set()
|
|
loop.run_forever()
|
|
|
|
holder["ready"] = threading.Event()
|
|
thread = threading.Thread(target=run, daemon=True)
|
|
thread.start()
|
|
assert holder["ready"].wait(20), "the test server did not start"
|
|
yield holder["port"]
|
|
holder["loop"].call_soon_threadsafe(holder["loop"].stop)
|
|
|
|
|
|
def _request(port, method, path, body=None):
|
|
payload = b"" if body is None else json.dumps(body).encode()
|
|
head = (f"{method} {path} HTTP/1.1\r\nHost: localhost\r\n"
|
|
f"Content-Type: application/json\r\n"
|
|
f"Content-Length: {len(payload)}\r\nConnection: close\r\n\r\n").encode()
|
|
with socket.create_connection(("127.0.0.1", port), timeout=30) as sock:
|
|
sock.sendall(head + payload)
|
|
chunks = []
|
|
while True:
|
|
got = sock.recv(65536)
|
|
if not got:
|
|
break
|
|
chunks.append(got)
|
|
raw = b"".join(chunks)
|
|
header, _, rest = raw.partition(b"\r\n\r\n")
|
|
status = int(header.split()[1])
|
|
try:
|
|
return status, json.loads(rest.decode() or "{}")
|
|
except ValueError:
|
|
return status, {"raw": rest.decode(errors="replace")}
|
|
|
|
|
|
def test_the_default_pipeline_is_served(server):
|
|
status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
assert status == 200
|
|
assert [slot["id"] for slot in body["slots"]], "no slots came back"
|
|
assert "queueable" in body
|
|
# A fresh install has no model picked, so the default pipeline is
|
|
# incomplete rather than wrong -- and the two are reported separately.
|
|
assert body["refused"] == []
|
|
assert any("nothing fills it" in p for p in body["incomplete"])
|
|
|
|
|
|
def test_a_built_in_slot_can_be_removed_over_http(server):
|
|
"""The claim, exercised the way the app will: the modifiers node taken out,
|
|
and what it fed rewired to what fed it."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "remove", "slot": "modifiers"})
|
|
assert status == 200, body
|
|
assert body["refused"] == [], body["refused"]
|
|
ids = [slot["id"] for slot in body["slots"]]
|
|
assert "modifiers" not in ids
|
|
|
|
sampler = next(s for s in body["slots"] if s["id"] == "sampler")
|
|
assert sampler["inputs"]["model"] == ["model", 0], "the sampler was not rewired"
|
|
|
|
|
|
def test_a_node_can_be_added_over_http_and_a_missing_one_comes_back_as_a_reason(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "add", "node": "PrimitiveInt", "group": "Mine"})
|
|
assert status == 200 and body["refused"] == [], body
|
|
added = body["slots"][-1]
|
|
assert added["node"] == "PrimitiveInt" and added["group"] == "Mine" and added["inputs"] == {}
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"action": "add", "node": "NoSuchNode"})
|
|
assert status == 200 and any("NoSuchNode" in r for r in body["refused"])
|
|
assert _request(server, "POST", "/fun/api/pipeline", {"action": "add", "node": 5})[0] == 400
|
|
|
|
|
|
def test_a_removal_that_cannot_work_comes_back_as_a_reason(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "remove", "slot": "latent"})
|
|
assert status == 200
|
|
assert any("LATENT" in p for p in body["refused"])
|
|
assert body["queueable"] is False
|
|
assert body["prompt"] is None
|
|
|
|
|
|
def test_a_built_in_slot_can_be_replaced_over_http(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "replace", "slot": "vae", "node": "VAELoader"})
|
|
assert status == 200, body
|
|
vae = next(s for s in body["slots"] if s["id"] == "vae")
|
|
assert vae["node"] == "VAELoader"
|
|
|
|
|
|
def test_a_replacement_that_gives_the_wrong_thing_is_refused(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "replace", "slot": "model", "node": "SaveImage"})
|
|
assert status == 200
|
|
assert body["refused"], "a nonsense replacement was accepted"
|
|
|
|
|
|
def test_an_unknown_action_is_refused(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "explode", "slot": "model"})
|
|
assert status == 400 and body["problems"] # a malformed request, not an edit
|
|
|
|
|
|
# --- wiring, over HTTP ------------------------------------------------------
|
|
|
|
def test_a_socket_can_be_wired_over_http(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "wire", "slot": "decode", "input": "audio_vae",
|
|
"from_slot": "vae", "from_output": 0})
|
|
assert status == 200, body
|
|
assert body["refused"] == [], body["refused"]
|
|
decode = next(s for s in body["slots"] if s["id"] == "decode")
|
|
assert decode["inputs"]["audio_vae"] == ["vae", 0]
|
|
|
|
|
|
def test_a_type_mismatched_wire_is_refused_over_http(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "wire", "slot": "latent", "input": "width",
|
|
"from_slot": "vae", "from_output": 0})
|
|
assert status == 200
|
|
assert any("wants" in p for p in body["refused"])
|
|
|
|
|
|
def test_wiring_a_slot_that_does_not_exist_is_refused_over_http(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "wire", "slot": "gone", "input": "x",
|
|
"from_slot": "vae", "from_output": 0})
|
|
assert status == 200
|
|
assert any("gone" in p for p in body["refused"])
|
|
|
|
|
|
@pytest.mark.parametrize("payload,expected", [
|
|
({"action": "wire", "slot": ["a"], "input": "x", "from_slot": "vae", "from_output": 0}, "named by a string"),
|
|
({"action": "wire", "slot": "decode", "input": 7, "from_slot": "vae", "from_output": 0}, "named by a string"),
|
|
({"action": "wire", "slot": "decode", "input": "x", "from_slot": 7, "from_output": 0}, "named by a string"),
|
|
({"action": "wire", "slot": "decode", "input": "x", "from_slot": "vae", "from_output": "0"}, "a number"),
|
|
({"action": "wire", "slot": "decode", "input": "x", "from_slot": "vae", "from_output": True}, "a number"),
|
|
({"action": "unwire", "slot": ["a"], "input": "x"}, "named by a string"),
|
|
({"action": "unwire", "slot": "decode", "input": 7}, "named by a string"),
|
|
])
|
|
def test_a_malformed_wire_request_comes_back_as_a_400_not_a_500(server, payload, expected):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", payload)
|
|
assert status == 400, body
|
|
assert any(expected in p for p in body["problems"]), body
|
|
|
|
|
|
def test_unwiring_a_socket_over_http(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "unwire", "slot": "latent", "input": "model"})
|
|
assert status == 200, body
|
|
assert body["refused"] == []
|
|
latent = next(s for s in body["slots"] if s["id"] == "latent")
|
|
assert "model" not in latent["inputs"]
|
|
# Unwired now means unfilled: the default pipeline's own latent.model is
|
|
# required, so removing its feed is a real, reportable incompleteness.
|
|
assert any("model" in p for p in body["incomplete"])
|
|
|
|
|
|
def test_an_explicitly_empty_pipeline_is_not_replaced_by_the_default(server):
|
|
"""`slots or default()` resurrects the default, because an empty list is
|
|
falsy. A client that removed every slot is entitled to be told it has none
|
|
rather than handed the built-ins back."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "check", "slots": []})
|
|
assert status == 200
|
|
assert body["slots"] == [], "an empty pipeline came back full"
|
|
|
|
|
|
def test_omitting_slots_still_falls_back_to_the_default(server):
|
|
"""The distinction that matters: absent means "use the default", empty means
|
|
"there is nothing". Only one of them is a fallback."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "check"})
|
|
assert status == 200
|
|
assert body["slots"], "omitting slots should use the default pipeline"
|
|
|
|
|
|
def test_removing_slots_one_by_one_never_resurrects_the_default(server):
|
|
"""The failure this guards: each request carries the client's current slots,
|
|
so a fallback that fires on an empty list would silently hand the built-ins
|
|
back mid-way through and the count would jump UP."""
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
slots = body["slots"]
|
|
started_with = len(slots)
|
|
assert started_with > 1, "the default pipeline is too small to test this"
|
|
|
|
counts = [started_with]
|
|
# Consumers before sources: removing a source nothing can replace is refused,
|
|
# which is correct and not what this test is about.
|
|
for slot_id in [s["id"] for s in reversed(slots)]:
|
|
_status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "remove", "slot": slot_id, "slots": slots})
|
|
if body["refused"]:
|
|
continue
|
|
slots = body["slots"]
|
|
counts.append(len(slots))
|
|
|
|
assert len(slots) < started_with, "nothing was actually removed"
|
|
assert counts == sorted(counts, reverse=True), (
|
|
f"the pipeline grew back part way through: {counts}")
|
|
|
|
|
|
@pytest.mark.parametrize("payload,expected", [
|
|
({"action": "check", "slots": ["not-a-dict"]}, "not an object"),
|
|
({"action": "check", "slots": [{"node": "LoadModel"}]}, "has no id"),
|
|
({"action": "check", "slots": [{"id": "a"}]}, "has no node"),
|
|
({"action": "check", "slots": [{"id": "a", "node": "X", "inputs": "nope"}]}, "must be an object"),
|
|
({"action": "check", "slots": "a pipeline"}, "is a list of slots"),
|
|
# The body itself. Every case above is already a well-formed object, so the
|
|
# layer that reads named fields off it had never been handed something with
|
|
# no names: `body.get` on a list is an AttributeError and a 500.
|
|
([], "an object, not a list"),
|
|
("just a string", "an object, not a str"),
|
|
(42, "an object, not a int"),
|
|
# The fields naming what to edit. The slots array is shape-checked; these
|
|
# come from the same body and were not, and both reach an `in` against a
|
|
# dict, where an unhashable value is a TypeError rather than a miss.
|
|
({"action": "replace", "slot": ["a"], "node": "X", "slots": []}, "named by a string"),
|
|
({"action": "remove", "slot": {"x": 1}, "slots": []}, "named by a string"),
|
|
({"action": "replace", "slot": "a", "node": ["X"], "slots": []}, "named by a string"),
|
|
# A group is the pipeline's own arrangement and core knows nothing about
|
|
# what the names mean -- only that they are names. A number here becomes a
|
|
# card titled `5`, and `5` and `"5"` become two cards reading the same.
|
|
({"action": "check", "slots": [{"id": "a", "node": "X", "group": 5}]}, "must be a name"),
|
|
({"action": "check", "slots": [{"id": "a", "node": "X", "group": " "}]}, "is blank"),
|
|
])
|
|
def test_a_malformed_pipeline_comes_back_as_a_reason_not_a_500(server, payload, expected):
|
|
"""A payload arrives from HTTP, so "it is a list" is all that has been
|
|
established. Reading slot["id"] off a string raised a TypeError and the
|
|
route answered 500 with plain text, which the app cannot parse into
|
|
anything to show."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", payload)
|
|
assert status in (200, 400), f"a malformed payload crashed the route: {status}"
|
|
reasons = body.get("refused", []) + body.get("problems", [])
|
|
assert any(expected in r for r in reasons), reasons
|
|
assert body.get("queueable") is False
|
|
|
|
|
|
def test_the_default_pipeline_arrives_already_arranged(server):
|
|
"""The window draws a card per group, so a pipeline with no groups is one
|
|
card holding everything -- which is the arrangement it exists to avoid."""
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
groups = [slot.get("group") for slot in body["slots"]]
|
|
assert all(groups), f"slots with no group: {[s['id'] for s in body['slots'] if not s.get('group')]}"
|
|
assert len(set(groups)) > 1, "everything landed in one group"
|
|
|
|
|
|
def test_a_group_survives_an_edit(server):
|
|
"""Replacing a node keeps its arrangement. `replace` deliberately drops the
|
|
slot's inputs -- a different node has different inputs -- and dropping the
|
|
group with them would scatter the pipeline every time a node was swapped."""
|
|
_status, before = _request(server, "GET", "/fun/api/pipeline")
|
|
was = {slot["id"]: slot.get("group") for slot in before["slots"]}
|
|
|
|
_status, after = _request(
|
|
server, "POST", "/fun/api/pipeline",
|
|
{"action": "replace", "slot": "save", "node": "PreviewImage",
|
|
"slots": before["slots"]})
|
|
assert not after["refused"], after["refused"]
|
|
now = {slot["id"]: slot.get("group") for slot in after["slots"]}
|
|
assert now == was
|
|
|
|
|
|
# --- the whole registry, which is the only place the odd shapes live --------
|
|
#
|
|
# This is the one test file that loads ComfyUI's extras, so it is the only place
|
|
# a sweep like this means anything. These were written after a v4 bug report
|
|
# about a node "showing only its LoRA selection"; the shapes themselves are
|
|
# covered in test_comfy_types.py.
|
|
|
|
def test_no_installed_node_is_described_with_a_type_nothing_can_edit(registered):
|
|
"""A widget's type is what decides which control is drawn. A union like
|
|
"FLOAT,INT" reaching the form asks for a control for a type that does not
|
|
exist, and the whole node then renders as nothing."""
|
|
from core import widgets
|
|
|
|
for class_type in registered.NODE_CLASS_MAPPINGS:
|
|
described = widgets.describe(class_type)
|
|
if not described:
|
|
continue
|
|
for widget in described["widgets"]:
|
|
assert "," not in widget["type"], f"{class_type}.{widget['name']} is {widget['type']}"
|
|
assert widget["type"] in widgets.PRIMITIVE, (
|
|
f"{class_type}.{widget['name']} is {widget['type']}, which has no control")
|
|
|
|
|
|
def test_every_dropdown_in_the_registry_is_offered_as_one(registered):
|
|
"""A dropdown classified as a socket is an input the user cannot set and the
|
|
window claims is wired.
|
|
|
|
The check is spelled out here rather than asked of `comfy_types.is_combo`.
|
|
It was written that way first, and reintroducing the bug did not fail this
|
|
test: the sweep was asking the broken function whether the function was
|
|
broken, and got the answer the bug implies. A test that consults the code
|
|
under test for its own verdict cannot fail.
|
|
"""
|
|
from core import comfy_types, widgets
|
|
|
|
dropdowns = 0
|
|
for class_type in registered.NODE_CLASS_MAPPINGS:
|
|
described = widgets.describe(class_type)
|
|
if not described:
|
|
continue
|
|
for socket in described["sockets"]:
|
|
assert "COMBO" not in str(socket["type"]).upper(), (
|
|
f"{class_type}.{socket['name']} is a dropdown ({socket['type']}) "
|
|
f"reported as a wire")
|
|
dropdowns += sum(1 for w in described["widgets"]
|
|
if w["type"] == comfy_types.COMBO)
|
|
|
|
assert dropdowns > 100, f"only {dropdowns} dropdowns found; the sweep is not seeing them"
|
|
|
|
|
|
def test_a_file_that_is_no_longer_there_is_refused_here_not_at_the_queue(server, registered):
|
|
"""v4's failure, reproduced against real nodes.
|
|
|
|
A LoRA file that had been deleted made ComfyUI refuse the WHOLE prompt --
|
|
a feature that was switched off still stopped every generation, and the
|
|
message was "Prompt outputs failed validation" over a graph of a dozen
|
|
loaders. ComfyUI checks this at /prompt, which is one step after our own
|
|
check says the pipeline is queueable.
|
|
"""
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
slots = body["slots"]
|
|
for slot in slots:
|
|
if slot["id"] == "model":
|
|
slot["inputs"]["model_name"] = "deleted-yesterday.safetensors"
|
|
|
|
_status, checked = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "check", "slots": slots})
|
|
assert checked["queueable"] is False
|
|
assert any("deleted-yesterday.safetensors" in problem and "not one of" in problem
|
|
for problem in checked["incomplete"]), checked["incomplete"]
|
|
assert checked["prompt"] is None, "a graph the queue would refuse was handed over"
|
|
|
|
|
|
def test_a_value_the_node_does_offer_still_builds(server, registered):
|
|
"""The control. Without it the test above passes on a check that refuses
|
|
everything."""
|
|
from core import widgets
|
|
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
slots = body["slots"]
|
|
described = widgets.describe("FunUIDiffusionModelLoader")
|
|
choices = next(w for w in described["widgets"] if w["name"] == "model_name")["choices"]
|
|
if not choices:
|
|
pytest.skip("no model files on this machine, so there is no valid value to use")
|
|
|
|
for slot in slots:
|
|
if slot["id"] == "model":
|
|
slot["inputs"]["model_name"] = choices[0]
|
|
|
|
_status, checked = _request(server, "POST", "/fun/api/pipeline",
|
|
{"action": "check", "slots": slots})
|
|
assert not any("not one of" in problem for problem in checked["incomplete"]), \
|
|
checked["incomplete"]
|
|
|
|
|
|
def test_what_the_ui_holds_reaches_the_graph(server):
|
|
"""The settings a person picked, on the node that carries them.
|
|
|
|
Until this existed every panel in the app was decoration: the values were
|
|
kept in the browser and nothing ever sent them, so a modifier could be
|
|
switched on and the run would not know.
|
|
"""
|
|
values = {"sampling_alg": {"enabled": True, "strength": 0.4}}
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"values": values})
|
|
assert status == 200, body
|
|
assert body["refused"] == []
|
|
|
|
settings = next(s for s in body["slots"] if s["node"] == "FunUIModifierSettings")
|
|
assert json.loads(settings["inputs"]["settings"]) == values
|
|
assert body["notes"] == []
|
|
|
|
|
|
def test_settings_with_nowhere_to_go_are_said_and_not_swallowed(server):
|
|
"""A pipeline with nothing to accept them is legitimate -- and has to say so.
|
|
|
|
A run that silently ignores every switch in the app is the exact fault this
|
|
project keeps finding: a knob present and inert.
|
|
"""
|
|
status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
kept = [s for s in body["slots"] if s["node"] != "FunUIModifierSettings"]
|
|
for slot in kept: # nothing may still link to it
|
|
slot["inputs"] = {k: v for k, v in slot["inputs"].items()
|
|
if not (isinstance(v, list) and v and v[0] == "settings")}
|
|
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"slots": kept, "values": {"sampling_alg": {"enabled": True}}})
|
|
assert status == 200, body
|
|
assert any("will not be applied" in note for note in body["notes"]), body["notes"]
|
|
# Said, not blocked: the pipeline is otherwise exactly as valid as it was.
|
|
assert body["refused"] == []
|
|
|
|
|
|
def test_no_settings_sent_means_the_slot_is_left_alone(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {})
|
|
settings = next(s for s in body["slots"] if s["node"] == "FunUIModifierSettings")
|
|
assert settings["inputs"]["settings"] == "{}"
|
|
assert body["notes"] == []
|
|
|
|
|
|
def test_settings_that_are_not_an_object_are_refused(server):
|
|
for payload in (["a"], "everything on", 7):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"values": payload})
|
|
assert status == 400, (payload, status, body)
|
|
assert body["queueable"] is False
|
|
assert any("object keyed by module id" in p for p in body["problems"]), body
|
|
|
|
|
|
def test_empty_settings_are_not_reported_as_having_nowhere_to_go(server):
|
|
"""Nothing set is not a warning. Every page load starts here."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"values": {}})
|
|
assert status == 200
|
|
assert body["notes"] == []
|
|
|
|
|
|
def test_a_prompt_typed_in_the_app_reaches_the_node_that_encodes_it(server):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"inputs": {"positive": {"text": "a cat on a roof"}}})
|
|
assert status == 200, body
|
|
assert body["refused"] == []
|
|
positive = next(s for s in body["slots"] if s["id"] == "positive")
|
|
assert positive["inputs"]["text"] == "a cat on a roof"
|
|
assert positive["inputs"]["clip"] == ["clip", 0]
|
|
|
|
|
|
def test_a_value_for_a_slot_that_is_gone_stops_the_run(server):
|
|
"""Not a note: the prompt went nowhere, and a run without it is not the run
|
|
that was asked for."""
|
|
status, body = _request(server, "POST", "/fun/api/pipeline",
|
|
{"inputs": {"positive_2": {"text": "a cat"}}})
|
|
assert status == 200
|
|
assert any("no such slot" in p for p in body["refused"]), body["refused"]
|
|
assert body["queueable"] is False
|
|
assert body["prompt"] is None
|
|
|
|
|
|
def test_the_default_pipeline_says_where_its_prompt_belongs(server):
|
|
"""The app has no list of what a prompt is: the pipeline says which input
|
|
goes on the surface, and the app offers a place with that name or does not."""
|
|
status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
assert status == 200
|
|
roles = [(s["id"], r) for s in body["slots"] for r in (s.get("roles") or [])]
|
|
assert roles, "no slot offers anything to the main window"
|
|
places = {r["at"] for _id, r in roles}
|
|
assert places == {"generation.prompt", "project.video", "project.negative", "generation.sampling", "generation.seed"}
|
|
assert all(r["input"] and r["label"] for _id, r in roles)
|
|
|
|
def test_a_module_provided_preset_is_offered(server, registered):
|
|
"""MiniMax H3's Reference-to-Video pipeline, offered beside the app's one
|
|
true default -- not instead of it. See modules/models/minimax_h3/pipeline.py."""
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline/presets")
|
|
ids = [p["id"] for p in body["presets"]]
|
|
assert "minimax_h3_reference_to_video" in ids
|
|
preset = next(p for p in body["presets"] if p["id"] == "minimax_h3_reference_to_video")
|
|
assert preset["module"] == "model_minimax_h3"
|
|
assert preset["slots"]
|
|
|
|
|
|
def test_a_picture_model_preset_carries_the_project_numbers_it_goes_with(server, registered):
|
|
_status, body = _request(server, "GET", "/fun/api/pipeline/presets")
|
|
image = next(p for p in body["presets"] if p["id"] == "minimax_h3_image")
|
|
assert image["project"] == {"num_frames_per_scene": 1, "frame_rate": 1}
|
|
assert next(p for p in body["presets"] if p["id"] == "minimax_h3_reference_to_video")["project"] == {} # a video model changes none
|
|
|
|
|
|
def test_a_comfyui_workflow_imports_into_slots_that_build_a_real_graph(server, registered):
|
|
"""Over HTTP, against ComfyUI's own nodes: the API export of a plain text-to-image graph."""
|
|
from core import graph
|
|
wf = {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a fox", "clip": ["9", 0]}},
|
|
"2": {"class_type": "CLIPTextEncode", "inputs": {"text": "blur", "clip": ["9", 0]}},
|
|
"3": {"class_type": "EmptyLatentImage", "inputs": {"width": 512, "height": 512, "batch_size": 1}},
|
|
"4": {"class_type": "KSampler", "inputs": {"model": ["8", 0], "positive": ["1", 0], "negative": ["2", 0], "latent_image": ["3", 0],
|
|
"seed": 5, "steps": 4, "cfg": 7.0, "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0}}}
|
|
status, body = _request(server, "POST", "/fun/api/pipeline/import", {"workflow": wf})
|
|
assert status == 200
|
|
assert {"prompt", "negative", "seed", "width", "height"} <= set(body["bound"])
|
|
assert _request(server, "POST", "/fun/api/pipeline/import", [1, 2])[0] == 400
|
|
prompt, problems = graph.build(body["slots"])
|
|
assert set(prompt) >= {"w1", "w2", "w3", "w4"}
|
|
assert not [p for p in problems if "w4" in p and "positive" in p]
|
|
assert _request(server, "POST", "/fun/api/pipeline/import", {"workflow": {"nope": 1}})[0] == 400
|
|
|
|
|
|
def test_the_readiness_check_answers_with_rows_and_says_which_starting_point_lacks_nodes(server, registered):
|
|
status, body = _request(server, "GET", "/fun/api/readiness")
|
|
assert status == 200 and body["rows"]
|
|
assert all(r["level"] in ("ok", "warn", "fail") and r["text"] for r in body["rows"])
|
|
|
|
|
|
def test_the_manifest_says_which_modules_can_be_switched_and_a_quarantined_one_can_be_released(server):
|
|
from core import control, registry
|
|
status, body = _request(server, "GET", "/fun/api/modules")
|
|
assert status == 200 and body["control"]
|
|
sharp = registry.current().specs["sharpen"]
|
|
assert body["control"]["sharpen"]["controllable"] is True
|
|
control.quarantine(sharp, "RuntimeError: boom")
|
|
_, body = _request(server, "GET", "/fun/api/modules")
|
|
assert body["control"]["sharpen"]["quarantine"]["reason"].endswith("boom")
|
|
status, body = _request(server, "POST", "/fun/api/control/release", {"id": "sharpen"})
|
|
assert (status, body) == (200, {"released": True})
|
|
_, body = _request(server, "GET", "/fun/api/modules")
|
|
assert "quarantine" not in body["control"]["sharpen"]
|
|
assert _request(server, "POST", "/fun/api/control/release", {"id": "nope"})[0] == 400
|
|
|
|
|
|
def test_empty_values_do_not_wipe_the_settings_the_slots_carry(server):
|
|
"""The legacy Generate sends `values: {}` every time (its settings already sit in the
|
|
slots). Placing that empty object overwrote them, so every module ran switched off."""
|
|
_s, first = _request(server, "POST", "/fun/api/pipeline", {"values": {"sampling_alg": {"enabled": True}}})
|
|
carried = json.dumps({"sampling_alg": {"enabled": True}})
|
|
slots = [dict(s, inputs=dict(s["inputs"], settings=carried)) if s["node"] == "FunUIModifierSettings" else s
|
|
for s in first["slots"]]
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"slots": slots, "values": {}})
|
|
assert status == 200, body
|
|
settings = next(s for s in body["slots"] if s["node"] == "FunUIModifierSettings")
|
|
assert json.loads(settings["inputs"]["settings"]) == {"sampling_alg": {"enabled": True}}
|
|
|
|
|
|
def test_an_optional_node_that_is_not_installed_is_skipped_over_http_and_said(server, monkeypatch):
|
|
"""The call sites pass the table: the check route Generate uses, and the pipeline GET."""
|
|
monkeypatch.setattr(routes, "optional_nodes", lambda: {"NoSuchModuleNode": {"pass": {"0": "text"}}})
|
|
slots = [{"id": "gone", "node": "NoSuchModuleNode", "inputs": {"text": "typed prompt"}},
|
|
{"id": "use", "node": "PrimitiveString", "inputs": {"value": ["gone", 0]}}]
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"action": "check", "slots": slots})
|
|
assert status == 200, body
|
|
assert [s["slot"] for s in body["skipped"]] == ["gone"]
|
|
assert body["notes"][0].startswith("Skipped, its node is not available") and "NoSuchModuleNode" in body["notes"][0]
|
|
assert body["incomplete"] == [] and body["prompt"]["use"]["inputs"]["value"] == "typed prompt"
|
|
monkeypatch.setattr(routes, "optional_nodes", lambda: {})
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"action": "check", "slots": slots})
|
|
assert body["skipped"] == [] and any("no node called 'NoSuchModuleNode'" in p for p in body["incomplete"])
|