271 lines
13 KiB
Python
271 lines
13 KiB
Python
# import pytest
|
|
from src.basic_data_handling.control_flow_nodes import (
|
|
IfElse,
|
|
SwitchCase,
|
|
IfElifElse,
|
|
ContinueFlow,
|
|
ContinueIfNotEmpty,
|
|
FlowSelect,
|
|
ForceCalculation,
|
|
ExecutionOrder,
|
|
IsConnected,
|
|
)
|
|
|
|
|
|
def test_if_else():
|
|
node = IfElse()
|
|
|
|
# Test where condition is True
|
|
assert node.execute(True, "Value if True", "Value if False") == ("Value if True",)
|
|
|
|
# Test where condition is False
|
|
assert node.execute(False, "Value if True", "Value if False") == ("Value if False",)
|
|
|
|
# Test edge cases
|
|
assert node.execute(True, 1, 2) == (1,) # Integers
|
|
assert node.execute(False, [1, 2, 3], None) == (None,) # Complex types
|
|
assert node.execute(True, None, "fallback") == (None,) # Fallback to True branch
|
|
|
|
|
|
def test_if_else_lazy_status():
|
|
node = IfElse()
|
|
|
|
# condition is True, if_true is needed
|
|
assert node.check_lazy_status(True, None, "Value if False") == ["if_true"]
|
|
# condition is True, if_true is not needed
|
|
assert node.check_lazy_status(True, "Value if True", "Value if False") == []
|
|
# condition is False, if_false is needed
|
|
assert node.check_lazy_status(False, "Value if True", None) == ["if_false"]
|
|
# condition is False, if_false is not needed
|
|
assert node.check_lazy_status(False, "Value if True", "Value if False") == []
|
|
|
|
|
|
def test_switch_case():
|
|
node = SwitchCase()
|
|
|
|
# Selector within range, valid cases
|
|
assert node.execute(0, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3") == ("Case 0",)
|
|
assert node.execute(1, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3") == ("Case 1",)
|
|
assert node.execute(3, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3") == ("Case 3",)
|
|
|
|
# Selector out of range, fallback to default case
|
|
assert node.execute(4, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3", default="Default") == ("Default",)
|
|
assert node.execute(99, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3", default="Default") == ("Default",)
|
|
|
|
# Selector out of range, no default provided
|
|
assert node.execute(4, case_0="Case 0", case_1="Case 1", case_2="Case 2", case_3="Case 3") == (None,)
|
|
|
|
# Edge cases
|
|
assert node.execute(0, case_0=None, case_1="Case 1", case_2="Case 2") == (None,) # Case with None
|
|
assert node.execute(2, case_0="Case 0", case_1="Case 1", case_2="Case 2") == ("Case 2",)
|
|
assert node.execute(0, case_0=123, case_1=False, case_2="Case 2") == (123,) # Different data types
|
|
assert node.execute(10, case_0="Case 0", case_1="Case 1", case_2="Case 2", default="Fallback") == ("Fallback",)
|
|
|
|
|
|
def test_switch_case_lazy_status():
|
|
node = SwitchCase()
|
|
|
|
# select is valid, case is needed
|
|
assert node.check_lazy_status(select=1, case_0="val", case_1=None, case_2="val2") == ["case_1"]
|
|
# select is valid, case is not needed
|
|
assert node.check_lazy_status(select=1, case_0="val", case_1="val1", case_2="val2") == []
|
|
# select out of range, default is needed
|
|
assert node.check_lazy_status(select=5, case_0="val", case_1="val1", default=None) == ["default"]
|
|
# select out of range, default is not needed
|
|
assert node.check_lazy_status(select=5, case_0="val", case_1="val1", default="default_val") == []
|
|
# select out of range, no default
|
|
assert node.check_lazy_status(select=5, case_0="val", case_1="val1") == []
|
|
# non-numeric case key should be ignored
|
|
assert node.check_lazy_status(select=0, case_a="val") == []
|
|
|
|
|
|
def test_if_elif_else():
|
|
node = IfElifElse()
|
|
# if is True
|
|
assert node.execute(**{"if": True, "then": "if_val", "elif_0": False, "then_0": "elif0_val", "else": "else_val"}) == ("if_val",)
|
|
# if is False, elif_0 is True
|
|
assert node.execute(**{"if": False, "then": "if_val", "elif_0": True, "then_0": "elif0_val", "else": "else_val"}) == ("elif0_val",)
|
|
# if is False, elif_0 is False, else
|
|
assert node.execute(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif0_val", "else": "else_val"}) == ("else_val",)
|
|
# if is True, and one elif that is False, else
|
|
assert node.execute(**{"if": True, "then": "if_val", "elif_0": False, "then_0": "elif0_val", "else": "else_val"}) == ("if_val",)
|
|
# if is True, and one elif that is True, else
|
|
assert node.execute(**{"if": True, "then": "if_val", "elif_0": True, "then_0": "elif0_val", "else": "else_val"}) == ("if_val",)
|
|
# if is False, and one elif that is True, else
|
|
assert node.execute(**{"if": False, "then": "if_val", "elif_0": True, "then_0": "elif0_val", "else": "else_val"}) == ("elif0_val",)
|
|
# if is False, and one elif that is False, else
|
|
assert node.execute(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif0_val", "else": "else_val"}) == ("else_val",)
|
|
# if is False, multiple elifs, one is true
|
|
assert node.execute(
|
|
**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif0_val", "elif_1": True, "then_1": "elif1_val", "else": "else_val"}
|
|
) == ("elif1_val",)
|
|
# all false, else
|
|
assert node.execute(
|
|
**{
|
|
"if": False,
|
|
"then": "if_val",
|
|
"elif_0": False,
|
|
"then_0": "elif0_val",
|
|
"elif_1": False,
|
|
"then_1": "elif1_val",
|
|
"else": "else_val",
|
|
}
|
|
) == ("else_val",)
|
|
# all false, no else
|
|
assert node.execute(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif0_val"}) == (None,)
|
|
# Missing then for a true condition
|
|
assert node.execute(**{"if": True, "elif_0": False, "then_0": "elif0_val"}) == (None,)
|
|
|
|
|
|
def test_if_elif_else_lazy_status():
|
|
node = IfElifElse()
|
|
# if is True, then is needed
|
|
assert node.check_lazy_status(**{"if": True, "then": None, "else": "else_val"}) == ["then"]
|
|
# if is True, then is not needed
|
|
assert node.check_lazy_status(**{"if": True, "then": "if_val", "else": "else_val"}) == []
|
|
# if is False, else is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "else": None}) == ["else"]
|
|
# if is False, else is not needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "else": "else_val"}) == []
|
|
# if is False, elif_0 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": None}) == ["elif_0"]
|
|
# if is False, elif_0 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": None, "else": None}) == ["elif_0"]
|
|
# if is False, elif_0 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": None, "else": "else_val"}) == ["elif_0"]
|
|
# if is False, elif_0 False, else needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "else": None}) == ["else"]
|
|
# if is False, elif_0 False, else needs evaluation but is already fulfilled
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "else": "else_val"}) == []
|
|
# if is False, elif_0 is True, then_0 is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": True, "then_0": None, "else": None}) == ["then_0"]
|
|
# if is False, elif_0 is True, then_0 is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": True, "then_0": None, "else": "else_val"}) == ["then_0"]
|
|
# if is False, elif_0 is True, then_0 is not needed, but else is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": True, "then_0": "elif0_val", "else": None}) == ["else"]
|
|
# if is False, elif_0 is True, then_0 is not needed, but else is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": True, "then_0": "elif0_val", "else": "else_val"}) == []
|
|
# if is False, elif_0 False, elif_1 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": None}) == ["elif_1"]
|
|
# if is False, elif_0 False, elif_1 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": None, "else": None}) == ["elif_1"]
|
|
# if is False, elif_0 False, elif_1 needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": None, "else": "else_val"}) == ["elif_1"]
|
|
# if is False, elif_0 False, elif_1 False, else needs evaluation
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": False, "else": None}) == ["else"]
|
|
# if is False, elif_0 False, elif_1 False, else needs evaluation but is already fulfilled
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": False, "else": "else_val"}) == []
|
|
# if is False, elif_0 False, elif_1 is True, then_1 is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "elif_1": True, "then_1": None, "else": None}) == [
|
|
"then_1"
|
|
]
|
|
# if is False, elif_0 False, elif_1 is True, then_1 is needed
|
|
assert node.check_lazy_status(
|
|
**{"if": False, "then": "if_val", "elif_0": False, "elif_1": True, "then_1": None, "else": "else_val"}
|
|
) == ["then_1"]
|
|
# if is False, elif_0 False, elif_1 is True, then_1 is not needed, but else is needed
|
|
assert node.check_lazy_status(
|
|
**{"if": False, "then": "if_val", "elif_0": False, "elif_1": True, "then_1": "elif1_val", "else": None}
|
|
) == ["else"]
|
|
# if is False, elif_0 False, elif_1 is True, then_1 is not needed, but else is needed
|
|
assert (
|
|
node.check_lazy_status(
|
|
**{"if": False, "then": "if_val", "elif_0": False, "elif_1": True, "then_1": "elif1_val", "else": "else_val"}
|
|
)
|
|
== []
|
|
)
|
|
# all false, else is needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif1_val", "else": None}) == ["else"]
|
|
# all false, else is not needed
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif1_val", "else": "else_val"}) == []
|
|
# all false, no else
|
|
assert node.check_lazy_status(**{"if": False, "then": "if_val", "elif_0": False, "then_0": "elif0_val"}) == []
|
|
|
|
|
|
def test_continue_flow():
|
|
node = ContinueFlow()
|
|
# select is True
|
|
assert node.execute("some value", select=True) == ("some value",)
|
|
# select is False
|
|
res = node.execute("some value", select=False)
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
# select is default (True)
|
|
assert node.execute("some value") == ("some value",)
|
|
# Test with different value types
|
|
assert node.execute(123, select=True) == (123,)
|
|
res = node.execute([1, 2], select=False)
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
|
|
|
|
def test_continue_if_not_empty():
|
|
node = ContinueIfNotEmpty()
|
|
# Non-empty values pass through (single values wrapped into a list for OUTPUT_IS_LIST).
|
|
assert node.execute("some value") == (["some value"],)
|
|
assert node.execute([1, 2]) == ([1, 2],)
|
|
assert node.execute({"key": "value"}) == ([{"key": "value"}],)
|
|
assert node.execute({1, 2}) == ([{1, 2}],)
|
|
# Empty values block execution (ExecutionBlocker under ComfyUI). Silent by default.
|
|
res = node.execute(None)
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
for empty in ([], {}, set(), "", 0, False):
|
|
res = node.execute(empty)
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
# A provided message is passed to the blocker (as a list under INPUT_IS_LIST).
|
|
res = node.execute(None, message=["Interrupted"])
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
res = node.execute(None, message="Interrupted")
|
|
assert res == (None,) or "ExecutionBlocker" in str(res[0])
|
|
|
|
|
|
def test_flow_select():
|
|
node = FlowSelect()
|
|
# select is True
|
|
res = node.select("some value", select=True)
|
|
assert res[0] == "some value"
|
|
assert res[1] is None or "ExecutionBlocker" in str(res[1])
|
|
# select is False
|
|
res = node.select("some value", select=False)
|
|
assert res[0] is None or "ExecutionBlocker" in str(res[0])
|
|
assert res[1] == "some value"
|
|
# select is default (True)
|
|
res = node.select("some value")
|
|
assert res[0] == "some value"
|
|
assert res[1] is None or "ExecutionBlocker" in str(res[1])
|
|
# Test with different value types
|
|
res = node.select(42, select=True)
|
|
assert res[0] == 42
|
|
assert res[1] is None or "ExecutionBlocker" in str(res[1])
|
|
res = node.select({"a": 1}, select=False)
|
|
assert res[0] is None or "ExecutionBlocker" in str(res[0])
|
|
assert res[1] == {"a": 1}
|
|
|
|
|
|
def test_force_calculation():
|
|
node = ForceCalculation()
|
|
# execute
|
|
assert node.execute("some value") == ("some value",)
|
|
assert node.execute(None) == (None,)
|
|
|
|
|
|
def test_execution_order():
|
|
node = ExecutionOrder()
|
|
# with "any node output"
|
|
assert node.execute(**{"any node output": "passthrough_val"}) == (None, "passthrough_val")
|
|
# without "any node output"
|
|
assert node.execute() == (None, [])
|
|
# with other kwargs
|
|
assert node.execute(other_kwarg="some_val") == (None, [])
|
|
# with multiple kwargs
|
|
assert node.execute(**{"any node output": "passthrough_val", "E/O": "ignored"}) == (None, "passthrough_val")
|
|
|
|
|
|
def test_is_connected():
|
|
node = IsConnected()
|
|
# Test when input is connected (not None)
|
|
assert node.execute("connected") == (True,)
|
|
assert node.execute(0) == (True,)
|
|
assert node.execute(False) == (True,)
|
|
# Test when input is NOT connected (None)
|
|
assert node.execute(None) == (False,)
|
|
assert node.execute() == (False,)
|