nodes/ethanfel-ComfyUI-LoRA-Optimizer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
75aafa93193e7b72bda973ace2c0e8a3ba130fd7
ethanfel-ComfyUI-LoRA-Optim…/tests/test_inline_optimizer.py
T
EthanfelandClaude Fable 5 75aafa9319 feat(inline): LoRAOptimizerInline node — chain capture, strip, merge, re-apply 
INPUT_TYPES (widgets in required, clip/settings as optional sockets,
clip_strength_multiplier required for a predictable JS control-widget
count), execute_inline flow (reconstruct chain groups -> build salted
virtual stack -> strip captured entries from clones -> optimize_merge
with normalize_keys pinned to "disabled" -> prepend fingerprint report),
per-group fingerprint report with slot-count and model/clip group-count
mismatch warnings plus passthrough-entry count, and advanced-settings
delegation (AutoTuner settings fall back to defaults with a report line).

Virtual item names are salted with the source patcher patches_uuid
digests (model + clip when connected) so two different LoRA files loaded
at identical strengths cannot false-hit the in-node merge cache; the
salt rides an optional name_salt parameter on _chain_groups_to_stack so
existing callers/tests are unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-06 20:19:53 +02:00

607 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tests for the inline chain-filter optimizer node."""
import struct
import types
import unittest
import uuid
import torch
# Reuse the stub installer / module instance from the main test module.
from tests.test_lora_optimizer import lora_optimizer
def _adapter(rank=4, out_dim=8, in_dim=8):
"""Minimal LoRAAdapter-like payload the engine can expand."""
return lora_optimizer.LoRAAdapter(
loaded_keys=set(),
weights=(torch.randn(out_dim, rank), torch.randn(rank, in_dim),
float(rank), None, None, None),
)
def _entry(strength, payload, strength_model=1.0, offset=None, function=None):
"""A ModelPatcher patch-list entry, shaped like model_patcher.py:807."""
return (strength, payload, strength_model, offset, function)
class TestPatchClassification(unittest.TestCase):
def test_adapter_is_capturable(self):
e = _entry(0.8, _adapter())
self.assertTrue(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_diff_tuple_is_capturable(self):
e = _entry(1.0, ("diff", (torch.randn(4, 4),)))
self.assertTrue(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_set_tuple_passes_through(self):
e = _entry(1.0, ("set", (torch.randn(4, 4),)))
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_nonunit_strength_model_passes_through(self):
e = _entry(1.0, _adapter(), strength_model=0.5)
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_function_entry_passes_through(self):
e = _entry(1.0, _adapter(), function=lambda w: w)
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_unknown_object_passes_through(self):
e = _entry(1.0, object())
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_non_5_tuple_entry_passes_through(self):
# older/nonstandard third-party nodes may store short entries — the
# classifier must not crash on unpack, just pass them through
e = (1.0, _adapter(), 1.0)
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_padded_diff_passes_through(self):
# comfy pads the BASE weight at apply time for this shape — we cannot
# faithfully expand it to a bare diff tensor
e = _entry(1.0, ("diff", (torch.randn(4, 4), {"pad_weight": True})))
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def test_malformed_diff_passes_through(self):
e = _entry(1.0, ("diff", None))
self.assertFalse(lora_optimizer._LoRAMergeBase._is_capturable_entry(e))
def _chain_patches(*loras):
"""Simulate a loader chain: each lora is {key: payload} applied at a
strength. Distinct float objects per call, entries appended in order —
exactly what ModelPatcher.add_patches does."""
patches = {}
seen_ids = set()
for strength_value, lora in loras:
# struct round-trip mints a FRESH float object even for values whose
# literals CPython folds/interns (e.g. 1.0 used twice in one test).
# float(x) would return the SAME object for an exact float input.
s = struct.unpack("d", struct.pack("d", strength_value))[0]
assert id(s) not in seen_ids, "float object reused across simulated calls"
seen_ids.add(id(s))
for key, payload in lora.items():
patches.setdefault(key, []).append(_entry(s, payload))
return patches
class TestChainGroupReconstruction(unittest.TestCase):
def _reconstruct(self, patches):
return lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(patches)
def test_single_lora(self):
patches = _chain_patches((0.8, {"a": _adapter(), "b": _adapter()}))
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 1)
self.assertAlmostEqual(groups[0]["strength"], 0.8)
self.assertEqual(set(groups[0]["entries"]), {"a", "b"})
def test_two_loras_chain_order(self):
patches = _chain_patches(
(0.8, {"a": _adapter(), "b": _adapter()}),
(0.5, {"a": _adapter(), "c": _adapter()}),
)
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 2)
self.assertAlmostEqual(groups[0]["strength"], 0.8)
self.assertAlmostEqual(groups[1]["strength"], 0.5)
self.assertEqual(set(groups[1]["entries"]), {"a", "c"})
def test_subset_alignment(self):
# A patches attn only; B patches attn+mlp. Naive index-as-identity
# would misattribute B's mlp entry (position 0 there) to A.
patches = _chain_patches(
(0.8, {"attn": _adapter()}),
(0.5, {"attn": _adapter(), "mlp": _adapter()}),
)
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 2)
self.assertEqual(set(groups[0]["entries"]), {"attn"})
self.assertEqual(set(groups[1]["entries"]), {"attn", "mlp"})
def test_same_strength_value_distinct_objects(self):
patches = _chain_patches(
(1.0, {"a": _adapter()}),
(1.0, {"a": _adapter(), "b": _adapter()}),
)
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 2)
def test_shared_float_object_fallback(self):
# Pathological: two calls share ONE float object (interning). The
# id-group then holds two entries on key "a" — must split by per-key
# order instead of returning a corrupt group.
s = 1.0
patches = {}
for lora in ({"a": _adapter()}, {"a": _adapter(), "b": _adapter()}):
for key, payload in lora.items():
patches.setdefault(key, []).append(_entry(s, payload))
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 2)
for g in groups:
keys = [k for k in g["entries"]]
self.assertEqual(len(keys), len(set(keys)))
def test_offset_entries_get_tuple_keys(self):
off = (0, 0, 4)
patches = {"qkv": [_entry(0.7, _adapter(), offset=off)]}
groups = self._reconstruct(patches)
self.assertEqual(list(groups[0]["entries"]), [("qkv", off)])
def test_noncapturable_entries_ignored(self):
patches = _chain_patches((0.8, {"a": _adapter()}))
patches["a"].append(_entry(1.0, ("set", (torch.zeros(2, 2),))))
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 1)
self.assertEqual(set(groups[0]["entries"]), {"a"})
def test_empty_patches(self):
self.assertEqual(self._reconstruct({}), [])
def test_non_5_tuple_entry_ignored_without_raising(self):
patches = _chain_patches((0.8, {"a": _adapter()}))
patches["a"].append((1.0, _adapter(), 1.0)) # nonstandard 3-tuple
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 1)
self.assertEqual(set(groups[0]["entries"]), {"a"})
def test_interleaved_collision_no_fragmentation(self):
# Distinct-strength loader X on {a} first, then two interned-strength
# loaders B and C both patching {a, b}. The collision sub-gid must be
# the per-target-key collision ORDINAL — splitting on the absolute
# per-key position fragments this into 4 groups [a],[a,b],[a],[b]
# because X shifts the positions on "a" but not on "b".
d = struct.unpack("d", struct.pack("d", 0.6))[0]
s = 1.0
pX = _adapter()
pB_a, pB_b = _adapter(), _adapter()
pC_a, pC_b = _adapter(), _adapter()
patches = {
"a": [_entry(d, pX), _entry(s, pB_a), _entry(s, pC_a)],
"b": [_entry(s, pB_b), _entry(s, pC_b)],
}
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 3)
self.assertEqual(set(groups[0]["entries"]), {"a"})
self.assertEqual(set(groups[1]["entries"]), {"a", "b"})
self.assertEqual(set(groups[2]["entries"]), {"a", "b"})
# ordinal-k entries belong to the k-th colliding call on EVERY key
self.assertIs(groups[1]["entries"]["a"], pB_a)
self.assertIs(groups[1]["entries"]["b"], pB_b)
self.assertIs(groups[2]["entries"]["a"], pC_a)
self.assertIs(groups[2]["entries"]["b"], pC_b)
def test_collision_subgroup_reordered_by_precedence(self):
# Chain: two interned-strength calls A, B on "a", then distinct call D
# on {z, a}. Dict iteration starts at "z", so D's group is CREATED
# first; both collision groups (base + ordinal sub-group) must be
# sorted forward past D via the shared-key "a" positions — this
# exercises the insertion sort on a collision-created sub-group.
s = 1.0
d = struct.unpack("d", struct.pack("d", 0.5))[0]
pA, pB, pD_a, pD_z = _adapter(), _adapter(), _adapter(), _adapter()
patches = {
"z": [_entry(d, pD_z)],
"a": [_entry(s, pA), _entry(s, pB), _entry(d, pD_a)],
}
groups = self._reconstruct(patches)
self.assertEqual(len(groups), 3)
self.assertIs(groups[0]["entries"]["a"], pA)
self.assertIs(groups[1]["entries"]["a"], pB)
self.assertEqual(set(groups[2]["entries"]), {"z", "a"})
class _FakePatcher:
"""Minimal ModelPatcher stand-in: ordered patch lists + clone()."""
def __init__(self, patches=None):
self.patches = patches if patches is not None else {}
self.patches_uuid = object()
def clone(self):
return _FakePatcher({k: v[:] for k, v in self.patches.items()})
class TestStripCaptured(unittest.TestCase):
def test_strips_only_captured_entries(self):
keep = _entry(1.0, ("set", (torch.zeros(2, 2),)))
patches = _chain_patches((0.8, {"a": _adapter(), "b": _adapter()}))
patches["a"].append(keep)
patcher = _FakePatcher(patches)
groups = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(patcher.patches)
clone = patcher.clone()
lora_optimizer._LoRAMergeBase._strip_captured_entries(clone, groups)
self.assertEqual(clone.patches, {"a": [keep]})
self.assertEqual(len(patcher.patches["a"]), 2) # original untouched
def test_uuid_regenerated(self):
patcher = _FakePatcher(_chain_patches((0.8, {"a": _adapter()})))
groups = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(patcher.patches)
clone = patcher.clone()
before = clone.patches_uuid
lora_optimizer._LoRAMergeBase._strip_captured_entries(clone, groups)
self.assertNotEqual(clone.patches_uuid, before)
def _slot(enabled=True, strength=1.0, model_strength=1.0, clip_strength=1.0,
conflict_mode="all", key_filter="all", preserve=False):
return dict(enabled=enabled, strength=strength, model_strength=model_strength,
clip_strength=clip_strength, conflict_mode=conflict_mode,
key_filter=key_filter, preserve=preserve)
class TestChainStackBuild(unittest.TestCase):
def _build(self, model_groups, clip_groups, slots, visibility="simple"):
return lora_optimizer.LoRAOptimizerInline._chain_groups_to_stack(
model_groups, clip_groups, slots, visibility)
def test_basic_item_schema(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
stack = self._build(mg, [], [_slot()])
item = stack[0]
self.assertTrue(item["_precomputed_diffs"])
self.assertAlmostEqual(item["strength"], 0.8) # loader strength kept
self.assertIsNone(item["clip_strength"])
self.assertEqual(item["conflict_mode"], "all")
self.assertIn("a", item["lora"])
def test_simple_mode_multiplier(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
stack = self._build(mg, [], [_slot(strength=0.5)])
self.assertAlmostEqual(stack[0]["strength"], 0.4) # 0.8 loader × 0.5 slot
def test_advanced_mode_split_multipliers(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build(mg, cg, [_slot(model_strength=0.5, clip_strength=2.0)],
visibility="advanced")
self.assertAlmostEqual(stack[0]["strength"], 0.4) # 0.8 × 0.5
self.assertAlmostEqual(stack[0]["clip_strength"], 1.2) # 0.6 × 2.0
self.assertIn("te.a", stack[0]["lora"]) # clip keys merged in
def test_disabled_slot_excluded(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
stack = self._build(mg, [], [_slot(enabled=False), _slot()])
self.assertEqual(len(stack), 1)
self.assertAlmostEqual(stack[0]["strength"], 0.5)
def test_missing_slots_get_defaults(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
stack = self._build(mg, [], [_slot(preserve=True)]) # only 1 slot for 2 loras
self.assertEqual(len(stack), 2)
self.assertTrue(stack[0]["preserve"])
self.assertFalse(stack[1]["preserve"])
def test_leftover_clip_groups_become_clip_only_items(self):
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build([], cg, [])
self.assertEqual(len(stack), 1)
self.assertIn("te.a", stack[0]["lora"])
def test_leftover_strength_is_clip_product_nonzero(self):
# the engine drops strength == 0 items (active_loras filter in
# optimize_merge) — clip-only items must ride on the clip product
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build([], cg, [])
self.assertNotEqual(stack[0]["strength"], 0.0)
self.assertAlmostEqual(stack[0]["strength"], 0.6)
self.assertAlmostEqual(stack[0]["clip_strength"], 0.6)
def test_zeroed_model_branch_keeps_live_clip_branch(self):
# stock-UI case: LoraLoader with strength_model=0, strength_clip=1.0
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build(mg, cg, [_slot(model_strength=0.0, clip_strength=1.0)],
visibility="advanced")
self.assertEqual(len(stack), 1)
self.assertNotIn("a", stack[0]["lora"]) # unet keys dropped
self.assertIn("te.a", stack[0]["lora"])
self.assertAlmostEqual(stack[0]["strength"], 0.6) # clip product
self.assertAlmostEqual(stack[0]["clip_strength"], 0.6)
def test_both_branches_zeroed_item_excluded(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build(mg, cg, [_slot(model_strength=0.0, clip_strength=0.0)],
visibility="advanced")
self.assertEqual(stack, [])
def test_leftover_after_model_groups_uses_own_slot_index(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()}), (0.4, {"te.b": _adapter()})))
stack = self._build(mg, cg, [_slot(), _slot()])
self.assertEqual(len(stack), 2)
self.assertEqual(stack[1]["name"], "chain lora #2 (clip-only)")
self.assertIn("te.b", stack[1]["lora"])
self.assertAlmostEqual(stack[1]["strength"], 0.4)
# slot index j (not 0) gates the leftover — disabling slot 2 drops it
stack = self._build(mg, cg, [_slot(), _slot(enabled=False)])
self.assertEqual(len(stack), 1)
self.assertNotIn("(clip-only)", stack[0]["name"])
def test_simple_visibility_scales_paired_clip(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
cg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.6, {"te.a": _adapter()})))
stack = self._build(mg, cg, [_slot(strength=0.5)]) # simple visibility
self.assertAlmostEqual(stack[0]["strength"], 0.4) # 0.8 × 0.5
self.assertAlmostEqual(stack[0]["clip_strength"], 0.3) # 0.6 × 0.5
def test_extra_slots_ignored(self):
mg = lora_optimizer._LoRAMergeBase._reconstruct_chain_groups(
_chain_patches((0.8, {"a": _adapter()})))
stack = self._build(mg, [], [_slot(), _slot(strength=9.0), _slot(enabled=False)])
self.assertEqual(len(stack), 1)
self.assertAlmostEqual(stack[0]["strength"], 0.8)
class _FakeCLIP:
def __init__(self, patcher):
self.patcher = patcher
self.cond_stage_model = types.SimpleNamespace()
def clone(self):
return _FakeCLIP(self.patcher.clone())
class TestInlineExecute(unittest.TestCase):
def _node(self):
return lora_optimizer.LoRAOptimizerInline()
def _model(self, patches):
m = _FakePatcher(patches)
m.model = types.SimpleNamespace()
return m
@staticmethod
def _out(result):
return result["result"] if isinstance(result, dict) else result
def _capture_merge(self, node, seen):
"""Mock optimize_merge, recording every kwarg + the stack/model."""
def fake_merge(m, stack, output_strength, **kw):
seen["model"] = m
seen["stack"] = stack
seen["output_strength"] = output_strength
seen.update(kw)
return (m, kw.get("clip"), "engine report", None, None)
node.optimize_merge = fake_merge
def test_no_lora_patches_passthrough(self):
model = self._model({})
node = self._node()
result = node.execute_inline(model, output_strength=1.0)
out = self._out(result)
self.assertIs(out[0], model) # unchanged object back
self.assertIn("No LoRA patches", out[2])
self.assertIn("AFTER your Load LoRA nodes", out[2])
self.assertIsNone(out[3])
self.assertIsNone(out[4])
def test_merge_called_with_virtual_stack_and_stripped_model(self):
model = self._model(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"a": _adapter()})))
node = self._node()
seen = {}
self._capture_merge(node, seen)
node.execute_inline(model, output_strength=1.0)
self.assertEqual(len(seen["stack"]), 2)
self.assertTrue(all(i["_precomputed_diffs"] for i in seen["stack"]))
self.assertEqual(seen["model"].patches, {}) # stripped clone
self.assertIsNot(seen["model"], model)
self.assertEqual(seen["normalize_keys"], "disabled")
def test_report_prepends_fingerprints(self):
model = self._model(_chain_patches((0.8, {"a": _adapter(), "b": _adapter()})))
node = self._node()
node.optimize_merge = lambda m, s, o, **kw: (m, None, "engine report", None, None)
result = node.execute_inline(model, output_strength=1.0)
report = self._out(result)[2]
self.assertIn("#1", report)
self.assertIn("2 keys", report)
self.assertIn("0.80", report)
self.assertIn("rank 4", report)
self.assertIn("engine report", report)
def test_disabled_slot_stripped_but_not_merged(self):
model = self._model(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
node = self._node()
seen = {}
self._capture_merge(node, seen)
node.execute_inline(model, output_strength=1.0, enabled_1=False)
self.assertEqual(len(seen["stack"]), 1)
self.assertAlmostEqual(seen["stack"][0]["strength"], 0.5)
self.assertEqual(seen["model"].patches, {}) # disabled LoRA stripped too
def test_fingerprint_warns_on_group_count_mismatch(self):
# 2 model groups vs 1 clip group (both non-empty) -> order-attribution
# warning naming the usual suspects
model = self._model(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
clip = _FakeCLIP(_FakePatcher(_chain_patches((0.6, {"te.a": _adapter()}))))
node = self._node()
seen = {}
self._capture_merge(node, seen)
result = node.execute_inline(model, output_strength=1.0, clip=clip)
report = self._out(result)[2]
self.assertIn("2 model-side vs 1 clip-side", report)
self.assertIn("LoraLoaderModelOnly", report)
# the clip clone passed to the engine is stripped, original untouched
self.assertIsNot(seen["clip"], clip)
self.assertEqual(seen["clip"].patcher.patches, {})
self.assertEqual(len(clip.patcher.patches["te.a"]), 1)
def test_fingerprint_counts_passthrough_entries(self):
patches = _chain_patches((0.8, {"a": _adapter()}))
patches["a"].append(_entry(1.0, ("set", (torch.zeros(2, 2),))))
model = self._model(patches)
node = self._node()
node.optimize_merge = lambda m, s, o, **kw: (m, None, "r", None, None)
result = node.execute_inline(model, output_strength=1.0)
report = self._out(result)[2]
self.assertIn("1 non-LoRA", report)
self.assertIn("passed through untouched", report)
def test_no_passthrough_note_when_all_captured(self):
model = self._model(_chain_patches((0.8, {"a": _adapter()})))
node = self._node()
node.optimize_merge = lambda m, s, o, **kw: (m, None, "r", None, None)
report = self._out(node.execute_inline(model, output_strength=1.0))[2]
self.assertNotIn("non-LoRA", report)
def test_fingerprint_notes_fewer_slots_than_loras(self):
model = self._model(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
node = self._node()
node.optimize_merge = lambda m, s, o, **kw: (m, None, "r", None, None)
report = self._out(node.execute_inline(
model, output_strength=1.0, lora_count=1))[2]
self.assertIn("2 LoRAs detected but only 1 option slot", report)
self.assertIn("default options", report)
def test_fingerprint_notes_more_slots_than_loras(self):
model = self._model(_chain_patches((0.8, {"a": _adapter()})))
node = self._node()
node.optimize_merge = lambda m, s, o, **kw: (m, None, "r", None, None)
report = self._out(node.execute_inline(
model, output_strength=1.0, lora_count=3))[2]
self.assertIn("3 option slots but only 1 LoRA", report)
self.assertIn("extra slots ignored", report)
def test_autotuner_settings_fall_back_to_defaults(self):
model = self._model(_chain_patches((0.8, {"a": _adapter()})))
node = self._node()
seen = {}
self._capture_merge(node, seen)
result = node.execute_inline(model, output_strength=1.0,
settings={"mode": "autotuner", "top_n": 3})
report = self._out(result)[2]
self.assertIn("AutoTuner settings are not supported inline", report)
self.assertNotIn("top_n", seen) # defaults used
self.assertEqual(seen["normalize_keys"], "disabled")
def test_advanced_settings_delegated_normalize_keys_pinned(self):
settings = {
"mode": "advanced",
"auto_strength": "disabled", "auto_strength_floor": 0.5,
"optimization_mode": "global",
"sparsification": "dare", "sparsification_density": 0.5,
"dare_dampening": 0.1,
"merge_refinement": "refine", "strategy_set": "basic",
"normalize_keys": "enabled", # must NOT reach the engine
"architecture_preset": "dit", "decision_smoothing": 0.1,
"smooth_slerp_gate": True,
"star_eta": 0.9, "tame_layers": 0.1, "tame_threshold": 0.2,
"cache_patches": "disabled", "patch_compression": "disabled",
"svd_device": "cpu", "free_vram_between_passes": "enabled",
"vram_budget": 0.25,
}
model = self._model(_chain_patches((0.8, {"a": _adapter()})))
node = self._node()
seen = {}
self._capture_merge(node, seen)
result = node.execute_inline(model, output_strength=1.0, settings=settings)
report = self._out(result)[2]
self.assertNotIn("not supported inline", report)
self.assertEqual(seen["normalize_keys"], "disabled") # pinned
self.assertEqual(seen["optimization_mode"], "global")
self.assertEqual(seen["sparsification"], "dare")
self.assertAlmostEqual(seen["sparsification_density"], 0.5)
self.assertEqual(seen["cache_patches"], "disabled")
self.assertEqual(seen["svd_device"], "cpu")
self.assertAlmostEqual(seen["star_eta"], 0.9)
def test_salt_names_differ_across_source_patchers(self):
# Two DIFFERENT loader chains at identical strengths must yield
# different virtual item names, or optimize_merge's in-node cache
# would false-hit when the user swaps a LoRA file.
names_per_run = []
for _ in range(2):
model = self._model(_chain_patches(
(0.8, {"a": _adapter()}), (0.5, {"b": _adapter()})))
model.patches_uuid = uuid.uuid4()
node = self._node()
seen = {}
self._capture_merge(node, seen)
node.execute_inline(model, output_strength=1.0)
names_per_run.append([i["name"] for i in seen["stack"]])
self.assertNotEqual(names_per_run[0], names_per_run[1])
for names in names_per_run:
self.assertTrue(names[0].startswith("chain lora #1"))
self.assertTrue(names[1].startswith("chain lora #2"))
def test_salt_names_stable_for_same_patcher(self):
model = self._model(_chain_patches((0.8, {"a": _adapter()})))
model.patches_uuid = uuid.uuid4()
node = self._node()
names_per_run = []
def fake_merge(m, stack, o, **kw):
names_per_run.append([i["name"] for i in stack])
return (m, None, "r", None, None)
node.optimize_merge = fake_merge
node.execute_inline(model, output_strength=1.0)
node.execute_inline(model, output_strength=1.0)
self.assertEqual(names_per_run[0], names_per_run[1])
def test_input_types_surface(self):
cls = lora_optimizer.LoRAOptimizerInline
it = cls.INPUT_TYPES()
req, opt = it["required"], it["optional"]
# widgets the user types/toggles live in "required" (optional widget
# inputs render as sockets in some frontends); JS control widgets
# must all be present and required
for w in ("model", "settings_visibility", "lora_count",
"output_strength", "clip_strength_multiplier"):
self.assertIn(w, req)
for i in (1, cls.MAX_LORAS):
for base in ("enabled", "strength", "model_strength",
"clip_strength", "conflict_mode", "key_filter",
"preserve"):
self.assertIn(f"{base}_{i}", req)
# genuine node-to-node wires stay optional
self.assertIn("clip", opt)
self.assertIn("settings", opt)
self.assertEqual(cls.FUNCTION, "execute_inline")
# inherited from LoRAOptimizer — lora_data keeps SaveMergedLoRA working
self.assertEqual(cls.RETURN_TYPES,
("MODEL", "CLIP", "STRING", "TUNER_DATA", "LORA_DATA"))
self.assertNotIn("RETURN_TYPES", vars(cls)) # not redeclared
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink