367 lines
14 KiB
Python
367 lines
14 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import inspect
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from tests.fake_comfy_api import install_fake_comfy_api
|
|
|
|
install_fake_comfy_api()
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _load_package():
|
|
package_name = "json_nodes_package_integration"
|
|
package_spec = importlib.util.spec_from_file_location(
|
|
package_name,
|
|
PROJECT_ROOT / "__init__.py",
|
|
submodule_search_locations=[str(PROJECT_ROOT)],
|
|
)
|
|
package = importlib.util.module_from_spec(package_spec)
|
|
sys.modules[package_name] = package
|
|
package_spec.loader.exec_module(package)
|
|
return package
|
|
|
|
|
|
async def _load_node_classes():
|
|
package = _load_package()
|
|
extension = await package.comfy_entrypoint()
|
|
node_classes = await extension.get_node_list()
|
|
return package, extension, node_classes
|
|
|
|
|
|
def _schema_default(input_def):
|
|
kwargs = getattr(input_def, "kwargs", {})
|
|
if "default" in kwargs:
|
|
return kwargs["default"]
|
|
options = kwargs.get("options", [])
|
|
if options:
|
|
return options[0]
|
|
kind = getattr(input_def, "kind", None)
|
|
if kind == "BOOLEAN":
|
|
return False
|
|
if kind == "STRING":
|
|
return ""
|
|
return None
|
|
|
|
|
|
def _build_args(schema, overrides):
|
|
args = []
|
|
for input_def in schema.inputs:
|
|
input_id = getattr(input_def, "id", None)
|
|
if input_id in overrides:
|
|
args.append(overrides[input_id])
|
|
continue
|
|
args.append(_schema_default(input_def))
|
|
return args
|
|
|
|
|
|
class TestComfyIntegration(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.package, cls.extension, cls.node_classes = asyncio.run(_load_node_classes())
|
|
cls.node_classes_by_name = {node_cls.__name__: node_cls for node_cls in cls.node_classes}
|
|
|
|
def test_real_entrypoint_loading_path_and_define_schema(self):
|
|
self.assertEqual(self.package.WEB_DIRECTORY, "./web/js")
|
|
self.assertEqual(
|
|
[node_cls.__name__ for node_cls in self.node_classes],
|
|
["JsonField", "QuickJsonNode", "JsonArrayMerge", "JsonObjectMerge", "JsonRootObject"],
|
|
)
|
|
|
|
node_ids = set()
|
|
for node_cls in self.node_classes:
|
|
schema = node_cls.define_schema()
|
|
self.assertNotIn(schema.node_id, node_ids)
|
|
node_ids.add(schema.node_id)
|
|
self.assertTrue(schema.inputs)
|
|
self.assertTrue(schema.outputs)
|
|
|
|
def test_every_node_executes_positionally_from_schema_inputs(self):
|
|
cases = {
|
|
"JsonField": {
|
|
"overrides": {
|
|
"key": "scene",
|
|
"type": "string",
|
|
"array_separator": "newline",
|
|
"value": "medina",
|
|
},
|
|
"output_type": dict,
|
|
},
|
|
"QuickJsonNode": {
|
|
"overrides": {
|
|
"entry_0": "scene: medina",
|
|
"entry_1": "mood: dusk",
|
|
"entry_2": "",
|
|
"entry_3": "",
|
|
"entry_4": "",
|
|
"entry_5": "",
|
|
"entry_6": "",
|
|
"entry_7": "",
|
|
"pretty_print": False,
|
|
},
|
|
"output_type": dict,
|
|
},
|
|
"JsonArrayMerge": {
|
|
"overrides": {
|
|
"items": {"item_1": "medina", "item_2": "42"},
|
|
"pretty_print": False,
|
|
},
|
|
"output_type": list,
|
|
},
|
|
"JsonObjectMerge": {
|
|
"overrides": {
|
|
"objects": {"json_object_1": '{"scene":"medina"}', "json_object_2": '{"mood":"dusk"}'},
|
|
"duplicate_key_policy": "error",
|
|
"pretty_print": False,
|
|
},
|
|
"output_type": dict,
|
|
},
|
|
"JsonRootObject": {
|
|
"overrides": {
|
|
"objects": {"json_input_1": '{"scene":"medina"}'},
|
|
"preview_json": "{}",
|
|
"duplicate_key_policy": "error",
|
|
"schema_json": '{"type":"object","required":["scene"],"properties":{"scene":{"type":"string"}}}',
|
|
"strict_mode": True,
|
|
"pretty_print": False,
|
|
},
|
|
"output_type": dict,
|
|
},
|
|
}
|
|
|
|
for node_cls in self.node_classes:
|
|
with self.subTest(node=node_cls.__name__):
|
|
schema = node_cls.define_schema()
|
|
case = cases[node_cls.__name__]
|
|
args = _build_args(schema, case["overrides"])
|
|
output = node_cls.execute(*args)
|
|
|
|
self.assertIsInstance(output, tuple)
|
|
self.assertEqual(len(output), len(schema.outputs))
|
|
self.assertIsInstance(output[0], str)
|
|
self.assertIsInstance(json.loads(output[0]), case["output_type"])
|
|
|
|
def test_json_field_lifecycle_supports_every_declared_type(self):
|
|
node_cls = self.node_classes_by_name["JsonField"]
|
|
cases = {
|
|
"string": ("medina", "medina"),
|
|
"int": ("85", 85),
|
|
"float": ("1.5", 1.5),
|
|
"boolean": ("true", True),
|
|
"array": ("alpha\nbeta", ["alpha", "beta"]),
|
|
"object": ('{"lens": 85}', {"lens": 85}),
|
|
}
|
|
|
|
for field_type, (raw_value, expected_value) in cases.items():
|
|
with self.subTest(field_type=field_type):
|
|
output = node_cls.execute("value", field_type, "newline", raw_value)
|
|
parsed = json.loads(output[0])
|
|
self.assertEqual(parsed["value"], expected_value)
|
|
|
|
def test_json_field_array_type_survives_missing_array_separator_arg(self):
|
|
node_cls = self.node_classes_by_name["JsonField"]
|
|
output = node_cls.execute("items", "array", "alpha\nbeta")
|
|
self.assertEqual(json.loads(output[0]), {"items": ["alpha", "beta"]})
|
|
|
|
def test_quick_json_lifecycle_with_partial_and_empty_slots(self):
|
|
node_cls = self.node_classes_by_name["QuickJsonNode"]
|
|
output = node_cls.execute(
|
|
"scene: medina",
|
|
"",
|
|
"mood: dusk",
|
|
"",
|
|
"",
|
|
"camera: 50mm",
|
|
"",
|
|
"",
|
|
False,
|
|
)
|
|
self.assertEqual(
|
|
json.loads(output[0]),
|
|
{"scene": "medina", "mood": "dusk", "camera": "50mm"},
|
|
)
|
|
|
|
def test_quick_json_all_empty_entries_return_empty_object(self):
|
|
node_cls = self.node_classes_by_name["QuickJsonNode"]
|
|
output = node_cls.execute("", "", "", "", "", "", "", "", False)
|
|
self.assertEqual(json.loads(output[0]), {})
|
|
|
|
def test_array_merge_lifecycle_parses_json_and_plain_text_values(self):
|
|
node_cls = self.node_classes_by_name["JsonArrayMerge"]
|
|
output = node_cls.execute(
|
|
{
|
|
"item_10": "false",
|
|
"item_2": '{"mood":"dusk"}',
|
|
"item_1": "medina",
|
|
"item_3": "42",
|
|
},
|
|
False,
|
|
)
|
|
self.assertEqual(json.loads(output[0]), ["medina", {"mood": "dusk"}, 42, False])
|
|
|
|
def test_merge_lifecycle_merges_two_valid_objects(self):
|
|
node_cls = self.node_classes_by_name["JsonObjectMerge"]
|
|
output = node_cls.execute(
|
|
{"json_object_1": '{"scene":"medina"}', "json_object_2": '{"mood":"dusk"}'},
|
|
"error",
|
|
False,
|
|
)
|
|
self.assertEqual(json.loads(output[0]), {"scene": "medina", "mood": "dusk"})
|
|
|
|
def test_merge_lifecycle_merges_multiple_valid_objects(self):
|
|
node_cls = self.node_classes_by_name["JsonObjectMerge"]
|
|
output = node_cls.execute(
|
|
{
|
|
"json_object_1": '{"scene":"medina"}',
|
|
"json_object_2": '{"mood":"dusk"}',
|
|
"json_object_3": '{"camera":{"lens":85}}',
|
|
},
|
|
"error",
|
|
False,
|
|
)
|
|
self.assertEqual(
|
|
json.loads(output[0]),
|
|
{"scene": "medina", "mood": "dusk", "camera": {"lens": 85}},
|
|
)
|
|
|
|
def test_merge_lifecycle_accepts_legacy_object_input_names(self):
|
|
node_cls = self.node_classes_by_name["JsonObjectMerge"]
|
|
output = node_cls.execute({"object_1": '{"scene":"medina"}'}, "error", False)
|
|
self.assertEqual(json.loads(output[0]), {"scene": "medina"})
|
|
|
|
def test_root_lifecycle_with_schema_and_strict_mode(self):
|
|
node_cls = self.node_classes_by_name["JsonRootObject"]
|
|
output = node_cls.execute(
|
|
{"json_input_1": '{"scene":"medina","camera":{"lens":85}}'},
|
|
"{}",
|
|
"overwrite",
|
|
json.dumps(
|
|
{
|
|
"type": "object",
|
|
"required": ["scene", "camera"],
|
|
"properties": {
|
|
"scene": {"type": "string"},
|
|
"camera": {
|
|
"type": "object",
|
|
"required": ["lens"],
|
|
"properties": {"lens": {"type": "int"}},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
True,
|
|
False,
|
|
)
|
|
self.assertEqual(json.loads(output[0]), {"scene": "medina", "camera": {"lens": 85}})
|
|
|
|
def test_root_can_execute_twice_with_same_inputs(self):
|
|
node_cls = self.node_classes_by_name["JsonRootObject"]
|
|
args = (
|
|
{"json_input_1": '{"scene":"medina"}'},
|
|
"{}",
|
|
"overwrite",
|
|
'{"type":"object","required":["scene"],"properties":{"scene":{"type":"string"}}}',
|
|
True,
|
|
False,
|
|
)
|
|
|
|
first = node_cls.execute(*args)
|
|
second = node_cls.execute(*args)
|
|
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(json.loads(second[0]), {"scene": "medina"})
|
|
|
|
def test_array_merge_to_field_to_root_pipeline(self):
|
|
field_node = self.node_classes_by_name["JsonField"]
|
|
array_node = self.node_classes_by_name["JsonArrayMerge"]
|
|
root_node = self.node_classes_by_name["JsonRootObject"]
|
|
|
|
(first_subject,) = field_node.execute("name", "string", "newline", "Alice")
|
|
(second_subject,) = field_node.execute("name", "string", "newline", "Bob")
|
|
(subjects_array,) = array_node.execute({"item_2": second_subject, "item_1": first_subject}, False)
|
|
(subjects_field,) = field_node.execute("subjects", "array", "newline", subjects_array)
|
|
|
|
output = root_node.execute(
|
|
{"json_input_1": subjects_field},
|
|
"{}",
|
|
"overwrite",
|
|
json.dumps(
|
|
{
|
|
"type": "object",
|
|
"required": ["subjects"],
|
|
"properties": {
|
|
"subjects": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"required": ["name"],
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
},
|
|
},
|
|
}
|
|
},
|
|
}
|
|
),
|
|
True,
|
|
False,
|
|
)
|
|
|
|
self.assertEqual(json.loads(output[0]), {"subjects": [{"name": "Alice"}, {"name": "Bob"}]})
|
|
|
|
def test_schema_optional_and_default_contract_matches_execute_signature(self):
|
|
for node_cls in self.node_classes:
|
|
with self.subTest(node=node_cls.__name__):
|
|
schema = node_cls.define_schema()
|
|
params = list(inspect.signature(node_cls.execute.__func__).parameters.values())[1:]
|
|
self.assertEqual([input_def.id for input_def in schema.inputs], [param.name for param in params])
|
|
|
|
for input_def, param in zip(schema.inputs, params, strict=True):
|
|
kwargs = getattr(input_def, "kwargs", {})
|
|
schema_has_default = "default" in kwargs
|
|
schema_default = kwargs.get("default")
|
|
is_optional = bool(getattr(input_def, "optional", False))
|
|
sig_has_default = param.default is not inspect._empty
|
|
|
|
if is_optional:
|
|
self.assertTrue(
|
|
sig_has_default,
|
|
f"{schema.node_id}.{input_def.id} is optional in schema but missing an execute default",
|
|
)
|
|
|
|
if schema_has_default:
|
|
self.assertTrue(
|
|
sig_has_default,
|
|
f"{schema.node_id}.{input_def.id} has a schema default but missing an execute default",
|
|
)
|
|
self.assertEqual(
|
|
param.default,
|
|
schema_default,
|
|
f"{schema.node_id}.{input_def.id} execute default does not match schema default",
|
|
)
|
|
|
|
if sig_has_default:
|
|
self.assertTrue(
|
|
is_optional or schema_has_default,
|
|
f"{schema.node_id}.{input_def.id} has an execute default without matching schema metadata",
|
|
)
|
|
|
|
def test_frontend_extension_js_parses_without_syntax_errors(self):
|
|
result = subprocess.run(
|
|
["node", "--check", str(PROJECT_ROOT / "web" / "js" / "jsonNodes.js")],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr or result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|