Vito Sansevero
c65b2d5449
Code scanning flagged the mixed 'import unittest' and 'from unittest import mock' forms. Use 'import unittest' plus 'import unittest.mock' everywhere.
839 lines
32 KiB
Python
839 lines
32 KiB
Python
"""The declarative inspection node semantics table and its loader (#240, spec 7).
|
|
|
|
``web/js/inspection-semantics.json`` is the single table. Python validates it at
|
|
import and fails loudly; the browser facts tables are checked against its
|
|
``facts`` section by ``tests/js/inspection-semantics-sync.test.js``.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
|
|
from utils import inspection_semantics as semantics
|
|
from utils.inspection_semantics import SemanticsError, load_semantics
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
TABLE = REPO / "web" / "js" / "inspection-semantics.json"
|
|
|
|
|
|
def _minimal():
|
|
"""A small valid table; mutation tests break exactly one thing in a copy."""
|
|
return {
|
|
"schema": 1,
|
|
"samplers": {
|
|
"KSampler": {
|
|
"strategy": "ksampler",
|
|
"conditioning": {"positive": "positive", "negative": "negative"},
|
|
"latent": "latent_image",
|
|
"facts": "ksampler",
|
|
}
|
|
},
|
|
"guiders": {"BasicGuider": {"strategy": "basic", "positive": "conditioning"}},
|
|
"conditioning": [
|
|
{
|
|
"class": "FluxGuidance",
|
|
"slot": 0,
|
|
"strategy": "passthrough",
|
|
"input": "conditioning",
|
|
}
|
|
],
|
|
"selectors": {},
|
|
"composites": {},
|
|
"encoders": [
|
|
{
|
|
"class": "CLIPTextEncode",
|
|
"slot": 0,
|
|
"strategy": "single",
|
|
"inputs": ["text"],
|
|
}
|
|
],
|
|
"strings": [],
|
|
"runtime": [],
|
|
"savers": {},
|
|
"non_file_outputs": [],
|
|
"stage_inputs": {"shaped": ["positive"], "with_model": ["samples"]},
|
|
"facts": {
|
|
"configuration_outputs": {},
|
|
"model_outputs": {},
|
|
"model_passthroughs": [],
|
|
"scalar_outputs": {},
|
|
},
|
|
}
|
|
|
|
|
|
class _TableCase(unittest.TestCase):
|
|
def load_text(self, text):
|
|
# Windows: the file handle is closed before the directory is removed.
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path = Path(root) / "semantics.json"
|
|
path.write_text(text, encoding="utf-8")
|
|
return load_semantics(path)
|
|
|
|
def load_doc(self, doc):
|
|
return self.load_text(json.dumps(doc))
|
|
|
|
def assertRejected(self, doc, fragment):
|
|
with self.assertRaises(SemanticsError) as caught:
|
|
self.load_doc(doc)
|
|
self.assertIn(fragment, str(caught.exception))
|
|
|
|
|
|
class SpecEntriesTest(unittest.TestCase):
|
|
def test_semantics_file_validates_and_lists_spec_entries(self):
|
|
table = load_semantics()
|
|
|
|
self.assertEqual(semantics.SEMANTICS_PATH, TABLE)
|
|
self.assertEqual(
|
|
semantics.SEMANTICS_PATH,
|
|
Path(semantics.__file__).resolve().parents[1]
|
|
/ "web"
|
|
/ "js"
|
|
/ "inspection-semantics.json",
|
|
)
|
|
|
|
pair = {"positive": "positive", "negative": "negative"}
|
|
for name in ("KSampler", "KSamplerAdvanced"):
|
|
self.assertEqual(
|
|
table.samplers[name],
|
|
{
|
|
"strategy": "ksampler",
|
|
"conditioning": pair,
|
|
"latent": "latent_image",
|
|
"facts": "ksampler",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
table.samplers["SamplerCustom"],
|
|
{
|
|
"strategy": "ksampler",
|
|
"conditioning": pair,
|
|
"latent": "latent_image",
|
|
"facts": "custom",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
table.samplers["SamplerCustomAdvanced"],
|
|
{
|
|
"strategy": "custom-advanced",
|
|
"conditioning": {"guider": "guider"},
|
|
"latent": "latent_image",
|
|
"facts": "custom-advanced",
|
|
},
|
|
)
|
|
# D19: FaceDetailer is a stage over its upstream image, ksampler-shaped facts.
|
|
# A non-empty wildcard replaces the detail prompt (Impact core.py:267).
|
|
self.assertEqual(
|
|
table.samplers["FaceDetailer"],
|
|
{
|
|
"strategy": "detailer-stage",
|
|
"conditioning": pair,
|
|
"image_stage": "image",
|
|
"facts": "ksampler",
|
|
"prompt_overrides": ("wildcard",),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
table.guiders["CFGGuider"],
|
|
{"strategy": "cfg", "positive": "positive", "negative": "negative"},
|
|
)
|
|
self.assertEqual(
|
|
table.guiders["BasicGuider"],
|
|
{
|
|
"strategy": "basic",
|
|
"positive": "conditioning",
|
|
"absent_reason": "Guider has no negative input",
|
|
},
|
|
)
|
|
# D17: DualCFGGuider has two positives and stays an unsupported node.
|
|
self.assertNotIn("DualCFGGuider", table.guiders)
|
|
self.assertNotIn("DualCFGGuider", table.all_classes)
|
|
|
|
single_input = (
|
|
"FluxGuidance",
|
|
"ReferenceLatent",
|
|
"ConditioningSetArea",
|
|
"ConditioningSetAreaPercentage",
|
|
"ConditioningSetAreaStrength",
|
|
"ConditioningSetMask",
|
|
"ConditioningSetTimestepRange",
|
|
"ControlNetApply",
|
|
)
|
|
for name in single_input:
|
|
self.assertEqual(
|
|
table.conditioning[(name, 0)],
|
|
{"strategy": "passthrough", "input": "conditioning"},
|
|
)
|
|
self.assertNotIn((name, 1), table.conditioning)
|
|
for name in (
|
|
"ControlNetApplyAdvanced",
|
|
"ControlNetApplySD3",
|
|
"InpaintModelConditioning",
|
|
"WanImageToVideo",
|
|
):
|
|
self.assertEqual(
|
|
table.conditioning[(name, 0)],
|
|
{"strategy": "passthrough", "input": "positive"},
|
|
)
|
|
self.assertEqual(
|
|
table.conditioning[(name, 1)],
|
|
{"strategy": "passthrough", "input": "negative"},
|
|
)
|
|
# Slot 2 (a latent) is not conditioning; spec 7.4 handles origins.
|
|
self.assertNotIn((name, 2), table.conditioning)
|
|
self.assertEqual(
|
|
table.conditioning[("ConditioningZeroOut", 0)],
|
|
{"strategy": "zeroed", "input": "conditioning"},
|
|
)
|
|
|
|
self.assertEqual(
|
|
table.selectors["easy ifElse"],
|
|
{
|
|
"strategy": "boolean-select",
|
|
"condition": "boolean",
|
|
"branches": {"true": "on_true", "false": "on_false"},
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
table.composites["ConditioningCombine"],
|
|
{"strategy": "combine", "inputs": ("conditioning_1", "conditioning_2")},
|
|
)
|
|
for name in ("ConditioningConcat", "ConditioningAverage"):
|
|
self.assertEqual(
|
|
table.composites[name],
|
|
{
|
|
"strategy": "combine",
|
|
"inputs": ("conditioning_to", "conditioning_from"),
|
|
},
|
|
)
|
|
|
|
encoders = {
|
|
("CLIPTextEncode", 0): ("single", ("text",)),
|
|
("CLIPTextEncodeSDXL", 0): ("parts-equal", ("text_g", "text_l")),
|
|
("CLIPTextEncodeFlux", 0): ("parts-equal", ("clip_l", "t5xxl")),
|
|
("CLIPTextEncodeSD3", 0): ("parts-equal", ("clip_l", "clip_g", "t5xxl")),
|
|
("TextEncodeQwenImageEdit", 0): ("single", ("prompt",)),
|
|
("TextEncodeQwenImageEditPlus", 0): ("single", ("prompt",)),
|
|
("TextEncodeQwenImage21", 0): ("single", ("prompt",)),
|
|
("TextEncodeQwenImage21", 1): ("single", ("negative_prompt",)),
|
|
# D11: a plain text encoder, no composition.
|
|
("KikoTextEncode", 0): ("single", ("text",)),
|
|
}
|
|
for key, (strategy, inputs) in encoders.items():
|
|
self.assertEqual(
|
|
table.encoders[key], {"strategy": strategy, "inputs": inputs}, key
|
|
)
|
|
self.assertNotIn(("TextEncodeQwenImage21", 2), table.encoders)
|
|
|
|
self.assertEqual(
|
|
table.strings[("StringConstant", 0)],
|
|
{"strategy": "literal", "inputs": ("string",)},
|
|
)
|
|
self.assertEqual(
|
|
table.strings[("StringConstantMultiline", 0)],
|
|
{
|
|
"strategy": "strip-newlines",
|
|
"inputs": ("string",),
|
|
"params": {"flag": "strip_newlines"},
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
table.strings[("JoinStringMulti", 0)],
|
|
{
|
|
"strategy": "join-multi",
|
|
"inputs": ("string_1",),
|
|
"params": {
|
|
"count": "inputcount",
|
|
"delimiter": "delimiter",
|
|
"list_flag": "return_list",
|
|
"input_prefix": "string_",
|
|
},
|
|
},
|
|
)
|
|
|
|
# D18: decided at run time.
|
|
self.assertEqual(table.runtime, frozenset({"PromptSearchList"}))
|
|
|
|
comfy = "{name}_{counter:05}_.{ext}"
|
|
for name, image in (
|
|
("SaveImage", "images"),
|
|
("SaveAnimatedWEBP", "images"),
|
|
("SaveAnimatedPNG", "images"),
|
|
("SaveVideo", "video"),
|
|
("SaveWEBM", "images"),
|
|
):
|
|
self.assertEqual(
|
|
table.savers[name],
|
|
{
|
|
"strategy": "comfy-counter",
|
|
"prefix_input": "filename_prefix",
|
|
"image_input": image,
|
|
"pattern": comfy,
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
table.savers["VHS_VideoCombine"],
|
|
{
|
|
"strategy": "vhs-counter",
|
|
"prefix_input": "filename_prefix",
|
|
"image_input": "images",
|
|
"pattern": "{name}_{counter:05}.{ext}",
|
|
},
|
|
)
|
|
# Until its naming is verified in pinned source it stays terminal-unknown.
|
|
self.assertNotIn("KikoSaveImage", table.all_classes)
|
|
self.assertEqual(
|
|
table.non_file_outputs,
|
|
frozenset(
|
|
{
|
|
"PreviewImage",
|
|
"PreviewAny",
|
|
"SaveImageWebsocket",
|
|
"Image Comparer (rgthree)",
|
|
}
|
|
),
|
|
)
|
|
|
|
# Not listed this slice: they stay unsupported-node with the class named.
|
|
for name in (
|
|
"ImpactWildcardProcessor",
|
|
"RegexReplace",
|
|
"LoRAfromText",
|
|
"XYZPrompt",
|
|
):
|
|
self.assertNotIn(name, table.all_classes)
|
|
# Existing code adapters stay in code, unchanged.
|
|
for name in ("PromptManager", "PromptManagerPair", "PromptManagerText"):
|
|
self.assertNotIn(name, table.all_classes)
|
|
|
|
def test_legacy_parser_ports_are_listed_with_their_source_inputs(self):
|
|
# Task 2.7b (R6): input names and slots read from each node's source;
|
|
# the report lists the file and line for every one.
|
|
table = load_semantics()
|
|
pair = {"positive": "positive", "negative": "negative"}
|
|
for name, image in (
|
|
("DetailerForEach", "image"),
|
|
("UltimateSDUpscale", "image"),
|
|
("UltimateSDUpscaleNoUpscale", "upscaled_image"),
|
|
):
|
|
entry = dict(table.samplers[name])
|
|
self.assertEqual(
|
|
entry.pop("prompt_overrides", None) is not None,
|
|
name == "DetailerForEach",
|
|
name,
|
|
)
|
|
self.assertEqual(
|
|
entry,
|
|
{
|
|
"strategy": "detailer-stage",
|
|
"conditioning": pair,
|
|
"image_stage": image,
|
|
"facts": "ksampler",
|
|
},
|
|
name,
|
|
)
|
|
for (name, slot), source in {
|
|
("ControlNetInpaintingAliMamaApply", 0): "positive",
|
|
("ControlNetInpaintingAliMamaApply", 1): "negative",
|
|
("LTXVConditioning", 0): "positive",
|
|
("LTXVConditioning", 1): "negative",
|
|
("FluxKontextMultiReferenceLatentMethod", 0): "conditioning",
|
|
}.items():
|
|
self.assertEqual(
|
|
table.conditioning[(name, slot)],
|
|
{"strategy": "passthrough", "input": source},
|
|
)
|
|
self.assertNotIn(
|
|
("FluxKontextMultiReferenceLatentMethod", 1), table.conditioning
|
|
)
|
|
self.assertEqual(
|
|
table.encoders[("Text to Conditioning", 0)],
|
|
{"strategy": "single", "inputs": ("text",)},
|
|
)
|
|
for name in ("TextInput", "DisplayText"):
|
|
self.assertEqual(
|
|
table.strings[(name, 0)], {"strategy": "literal", "inputs": ("text",)}
|
|
)
|
|
|
|
def test_stage_inputs_are_the_one_source_for_both_stage_walks(self):
|
|
# Review I-2: Python's resolver and comfy-metadata.js read these lists here.
|
|
stage_inputs = load_semantics().stage_inputs
|
|
self.assertEqual(
|
|
stage_inputs["shaped"],
|
|
(
|
|
"latent_image",
|
|
"guider",
|
|
"positive",
|
|
"negative",
|
|
"pipe",
|
|
"basic_pipe",
|
|
"detailer_pipe",
|
|
"sampler",
|
|
"sigmas",
|
|
"noise",
|
|
"text_embeds",
|
|
"hyvid_embeds",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
stage_inputs["with_model"],
|
|
("conditioning", "neg_conditioning", "samples", "latent"),
|
|
)
|
|
# The default profile's own sampler-shape names stay a subset of the table's.
|
|
from utils.prompt_pair_resolver import _SAMPLER_LIKE_INPUTS
|
|
|
|
self.assertLessEqual(set(_SAMPLER_LIKE_INPUTS), set(stage_inputs["shaped"]))
|
|
|
|
def test_facts_section_mirrors_the_browser_tables(self):
|
|
facts = load_semantics().facts
|
|
self.assertEqual(
|
|
facts["configuration_outputs"],
|
|
{
|
|
"RandomNoise": {"kind": "noise", "fields": ("noise_seed",)},
|
|
"KSamplerSelect": {"kind": "sampler", "fields": ("sampler_name",)},
|
|
"BasicScheduler": {"kind": "sigmas", "fields": ("steps", "scheduler")},
|
|
# D8: Flux2Scheduler has no scheduler input; its type names it.
|
|
"Flux2Scheduler": {
|
|
"kind": "sigmas",
|
|
"fields": ("steps",),
|
|
"type_fields": ("scheduler",),
|
|
},
|
|
"CFGGuider": {"kind": "guider", "fields": ("cfg", "model")},
|
|
"BasicGuider": {"kind": "guider", "fields": ("model",)},
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
facts["model_outputs"],
|
|
{
|
|
"CheckpointLoader": "ckpt_name",
|
|
"CheckpointLoaderSimple": "ckpt_name",
|
|
"unCLIPCheckpointLoader": "ckpt_name",
|
|
"UNETLoader": "unet_name",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
facts["model_passthroughs"],
|
|
(
|
|
"LoraLoader",
|
|
"LoraLoaderModelOnly",
|
|
"ModelSamplingFlux",
|
|
"ModelSamplingAuraFlow",
|
|
"ModelSamplingSD3",
|
|
# Task 2.7b (R9): model in, model out, verified in their sources.
|
|
"Lora Loader (LoraManager)",
|
|
"Power Lora Loader (rgthree)",
|
|
"ModelPatchTorchSettings",
|
|
"CFGNorm",
|
|
"PerturbedAttentionGuidance",
|
|
"CR Apply LoRA Stack",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
facts["scalar_outputs"],
|
|
{
|
|
"PrimitiveInt": "number",
|
|
"PrimitiveFloat": "number",
|
|
"PrimitiveString": "string",
|
|
"PrimitiveStringMultiline": "string",
|
|
},
|
|
)
|
|
|
|
def test_table_is_loaded_lazily_and_once(self):
|
|
# Importing the module reads nothing; the inspection path asks for it.
|
|
self.assertFalse(hasattr(semantics, "SEMANTICS"))
|
|
semantics.get_semantics.cache_clear()
|
|
first = semantics.get_semantics()
|
|
self.assertIs(first, semantics.get_semantics())
|
|
self.assertEqual(first, load_semantics())
|
|
|
|
def test_loaded_table_is_frozen(self):
|
|
table = load_semantics()
|
|
self.assertIsInstance(table.samplers, MappingProxyType)
|
|
self.assertIsInstance(table.samplers["KSampler"], MappingProxyType)
|
|
with self.assertRaises(TypeError):
|
|
table.samplers["Other"] = {}
|
|
with self.assertRaises(TypeError):
|
|
table.samplers["KSampler"]["latent"] = "x"
|
|
with self.assertRaises(AttributeError):
|
|
table.samplers = {}
|
|
self.assertIsInstance(table.composites["ConditioningCombine"]["inputs"], tuple)
|
|
|
|
|
|
class ValidationTest(_TableCase):
|
|
def test_minimal_table_loads(self):
|
|
table = self.load_doc(_minimal())
|
|
self.assertEqual(table.encoders[("CLIPTextEncode", 0)]["inputs"], ("text",))
|
|
self.assertEqual(table.runtime, frozenset())
|
|
|
|
def test_unknown_strategy_rejected(self):
|
|
for category, mutate in (
|
|
("samplers", lambda d: d["samplers"]["KSampler"].update(strategy="dual")),
|
|
# A known strategy of another category is still unknown here.
|
|
("conditioning", lambda d: d["conditioning"][0].update(strategy="single")),
|
|
("encoders", lambda d: d["encoders"][0].update(strategy="guess")),
|
|
("guiders", lambda d: d["guiders"]["BasicGuider"].update(strategy=None)),
|
|
):
|
|
with self.subTest(category=category):
|
|
doc = _minimal()
|
|
mutate(doc)
|
|
self.assertRejected(doc, "unknown strategy")
|
|
|
|
def test_strategies_are_the_closed_spec_set(self):
|
|
self.assertEqual(
|
|
dict(semantics.STRATEGIES),
|
|
{
|
|
"samplers": frozenset(
|
|
{"ksampler", "custom-advanced", "detailer-stage"}
|
|
),
|
|
"guiders": frozenset({"cfg", "basic"}),
|
|
"conditioning": frozenset({"passthrough", "zeroed"}),
|
|
"selectors": frozenset({"boolean-select"}),
|
|
"composites": frozenset({"combine"}),
|
|
"encoders": frozenset(
|
|
{"single", "parts-equal", "pm-compose", "flag-parts"}
|
|
),
|
|
"strings": frozenset(
|
|
{"literal", "strip-newlines", "join-multi", "concat"}
|
|
),
|
|
"savers": frozenset({"comfy-counter", "vhs-counter"}),
|
|
},
|
|
)
|
|
with self.assertRaises(TypeError):
|
|
semantics.STRATEGIES["samplers"] = frozenset()
|
|
|
|
def test_duplicate_output_key_rejected(self):
|
|
doc = _minimal()
|
|
doc["encoders"].append(copy.deepcopy(doc["encoders"][0]))
|
|
self.assertRejected(doc, "duplicate")
|
|
|
|
doc = _minimal()
|
|
doc["conditioning"].append(
|
|
{
|
|
"class": "CLIPTextEncode",
|
|
"slot": 0,
|
|
"strategy": "passthrough",
|
|
"input": "conditioning",
|
|
}
|
|
)
|
|
self.assertRejected(doc, "duplicate")
|
|
|
|
doc = _minimal()
|
|
doc["runtime"] = ["KSampler"]
|
|
self.assertRejected(doc, "duplicate")
|
|
|
|
doc = _minimal()
|
|
doc["non_file_outputs"] = ["PreviewImage", "PreviewImage"]
|
|
self.assertRejected(doc, "duplicate")
|
|
|
|
# JSON objects with a repeated key would silently keep the last one.
|
|
text = json.dumps(_minimal())
|
|
entry = json.dumps(_minimal()["samplers"]["KSampler"])
|
|
text = text.replace(
|
|
'"samplers": {', '"samplers": {"KSampler": ' + entry + ", ", 1
|
|
)
|
|
with self.assertRaises(SemanticsError) as caught:
|
|
self.load_text(text)
|
|
self.assertIn("duplicate", str(caught.exception))
|
|
|
|
def test_slot_must_be_nonnegative_int(self):
|
|
for slot in (-1, 1.5, True, "0", None, 2**31):
|
|
with self.subTest(slot=slot):
|
|
doc = _minimal()
|
|
doc["encoders"][0]["slot"] = slot
|
|
self.assertRejected(doc, "slot")
|
|
doc = _minimal()
|
|
del doc["conditioning"][0]["slot"]
|
|
self.assertRejected(doc, "slot")
|
|
|
|
def test_missing_input_name_rejected(self):
|
|
cases = (
|
|
lambda d: d["samplers"]["KSampler"]["conditioning"].pop("negative"),
|
|
lambda d: d["samplers"]["KSampler"].pop("latent"),
|
|
lambda d: d["guiders"]["BasicGuider"].update(positive=""),
|
|
lambda d: d["conditioning"][0].pop("input"),
|
|
lambda d: d["encoders"][0].update(inputs=[]),
|
|
lambda d: d["encoders"][0].update(inputs=["text", "text"]),
|
|
lambda d: d["encoders"][0].update(inputs=[3]),
|
|
lambda d: d["encoders"][0].update(inputs="text"),
|
|
)
|
|
for index, mutate in enumerate(cases):
|
|
with self.subTest(case=index):
|
|
doc = _minimal()
|
|
mutate(doc)
|
|
self.assertRejected(doc, "input")
|
|
|
|
def test_entry_shape_is_strict(self):
|
|
cases = (
|
|
(lambda d: d.update(extra={}), "category"),
|
|
(lambda d: d.pop("guiders"), "category"),
|
|
(lambda d: d.update(schema=2), "schema"),
|
|
(lambda d: d["samplers"]["KSampler"].update(note="x"), "field"),
|
|
(lambda d: d["samplers"]["KSampler"].update(facts="flux"), "facts"),
|
|
(lambda d: d["guiders"]["BasicGuider"].update(negative="n"), "field"),
|
|
(lambda d: d["encoders"][0].update({"class": ""}), "class"),
|
|
(lambda d: d["facts"].update(extra={}), "facts"),
|
|
(
|
|
lambda d: d["facts"]["scalar_outputs"].update(PrimitiveInt="int"),
|
|
"facts",
|
|
),
|
|
(lambda d: d.pop("stage_inputs"), "category"),
|
|
(lambda d: d["stage_inputs"].update(extra=["x"]), "stage_inputs"),
|
|
(lambda d: d["stage_inputs"].pop("with_model"), "stage_inputs"),
|
|
(lambda d: d["stage_inputs"].update(shaped=[]), "input name"),
|
|
(lambda d: d["stage_inputs"].update(shaped="pipe"), "input name"),
|
|
(lambda d: d["stage_inputs"].update(with_model=["positive"]), "both"),
|
|
(lambda d: d["stage_inputs"].update(shaped=["model"]), "model"),
|
|
(
|
|
lambda d: d["facts"]["configuration_outputs"].update(
|
|
X={"kind": "sigmas", "fields": ["steps"], "type_fields": []}
|
|
),
|
|
"input name",
|
|
),
|
|
(
|
|
lambda d: d["facts"]["configuration_outputs"].update(
|
|
X={"kind": "sigmas", "fields": ["steps"], "type_fields": ["steps"]}
|
|
),
|
|
"type_fields",
|
|
),
|
|
(
|
|
lambda d: d["facts"]["configuration_outputs"].update(
|
|
X={"kind": "sigmas", "fields": ["steps"], "names": ["x"]}
|
|
),
|
|
"field",
|
|
),
|
|
)
|
|
for index, (mutate, fragment) in enumerate(cases):
|
|
with self.subTest(case=index):
|
|
doc = _minimal()
|
|
mutate(doc)
|
|
self.assertRejected(doc, fragment)
|
|
|
|
def test_strategy_specific_requirements(self):
|
|
doc = _minimal()
|
|
doc["samplers"]["X"] = {
|
|
"strategy": "custom-advanced",
|
|
"conditioning": {"positive": "positive", "negative": "negative"},
|
|
"latent": "latent_image",
|
|
"facts": "custom-advanced",
|
|
}
|
|
self.assertRejected(doc, "input")
|
|
|
|
doc = _minimal()
|
|
doc["samplers"]["D"] = {
|
|
"strategy": "detailer-stage",
|
|
"conditioning": {"positive": "positive", "negative": "negative"},
|
|
"facts": "ksampler",
|
|
}
|
|
self.assertRejected(doc, "input")
|
|
|
|
doc = _minimal()
|
|
doc["guiders"]["G"] = {"strategy": "cfg", "positive": "positive"}
|
|
self.assertRejected(doc, "input")
|
|
|
|
doc = _minimal()
|
|
doc["encoders"][0].update(strategy="parts-equal", inputs=["text_g"])
|
|
self.assertRejected(doc, "input")
|
|
|
|
doc = _minimal()
|
|
doc["savers"]["S"] = {
|
|
"strategy": "comfy-counter",
|
|
"prefix_input": "filename_prefix",
|
|
"image_input": "images",
|
|
"pattern": "{name}_{counter:05}.{ext}",
|
|
}
|
|
self.assertRejected(doc, "pattern")
|
|
|
|
doc = _minimal()
|
|
doc["selectors"]["easy ifElse"] = {
|
|
"strategy": "boolean-select",
|
|
"condition": "boolean",
|
|
"branches": {"true": "on_true"},
|
|
}
|
|
self.assertRejected(doc, "input")
|
|
|
|
doc = _minimal()
|
|
doc["strings"].append(
|
|
{
|
|
"class": "JoinStringMulti",
|
|
"slot": 0,
|
|
"strategy": "join-multi",
|
|
"inputs": ["string_1"],
|
|
"params": {"count": "inputcount"},
|
|
}
|
|
)
|
|
self.assertRejected(doc, "params")
|
|
|
|
def test_unreadable_table_fails_loudly(self):
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_text("{not json")
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_text("[]")
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_text(
|
|
json.dumps(_minimal()).replace('"schema": 1', '"schema": NaN')
|
|
)
|
|
with tempfile.TemporaryDirectory() as root:
|
|
with self.assertRaises(SemanticsError):
|
|
load_semantics(Path(root) / "missing.json")
|
|
|
|
def test_oversized_table_is_rejected_before_parsing(self):
|
|
text = json.dumps(_minimal()) + " " * semantics.MAX_TABLE_BYTES
|
|
with self.assertRaises(SemanticsError) as caught:
|
|
self.load_text(text)
|
|
self.assertIn("exceeds", str(caught.exception))
|
|
|
|
def test_deeply_nested_table_is_a_semantics_error(self):
|
|
# Past the nesting bound, and far past the interpreter's recursion limit.
|
|
for depth in (semantics.MAX_NESTING + 1, 100_000):
|
|
with self.subTest(depth=depth):
|
|
nested = "[" * depth + "]" * depth
|
|
text = json.dumps(_minimal()).replace(
|
|
'"selectors": {}', '"selectors": ' + nested
|
|
)
|
|
with self.assertRaises(SemanticsError) as caught:
|
|
self.load_text(text)
|
|
self.assertIn("nest", str(caught.exception))
|
|
self.assertGreater(semantics.MAX_NESTING, _depth(json.loads(TABLE.read_text())))
|
|
# Whichever interpreter limit the parser hits, it is a SemanticsError.
|
|
with unittest.mock.patch.object(
|
|
semantics.json, "loads", side_effect=RecursionError
|
|
):
|
|
with self.assertRaises(SemanticsError) as caught:
|
|
self.load_doc(_minimal())
|
|
self.assertIn("nests deeper", str(caught.exception))
|
|
|
|
|
|
def _depth(value, level=1):
|
|
children = value.values() if isinstance(value, dict) else value
|
|
if not isinstance(value, (dict, list)):
|
|
return level - 1
|
|
return max([level] + [_depth(item, level + 1) for item in children])
|
|
|
|
|
|
class SamplerOptionsTest(_TableCase):
|
|
"""Task 2.7b: optional sampler lists are input names, or the table fails."""
|
|
|
|
def test_upstream_verified_ports_are_listed_with_their_source_inputs(self):
|
|
# Not installed here: read from the pinned upstream commits the saved
|
|
# workflows name (the report lists repository, commit, file and line).
|
|
table = load_semantics()
|
|
pair = {"positive": "positive", "negative": "negative"}
|
|
for name, latent, prompt_overrides, fact_overrides in (
|
|
(
|
|
"ClownsharKSampler_Beta",
|
|
"latent_image",
|
|
("options",),
|
|
("options", "sigmas"),
|
|
),
|
|
("KSampler (Efficient)", "latent_image", ("script",), ("script",)),
|
|
("ttN KSampler_v2", "latent", ("adv_xyPlot",), ("adv_xyPlot",)),
|
|
):
|
|
self.assertEqual(
|
|
table.samplers[name],
|
|
{
|
|
"strategy": "ksampler",
|
|
"conditioning": pair,
|
|
"latent": latent,
|
|
"facts": "ksampler",
|
|
"prompt_overrides": prompt_overrides,
|
|
"fact_overrides": fact_overrides,
|
|
},
|
|
name,
|
|
)
|
|
for (name, slot), source in {
|
|
("SeedVarianceEnhancer", 0): "conditioning",
|
|
("ApplyInstantIDAdvanced", 1): "positive",
|
|
("ApplyInstantIDAdvanced", 2): "negative",
|
|
}.items():
|
|
self.assertEqual(
|
|
table.conditioning[(name, slot)],
|
|
{"strategy": "passthrough", "input": source},
|
|
)
|
|
self.assertNotIn(("ApplyInstantIDAdvanced", 0), table.conditioning)
|
|
self.assertEqual(
|
|
table.selectors["Switch conditioning [Crystools]"],
|
|
table.selectors["easy ifElse"],
|
|
)
|
|
self.assertEqual(
|
|
table.encoders[("smZ CLIPTextEncode", 0)],
|
|
{
|
|
"strategy": "flag-parts",
|
|
"inputs": ("text", "text_g", "text_l"),
|
|
"params": {"flag": "with_SDXL"},
|
|
},
|
|
)
|
|
for name, source in (
|
|
("ttN text", "text"),
|
|
("CR Text", "text"),
|
|
("CR Prompt Text", "prompt"),
|
|
("TextBoxMira", "text"),
|
|
):
|
|
self.assertEqual(
|
|
table.strings[(name, 0)], {"strategy": "literal", "inputs": (source,)}
|
|
)
|
|
self.assertNotIn((name, 1), table.strings)
|
|
|
|
def test_flag_parts_needs_a_flag_and_two_inputs(self):
|
|
doc = _minimal()
|
|
entry = {
|
|
"class": "smZ CLIPTextEncode",
|
|
"slot": 0,
|
|
"strategy": "flag-parts",
|
|
"inputs": ["text", "text_g"],
|
|
"params": {"flag": "with_SDXL"},
|
|
}
|
|
doc["encoders"].append(entry)
|
|
self.assertEqual(
|
|
self.load_doc(doc).encoders[("smZ CLIPTextEncode", 0)]["params"],
|
|
{"flag": "with_SDXL"},
|
|
)
|
|
for change in (
|
|
{"params": {}},
|
|
{"params": {"flag": "x", "other": "y"}},
|
|
{"params": {"flag": ""}},
|
|
{"params": None},
|
|
{"inputs": ["text"]},
|
|
{"compose": True},
|
|
):
|
|
with self.subTest(change=change):
|
|
broken = _minimal()
|
|
broken["encoders"].append({**entry, **change})
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_doc(broken)
|
|
broken = _minimal()
|
|
broken["encoders"].append(
|
|
{key: item for key, item in entry.items() if key != "params"}
|
|
)
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_doc(broken)
|
|
|
|
def test_prompt_overrides_are_input_names(self):
|
|
doc = _minimal()
|
|
doc["samplers"]["KSampler"]["prompt_overrides"] = ["wildcard"]
|
|
sampler = self.load_doc(doc).samplers["KSampler"]
|
|
self.assertEqual(sampler["prompt_overrides"], ("wildcard",))
|
|
for value in ([], "wildcard", [""], ["a", "a"], [1]):
|
|
with self.subTest(value=value):
|
|
doc["samplers"]["KSampler"]["prompt_overrides"] = value
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_doc(doc)
|
|
doc["samplers"]["KSampler"].pop("prompt_overrides")
|
|
doc["samplers"]["KSampler"]["fact_overrides"] = ["script", "sigmas"]
|
|
sampler = self.load_doc(doc).samplers["KSampler"]
|
|
self.assertEqual(sampler["fact_overrides"], ("script", "sigmas"))
|
|
doc["samplers"]["KSampler"]["fact_overrides"] = ["script", "script"]
|
|
with self.assertRaises(SemanticsError):
|
|
self.load_doc(doc)
|
|
doc["samplers"]["KSampler"].pop("fact_overrides")
|
|
doc["samplers"]["KSampler"]["overrides"] = ["wildcard"]
|
|
self.assertRejected(doc, "unknown field")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|