opensource
/
dify
mirror of https://github.com/langgenius/dify.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(api): Initial support for `boolean` / `array[boolean]` types

This commit is contained in:
QuantumGhost 2025-07-24 01:39:11 +08:00
parent 6d03a15e0f
commit 80e08562be
20 changed files with 964 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
api/child_class.py Normal file
View File

@ -0,0 +1,11 @@
from tests.integration_tests.utils.parent_class import ParentClass
class ChildClass(ParentClass):
"""Test child class for module import helper tests"""
def __init__(self, name):
super().__init__(name)
def get_name(self):
return f"Child: {self.name}"

12
api/core/variables/segments.py
View File

@ -144,6 +144,11 @@ class FileSegment(Segment):
return ""
class BooleanSegment(Segment):
value_type: SegmentType = SegmentType.BOOLEAN
value: bool
class ArrayAnySegment(ArraySegment):
value_type: SegmentType = SegmentType.ARRAY_ANY
value: Sequence[Any]
@ -188,6 +193,11 @@ class ArrayFileSegment(ArraySegment):
return ""
class ArrayBooleanSegment(ArraySegment):
value_type: SegmentType = SegmentType.ARRAY_BOOLEAN
value: Sequence[bool]
def get_segment_discriminator(v: Any) -> SegmentType | None:
if isinstance(v, Segment):
return v.value_type
@ -221,11 +231,13 @@ SegmentUnion: TypeAlias = Annotated[
| Annotated[IntegerSegment, Tag(SegmentType.INTEGER)]
| Annotated[ObjectSegment, Tag(SegmentType.OBJECT)]
| Annotated[FileSegment, Tag(SegmentType.FILE)]
| Annotated[BooleanSegment, Tag(SegmentType.BOOLEAN)]
| Annotated[ArrayAnySegment, Tag(SegmentType.ARRAY_ANY)]
| Annotated[ArrayStringSegment, Tag(SegmentType.ARRAY_STRING)]
| Annotated[ArrayNumberSegment, Tag(SegmentType.ARRAY_NUMBER)]
| Annotated[ArrayObjectSegment, Tag(SegmentType.ARRAY_OBJECT)]
| Annotated[ArrayFileSegment, Tag(SegmentType.ARRAY_FILE)]
| Annotated[ArrayBooleanSegment, Tag(SegmentType.ARRAY_BOOLEAN)]
),
Discriminator(get_segment_discriminator),
]

36
api/core/variables/types.py
View File

@ -27,12 +27,14 @@ class SegmentType(StrEnum):
SECRET = "secret"
FILE = "file"
BOOLEAN = "boolean"
ARRAY_ANY = "array[any]"
ARRAY_STRING = "array[string]"
ARRAY_NUMBER = "array[number]"
ARRAY_OBJECT = "array[object]"
ARRAY_FILE = "array[file]"
ARRAY_BOOLEAN = "array[boolean]"
NONE = "none"
@ -76,12 +78,18 @@ class SegmentType(StrEnum):
return SegmentType.ARRAY_FILE
case SegmentType.NONE:
return SegmentType.ARRAY_ANY
case SegmentType.BOOLEAN:
return SegmentType.ARRAY_BOOLEAN
case _:
# This should be unreachable.
raise ValueError(f"not supported value {value}")
if value is None:
return SegmentType.NONE
elif isinstance(value, int) and not isinstance(value, bool):
# Important: The check for `bool` must precede the check for `int`,
# as `bool` is a subclass of `int` in Python's type hierarchy.
elif isinstance(value, bool):
return SegmentType.BOOLEAN
elif isinstance(value, int):
return SegmentType.INTEGER
elif isinstance(value, float):
return SegmentType.FLOAT
@ -126,6 +134,10 @@ class SegmentType(StrEnum):
"""
if self.is_array_type():
return self._validate_array(value, array_validation)
# Important: The check for `bool` must precede the check for `int`,
# as `bool` is a subclass of `int` in Python's type hierarchy.
elif self == SegmentType.BOOLEAN:
return isinstance(value, bool)
elif self == SegmentType.NUMBER:
return isinstance(value, (int, float))
elif self == SegmentType.STRING:
@ -141,6 +153,27 @@ class SegmentType(StrEnum):
else:
raise AssertionError("this statement should be unreachable.")
@staticmethod
def cast_value(value: Any, type_: "SegmentType") -> Any:
# Cast Python's `bool` type to `int` when the runtime type requires
# an integer or number.
#
# This ensures compatibility with existing workflows that may use `bool` as
# `int`, since in Python's type system, `bool` is a subtype of `int`.
#
# This function exists solely to maintain compatibility with existing workflows.
# It should not be used to compromise the integrity of the runtime type system.
# No additional casting rules should be introduced to this function.
if type_ in (
SegmentType.INTEGER,
SegmentType.NUMBER,
) and isinstance(value, bool):
return int(value)
if type_ == SegmentType.ARRAY_NUMBER and all(isinstance(i, bool) for i in value):
return [int(i) for i in value]
return value
def exposed_type(self) -> "SegmentType":
"""Returns the type exposed to the frontend.
@ -157,6 +190,7 @@ _ARRAY_ELEMENT_TYPES_MAPPING: Mapping[SegmentType, SegmentType] = {
SegmentType.ARRAY_NUMBER: SegmentType.NUMBER,
SegmentType.ARRAY_OBJECT: SegmentType.OBJECT,
SegmentType.ARRAY_FILE: SegmentType.FILE,
SegmentType.ARRAY_BOOLEAN: SegmentType.BOOLEAN,
}
_ARRAY_TYPES = frozenset(

12
api/core/variables/variables.py
View File

@ -8,11 +8,13 @@ from core.helper import encrypter
from .segments import (
ArrayAnySegment,
ArrayBooleanSegment,
ArrayFileSegment,
ArrayNumberSegment,
ArrayObjectSegment,
ArraySegment,
ArrayStringSegment,
BooleanSegment,
FileSegment,
FloatSegment,
IntegerSegment,
@ -96,10 +98,18 @@ class FileVariable(FileSegment, Variable):
pass
class BooleanVariable(BooleanSegment, Variable):
pass
class ArrayFileVariable(ArrayFileSegment, ArrayVariable):
pass
class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
pass
# The `VariableUnion`` type is used to enable serialization and deserialization with Pydantic.
# Use `Variable` for type hinting when serialization is not required.
#
@ -114,11 +124,13 @@ VariableUnion: TypeAlias = Annotated[
| Annotated[IntegerVariable, Tag(SegmentType.INTEGER)]
| Annotated[ObjectVariable, Tag(SegmentType.OBJECT)]
| Annotated[FileVariable, Tag(SegmentType.FILE)]
| Annotated[BooleanVariable, Tag(SegmentType.BOOLEAN)]
| Annotated[ArrayAnyVariable, Tag(SegmentType.ARRAY_ANY)]
| Annotated[ArrayStringVariable, Tag(SegmentType.ARRAY_STRING)]
| Annotated[ArrayNumberVariable, Tag(SegmentType.ARRAY_NUMBER)]
| Annotated[ArrayObjectVariable, Tag(SegmentType.ARRAY_OBJECT)]
| Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
| Annotated[ArrayBooleanVariable, Tag(SegmentType.ARRAY_BOOLEAN)]
| Annotated[SecretVariable, Tag(SegmentType.SECRET)]
),
Discriminator(get_segment_discriminator),

2
api/core/workflow/nodes/loop/entities.py
View File

@ -12,9 +12,11 @@ _VALID_VAR_TYPE = frozenset(
SegmentType.STRING,
SegmentType.NUMBER,
SegmentType.OBJECT,
SegmentType.BOOLEAN,
SegmentType.ARRAY_STRING,
SegmentType.ARRAY_NUMBER,
SegmentType.ARRAY_OBJECT,
SegmentType.ARRAY_BOOLEAN,
]
)

7
api/core/workflow/nodes/loop/loop_node.py
View File

@ -522,7 +522,12 @@ class LoopNode(BaseNode):
@staticmethod
def _get_segment_for_constant(var_type: SegmentType, value: Any) -> Segment:
"""Get the appropriate segment type for a constant value."""
if var_type in ["array[string]", "array[number]", "array[object]"]:
if var_type in [
SegmentType.ARRAY_NUMBER,
SegmentType.ARRAY_OBJECT,
SegmentType.ARRAY_STRING,
SegmentType.ARRAY_BOOLEAN,
]:
if value and isinstance(value, str):
value = json.loads(value)
else:

7
api/core/workflow/nodes/variable_assigner/v1/node.py
View File

@ -2,6 +2,7 @@ from collections.abc import Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Optional, TypeAlias
from core.variables import SegmentType, Variable
from core.variables.segments import BooleanSegment
from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
from core.workflow.conversation_variable_updater import ConversationVariableUpdater
from core.workflow.entities.node_entities import NodeRunResult
@ -158,8 +159,8 @@ class VariableAssignerNode(BaseNode):
def get_zero_value(t: SegmentType):
# TODO(QuantumGhost): this should be a method of `SegmentType`.
match t:
case SegmentType.ARRAY_OBJECT | SegmentType.ARRAY_STRING | SegmentType.ARRAY_NUMBER:
return variable_factory.build_segment([])
case SegmentType.ARRAY_OBJECT | SegmentType.ARRAY_STRING | SegmentType.ARRAY_NUMBER | SegmentType.ARRAY_BOOLEAN:
return variable_factory.build_segment_with_type(t, [])
case SegmentType.OBJECT:
return variable_factory.build_segment({})
case SegmentType.STRING:
@ -170,5 +171,7 @@ def get_zero_value(t: SegmentType):
return variable_factory.build_segment(0.0)
case SegmentType.NUMBER:
return variable_factory.build_segment(0)
case SegmentType.BOOLEAN:
return BooleanSegment(value=False)
case _:
raise VariableOperatorNodeError(f"unsupported variable type: {t}")

2
api/core/workflow/nodes/variable_assigner/v2/constants.py
View File

@ -4,9 +4,11 @@ from core.variables import SegmentType
EMPTY_VALUE_MAPPING = {
SegmentType.STRING: "",
SegmentType.NUMBER: 0,
SegmentType.BOOLEAN: False,
SegmentType.OBJECT: {},
SegmentType.ARRAY_ANY: [],
SegmentType.ARRAY_STRING: [],
SegmentType.ARRAY_NUMBER: [],
SegmentType.ARRAY_OBJECT: [],
SegmentType.ARRAY_BOOLEAN: [],
}

28
api/core/workflow/nodes/variable_assigner/v2/helpers.py
View File

@ -16,28 +16,15 @@ def is_operation_supported(*, variable_type: SegmentType, operation: Operation):
SegmentType.NUMBER,
SegmentType.INTEGER,
SegmentType.FLOAT,
SegmentType.BOOLEAN,
}
case Operation.ADD | Operation.SUBTRACT | Operation.MULTIPLY | Operation.DIVIDE:
# Only number variable can be added, subtracted, multiplied or divided
return variable_type in {SegmentType.NUMBER, SegmentType.INTEGER, SegmentType.FLOAT}
case Operation.APPEND | Operation.EXTEND:
case Operation.APPEND | Operation.EXTEND | Operation.REMOVE_FIRST | Operation.REMOVE_LAST:
# Only array variable can be appended or extended
return variable_type in {
SegmentType.ARRAY_ANY,
SegmentType.ARRAY_OBJECT,
SegmentType.ARRAY_STRING,
SegmentType.ARRAY_NUMBER,
SegmentType.ARRAY_FILE,
}
case Operation.REMOVE_FIRST | Operation.REMOVE_LAST:
# Only array variable can have elements removed
return variable_type in {
SegmentType.ARRAY_ANY,
SegmentType.ARRAY_OBJECT,
SegmentType.ARRAY_STRING,
SegmentType.ARRAY_NUMBER,
SegmentType.ARRAY_FILE,
}
return variable_type.is_array_type()
case _:
return False
@ -50,7 +37,7 @@ def is_variable_input_supported(*, operation: Operation):
def is_constant_input_supported(*, variable_type: SegmentType, operation: Operation):
match variable_type:
case SegmentType.STRING | SegmentType.OBJECT:
case SegmentType.STRING | SegmentType.OBJECT | SegmentType.BOOLEAN:
return operation in {Operation.OVER_WRITE, Operation.SET}
case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
return operation in {
@ -72,6 +59,9 @@ def is_input_value_valid(*, variable_type: SegmentType, operation: Operation, va
case SegmentType.STRING:
return isinstance(value, str)
case SegmentType.BOOLEAN:
return isinstance(value, bool)
case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
if not isinstance(value, int | float):
return False
@ -91,6 +81,8 @@ def is_input_value_valid(*, variable_type: SegmentType, operation: Operation, va
return isinstance(value, int | float)
case SegmentType.ARRAY_OBJECT if operation == Operation.APPEND:
return isinstance(value, dict)
case SegmentType.ARRAY_BOOLEAN if operation == Operation.APPEND:
return isinstance(value, bool)
# Array & Extend / Overwrite
case SegmentType.ARRAY_ANY if operation in {Operation.EXTEND, Operation.OVER_WRITE}:
@ -101,6 +93,8 @@ def is_input_value_valid(*, variable_type: SegmentType, operation: Operation, va
return isinstance(value, list) and all(isinstance(item, int | float) for item in value)
case SegmentType.ARRAY_OBJECT if operation in {Operation.EXTEND, Operation.OVER_WRITE}:
return isinstance(value, list) and all(isinstance(item, dict) for item in value)
case SegmentType.ARRAY_BOOLEAN if operation in {Operation.EXTEND, Operation.OVER_WRITE}:
return isinstance(value, list) and all(isinstance(item, bool) for item in value)
case _:
return False

2
api/core/workflow/utils/condition/entities.py
View File

@ -45,5 +45,5 @@ class SubVariableCondition(BaseModel):
class Condition(BaseModel):
variable_selector: list[str]
comparison_operator: SupportedComparisonOperator
value: str | Sequence[str] | None = None
value: str | Sequence[str] | bool | None = None
sub_variable_condition: SubVariableCondition | None = None

42
api/core/workflow/utils/condition/processor.py
View File

@ -1,5 +1,5 @@
from collections.abc import Sequence
from typing import Any, Literal
from typing import Any, Literal, Union
from core.file import FileAttribute, file_manager
from core.variables import ArrayFileSegment
@ -77,7 +77,7 @@ def _evaluate_condition(
*,
operator: SupportedComparisonOperator,
value: Any,
expected: str | Sequence[str] | None,
expected: Union[str, Sequence[str], None],
) -> bool:
match operator:
case "contains":
@ -130,7 +130,7 @@ def _assert_contains(*, value: Any, expected: Any) -> bool:
if not value:
return False
if not isinstance(value, str | list):
if not isinstance(value, (str, list)):
raise ValueError("Invalid actual value type: string or array")
if expected not in value:
@ -142,7 +142,7 @@ def _assert_not_contains(*, value: Any, expected: Any) -> bool:
if not value:
return True
if not isinstance(value, str | list):
if not isinstance(value, (str, list)):
raise ValueError("Invalid actual value type: string or array")
if expected in value:
@ -178,8 +178,8 @@ def _assert_is(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, str):
raise ValueError("Invalid actual value type: string")
if not isinstance(value, (str, bool)):
raise ValueError("Invalid actual value type: string or boolean")
if value != expected:
return False
@ -190,8 +190,8 @@ def _assert_is_not(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, str):
raise ValueError("Invalid actual value type: string")
if not isinstance(value, (str, bool)):
raise ValueError("Invalid actual value type: string or boolean")
if value == expected:
return False
@ -214,10 +214,13 @@ def _assert_equal(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
raise ValueError("Invalid actual value type: number")
if not isinstance(value, (int, float, bool)):
raise ValueError("Invalid actual value type: number or boolean")
if isinstance(value, int):
# Handle boolean comparison
if isinstance(value, bool):
expected = bool(expected)
elif isinstance(value, int):
expected = int(expected)
else:
expected = float(expected)
@ -231,10 +234,13 @@ def _assert_not_equal(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
raise ValueError("Invalid actual value type: number")
if not isinstance(value, (int, float, bool)):
raise ValueError("Invalid actual value type: number or boolean")
if isinstance(value, int):
# Handle boolean comparison
if isinstance(value, bool):
expected = bool(expected)
elif isinstance(value, int):
expected = int(expected)
else:
expected = float(expected)
@ -248,7 +254,7 @@ def _assert_greater_than(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
if not isinstance(value, (int, float)):
raise ValueError("Invalid actual value type: number")
if isinstance(value, int):
@ -265,7 +271,7 @@ def _assert_less_than(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
if not isinstance(value, (int, float)):
raise ValueError("Invalid actual value type: number")
if isinstance(value, int):
@ -282,7 +288,7 @@ def _assert_greater_than_or_equal(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
if not isinstance(value, (int, float)):
raise ValueError("Invalid actual value type: number")
if isinstance(value, int):
@ -299,7 +305,7 @@ def _assert_less_than_or_equal(*, value: Any, expected: Any) -> bool:
if value is None:
return False
if not isinstance(value, int | float):
if not isinstance(value, (int, float)):
raise ValueError("Invalid actual value type: number")
if isinstance(value, int):

34
api/factories/variable_factory.py
View File

@ -7,11 +7,13 @@ from core.file import File
from core.variables.exc import VariableError
from core.variables.segments import (
ArrayAnySegment,
ArrayBooleanSegment,
ArrayFileSegment,
ArrayNumberSegment,
ArrayObjectSegment,
ArraySegment,
ArrayStringSegment,
BooleanSegment,
FileSegment,
FloatSegment,
IntegerSegment,
@ -23,10 +25,12 @@ from core.variables.segments import (
from core.variables.types import SegmentType
from core.variables.variables import (
ArrayAnyVariable,
ArrayBooleanVariable,
ArrayFileVariable,
ArrayNumberVariable,
ArrayObjectVariable,
ArrayStringVariable,
BooleanVariable,
FileVariable,
FloatVariable,
IntegerVariable,
@ -49,17 +53,19 @@ class TypeMismatchError(Exception):
# Define the constant
SEGMENT_TO_VARIABLE_MAP = {
StringSegment: StringVariable,
IntegerSegment: IntegerVariable,
FloatSegment: FloatVariable,
ObjectSegment: ObjectVariable,
FileSegment: FileVariable,
ArrayStringSegment: ArrayStringVariable,
ArrayAnySegment: ArrayAnyVariable,
ArrayBooleanSegment: ArrayBooleanVariable,
ArrayFileSegment: ArrayFileVariable,
ArrayNumberSegment: ArrayNumberVariable,
ArrayObjectSegment: ArrayObjectVariable,
ArrayFileSegment: ArrayFileVariable,
ArrayAnySegment: ArrayAnyVariable,
ArrayStringSegment: ArrayStringVariable,
BooleanSegment: BooleanVariable,
FileSegment: FileVariable,
FloatSegment: FloatVariable,
IntegerSegment: IntegerVariable,
NoneSegment: NoneVariable,
ObjectSegment: ObjectVariable,
StringSegment: StringVariable,
}
@ -99,6 +105,8 @@ def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequen
mapping = dict(mapping)
mapping["value_type"] = SegmentType.FLOAT
result = FloatVariable.model_validate(mapping)
case SegmentType.BOOLEAN:
result = BooleanVariable.model_validate(mapping)
case SegmentType.NUMBER if not isinstance(value, float | int):
raise VariableError(f"invalid number value {value}")
case SegmentType.OBJECT if isinstance(value, dict):
@ -109,6 +117,8 @@ def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequen
result = ArrayNumberVariable.model_validate(mapping)
case SegmentType.ARRAY_OBJECT if isinstance(value, list):
result = ArrayObjectVariable.model_validate(mapping)
case SegmentType.ARRAY_BOOLEAN if isinstance(value, list):
result = ArrayBooleanVariable.model_validate(mapping)
case _:
raise VariableError(f"not supported value type {value_type}")
if result.size > dify_config.MAX_VARIABLE_SIZE:
@ -129,6 +139,8 @@ def build_segment(value: Any, /) -> Segment:
return NoneSegment()
if isinstance(value, str):
return StringSegment(value=value)
if isinstance(value, bool):
return BooleanSegment(value=value)
if isinstance(value, int):
return IntegerSegment(value=value)
if isinstance(value, float):
@ -152,6 +164,8 @@ def build_segment(value: Any, /) -> Segment:
return ArrayStringSegment(value=value)
case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
return ArrayNumberSegment(value=value)
case SegmentType.BOOLEAN:
return ArrayBooleanSegment(value=value)
case SegmentType.OBJECT:
return ArrayObjectSegment(value=value)
case SegmentType.FILE:
@ -170,6 +184,7 @@ _segment_factory: Mapping[SegmentType, type[Segment]] = {
SegmentType.INTEGER: IntegerSegment,
SegmentType.FLOAT: FloatSegment,
SegmentType.FILE: FileSegment,
SegmentType.BOOLEAN: BooleanSegment,
SegmentType.OBJECT: ObjectSegment,
# Array types
SegmentType.ARRAY_ANY: ArrayAnySegment,
@ -177,6 +192,7 @@ _segment_factory: Mapping[SegmentType, type[Segment]] = {
SegmentType.ARRAY_NUMBER: ArrayNumberSegment,
SegmentType.ARRAY_OBJECT: ArrayObjectSegment,
SegmentType.ARRAY_FILE: ArrayFileSegment,
SegmentType.ARRAY_BOOLEAN: ArrayBooleanSegment,
}
@ -225,6 +241,8 @@ def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
return ArrayAnySegment(value=value)
elif segment_type == SegmentType.ARRAY_STRING:
return ArrayStringSegment(value=value)
elif segment_type == SegmentType.ARRAY_BOOLEAN:
return ArrayBooleanSegment(value=value)
elif segment_type == SegmentType.ARRAY_NUMBER:
return ArrayNumberSegment(value=value)
elif segment_type == SegmentType.ARRAY_OBJECT:

11
api/lazy_load_class.py Normal file
View File

@ -0,0 +1,11 @@
from tests.integration_tests.utils.parent_class import ParentClass
class LazyLoadChildClass(ParentClass):
"""Test lazy load child class for module import helper tests"""
def __init__(self, name):
super().__init__(name)
def get_name(self):
return self.name

206
api/tests/unit_tests/core/workflow/nodes/test_if_else.py
View File

@ -272,3 +272,209 @@ def test_array_file_contains_file_name():
assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
assert result.outputs is not None
assert result.outputs["result"] is True
def test_execute_if_else_boolean_conditions():
"""Test IfElseNode with boolean conditions using various operators"""
graph_config = {"edges": [], "nodes": [{"data": {"type": "start"}, "id": "start"}]}
graph = Graph.init(graph_config=graph_config)
init_params = GraphInitParams(
tenant_id="1",
app_id="1",
workflow_type=WorkflowType.WORKFLOW,
workflow_id="1",
graph_config=graph_config,
user_id="1",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
)
# construct variable pool with boolean values
pool = VariablePool(
system_variables={SystemVariableKey.FILES: [], SystemVariableKey.USER_ID: "aaa"}, user_inputs={}
)
pool.add(["start", "bool_true"], True)
pool.add(["start", "bool_false"], False)
pool.add(["start", "bool_array"], [True, False, True])
pool.add(["start", "mixed_array"], [True, "false", 1, 0])
node = IfElseNode(
id=str(uuid.uuid4()),
graph_init_params=init_params,
graph=graph,
graph_runtime_state=GraphRuntimeState(variable_pool=pool, start_at=time.perf_counter()),
config={
"id": "if-else",
"data": {
"title": "Boolean Test",
"type": "if-else",
"logical_operator": "and",
"conditions": [
# Test boolean "is" operator
{"comparison_operator": "is", "variable_selector": ["start", "bool_true"], "value": "true"},
# Test boolean "is not" operator
{"comparison_operator": "is not", "variable_selector": ["start", "bool_false"], "value": "true"},
# Test boolean "=" operator
{"comparison_operator": "=", "variable_selector": ["start", "bool_true"], "value": "1"},
# Test boolean "≠" operator
{"comparison_operator": "≠", "variable_selector": ["start", "bool_false"], "value": "1"},
# Test boolean "not null" operator
{"comparison_operator": "not null", "variable_selector": ["start", "bool_true"]},
# Test boolean array "contains" operator
{"comparison_operator": "contains", "variable_selector": ["start", "bool_array"], "value": "true"},
# Test boolean "in" operator
{
"comparison_operator": "in",
"variable_selector": ["start", "bool_true"],
"value": ["true", "false"],
},
],
},
},
)
# Mock db.session.close()
db.session.close = MagicMock()
# execute node
result = node._run()
assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
assert result.outputs is not None
assert result.outputs["result"] is True
def test_execute_if_else_boolean_false_conditions():
"""Test IfElseNode with boolean conditions that should evaluate to false"""
graph_config = {"edges": [], "nodes": [{"data": {"type": "start"}, "id": "start"}]}
graph = Graph.init(graph_config=graph_config)
init_params = GraphInitParams(
tenant_id="1",
app_id="1",
workflow_type=WorkflowType.WORKFLOW,
workflow_id="1",
graph_config=graph_config,
user_id="1",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
)
# construct variable pool with boolean values
pool = VariablePool(
system_variables={SystemVariableKey.FILES: [], SystemVariableKey.USER_ID: "aaa"}, user_inputs={}
)
pool.add(["start", "bool_true"], True)
pool.add(["start", "bool_false"], False)
pool.add(["start", "bool_array"], [True, False, True])
node = IfElseNode(
id=str(uuid.uuid4()),
graph_init_params=init_params,
graph=graph,
graph_runtime_state=GraphRuntimeState(variable_pool=pool, start_at=time.perf_counter()),
config={
"id": "if-else",
"data": {
"title": "Boolean False Test",
"type": "if-else",
"logical_operator": "or",
"conditions": [
# Test boolean "is" operator (should be false)
{"comparison_operator": "is", "variable_selector": ["start", "bool_true"], "value": "false"},
# Test boolean "=" operator (should be false)
{"comparison_operator": "=", "variable_selector": ["start", "bool_false"], "value": "1"},
# Test boolean "not contains" operator (should be false)
{
"comparison_operator": "not contains",
"variable_selector": ["start", "bool_array"],
"value": "true",
},
],
},
},
)
# Mock db.session.close()
db.session.close = MagicMock()
# execute node
result = node._run()
assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
assert result.outputs is not None
assert result.outputs["result"] is False
def test_execute_if_else_boolean_cases_structure():
"""Test IfElseNode with boolean conditions using the new cases structure"""
graph_config = {"edges": [], "nodes": [{"data": {"type": "start"}, "id": "start"}]}
graph = Graph.init(graph_config=graph_config)
init_params = GraphInitParams(
tenant_id="1",
app_id="1",
workflow_type=WorkflowType.WORKFLOW,
workflow_id="1",
graph_config=graph_config,
user_id="1",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
)
# construct variable pool with boolean values
pool = VariablePool(
system_variables={SystemVariableKey.FILES: [], SystemVariableKey.USER_ID: "aaa"}, user_inputs={}
)
pool.add(["start", "bool_true"], True)
pool.add(["start", "bool_false"], False)
node = IfElseNode(
id=str(uuid.uuid4()),
graph_init_params=init_params,
graph=graph,
graph_runtime_state=GraphRuntimeState(variable_pool=pool, start_at=time.perf_counter()),
config={
"id": "if-else",
"data": {
"title": "Boolean Cases Test",
"type": "if-else",
"cases": [
{
"case_id": "true",
"logical_operator": "and",
"conditions": [
{
"comparison_operator": "is",
"variable_selector": ["start", "bool_true"],
"value": "true",
},
{
"comparison_operator": "is not",
"variable_selector": ["start", "bool_false"],
"value": "true",
},
],
}
],
},
},
)
# Mock db.session.close()
db.session.close = MagicMock()
# execute node
result = node._run()
assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
assert result.outputs is not None
assert result.outputs["result"] is True
assert result.outputs["selected_case_id"] == "true"

49
api/tests/unit_tests/factories/test_variable_factory.py
View File

@ -24,16 +24,18 @@ from core.variables.segments import (
ArrayNumberSegment,
ArrayObjectSegment,
ArrayStringSegment,
BooleanSegment,
FileSegment,
FloatSegment,
IntegerSegment,
NoneSegment,
ObjectSegment,
Segment,
StringSegment,
)
from core.variables.types import SegmentType
from factories import variable_factory
from factories.variable_factory import TypeMismatchError, build_segment_with_type
from factories.variable_factory import TypeMismatchError, build_segment, build_segment_with_type
def test_string_variable():
@ -139,6 +141,26 @@ def test_array_number_variable():
assert isinstance(variable.value[1], float)
def test_build_segment_scalar_values():
@dataclass
class TestCase:
value: Any
expected: Segment
description: str
cases = [
TestCase(
value=True,
expected=BooleanSegment(value=True),
description="build_segment with boolean should yield BooleanSegment",
)
]
for idx, c in enumerate(cases, 1):
seg = build_segment(c.value)
assert seg == c.expected, f"Test case {idx} failed: {c.description}"
def test_array_object_variable():
mapping = {
"id": str(uuid4()),
@ -847,15 +869,22 @@ class TestBuildSegmentValueErrors:
f"but got: {error_message}"
)
def test_build_segment_boolean_type_note(self):
"""Note: Boolean values are actually handled as integers in Python, so they don't raise ValueError."""
# Boolean values in Python are subclasses of int, so they get processed as integers
# True becomes IntegerSegment(value=1) and False becomes IntegerSegment(value=0)
def test_build_segment_boolean_type(self):
"""Test that Boolean values are correctly handled as boolean type, not integers."""
# Boolean values should now be processed as BooleanSegment, not IntegerSegment
# This is because the bool check now comes before the int check in build_segment
true_segment = variable_factory.build_segment(True)
false_segment = variable_factory.build_segment(False)
# Verify they are processed as integers, not as errors
assert true_segment.value == 1, "Test case 1 (boolean_true): Expected True to be processed as integer 1"
assert false_segment.value == 0, "Test case 2 (boolean_false): Expected False to be processed as integer 0"
assert true_segment.value_type == SegmentType.INTEGER
assert false_segment.value_type == SegmentType.INTEGER
# Verify they are processed as booleans, not integers
assert true_segment.value is True, "Test case 1 (boolean_true): Expected True to be processed as boolean True"
assert false_segment.value is False, (
"Test case 2 (boolean_false): Expected False to be processed as boolean False"
)
assert true_segment.value_type == SegmentType.BOOLEAN
assert false_segment.value_type == SegmentType.BOOLEAN
# Test array of booleans
bool_array_segment = variable_factory.build_segment([True, False, True])
assert bool_array_segment.value_type == SegmentType.ARRAY_BOOLEAN
assert bool_array_segment.value == [True, False, True]

47
simple_boolean_test.py Normal file
View File

@ -0,0 +1,47 @@
#!/usr/bin/env python3
"""
Simple test to verify boolean classes can be imported correctly.
"""
import sys
import os
# Add the api directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "api"))
try:
# Test that we can import the boolean classes
from core.variables.segments import BooleanSegment, ArrayBooleanSegment
from core.variables.variables import BooleanVariable, ArrayBooleanVariable
from core.variables.types import SegmentType
print("✅ Successfully imported BooleanSegment")
print("✅ Successfully imported ArrayBooleanSegment")
print("✅ Successfully imported BooleanVariable")
print("✅ Successfully imported ArrayBooleanVariable")
print("✅ Successfully imported SegmentType")
# Test that the segment types exist
print(f"✅ SegmentType.BOOLEAN = {SegmentType.BOOLEAN}")
print(f"✅ SegmentType.ARRAY_BOOLEAN = {SegmentType.ARRAY_BOOLEAN}")
# Test creating boolean segments directly
bool_seg = BooleanSegment(value=True)
print(f"✅ Created BooleanSegment: {bool_seg}")
print(f" Value type: {bool_seg.value_type}")
print(f" Value: {bool_seg.value}")
array_bool_seg = ArrayBooleanSegment(value=[True, False, True])
print(f"✅ Created ArrayBooleanSegment: {array_bool_seg}")
print(f" Value type: {array_bool_seg.value_type}")
print(f" Value: {array_bool_seg.value}")
print("\n🎉 All boolean class imports and basic functionality work correctly!")
except ImportError as e:
print(f"❌ Import error: {e}")
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()

118
test_boolean_conditions.py Normal file
View File

@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""
Simple test script to verify boolean condition support in IfElseNode
"""
import sys
import os
# Add the api directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "api"))
from core.workflow.utils.condition.processor import (
ConditionProcessor,
_evaluate_condition,
)
def test_boolean_conditions():
"""Test boolean condition evaluation"""
print("Testing boolean condition support...")
# Test boolean "is" operator
result = _evaluate_condition(value=True, operator="is", expected="true")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'is' with True value passed")
result = _evaluate_condition(value=False, operator="is", expected="false")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'is' with False value passed")
# Test boolean "is not" operator
result = _evaluate_condition(value=True, operator="is not", expected="false")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'is not' with True value passed")
result = _evaluate_condition(value=False, operator="is not", expected="true")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'is not' with False value passed")
# Test boolean "=" operator
result = _evaluate_condition(value=True, operator="=", expected="1")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean '=' with True=1 passed")
result = _evaluate_condition(value=False, operator="=", expected="0")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean '=' with False=0 passed")
# Test boolean "≠" operator
result = _evaluate_condition(value=True, operator="≠", expected="0")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean '≠' with True≠0 passed")
result = _evaluate_condition(value=False, operator="≠", expected="1")
assert result == True, f"Expected True, got {result}"
print("✓ Boolean '≠' with False≠1 passed")
# Test boolean "in" operator
result = _evaluate_condition(value=True, operator="in", expected=["true", "false"])
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'in' with True in array passed")
result = _evaluate_condition(value=False, operator="in", expected=["true", "false"])
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'in' with False in array passed")
# Test boolean "not in" operator
result = _evaluate_condition(value=True, operator="not in", expected=["false", "0"])
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'not in' with True not in [false, 0] passed")
# Test boolean "null" and "not null" operators
result = _evaluate_condition(value=True, operator="not null", expected=None)
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'not null' with True passed")
result = _evaluate_condition(value=False, operator="not null", expected=None)
assert result == True, f"Expected True, got {result}"
print("✓ Boolean 'not null' with False passed")
print("\n🎉 All boolean condition tests passed!")
def test_backward_compatibility():
"""Test that existing string and number conditions still work"""
print("\nTesting backward compatibility...")
# Test string conditions
result = _evaluate_condition(value="hello", operator="is", expected="hello")
assert result == True, f"Expected True, got {result}"
print("✓ String 'is' condition still works")
result = _evaluate_condition(value="hello", operator="contains", expected="ell")
assert result == True, f"Expected True, got {result}"
print("✓ String 'contains' condition still works")
# Test number conditions
result = _evaluate_condition(value=42, operator="=", expected="42")
assert result == True, f"Expected True, got {result}"
print("✓ Number '=' condition still works")
result = _evaluate_condition(value=42, operator=">", expected="40")
assert result == True, f"Expected True, got {result}"
print("✓ Number '>' condition still works")
print("✓ Backward compatibility maintained!")
if __name__ == "__main__":
try:
test_boolean_conditions()
test_backward_compatibility()
print(
"\n✅ All tests passed! Boolean support has been successfully added to IfElseNode."
)
except Exception as e:
print(f"\n❌ Test failed: {e}")
sys.exit(1)

67
test_boolean_contains_fix.py Normal file
View File

@ -0,0 +1,67 @@
#!/usr/bin/env python3
"""
Test script to verify the boolean array comparison fix in condition processor.
"""
import sys
import os
# Add the api directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "api"))
from core.workflow.utils.condition.processor import (
_assert_contains,
_assert_not_contains,
)
def test_boolean_array_contains():
"""Test that boolean arrays work correctly with string comparisons."""
# Test case 1: Boolean array [True, False, True] contains "true"
bool_array = [True, False, True]
# Should return True because "true" converts to True and True is in the array
result1 = _assert_contains(value=bool_array, expected="true")
print(f"Test 1 - [True, False, True] contains 'true': {result1}")
assert result1 == True, "Expected True but got False"
# Should return True because "false" converts to False and False is in the array
result2 = _assert_contains(value=bool_array, expected="false")
print(f"Test 2 - [True, False, True] contains 'false': {result2}")
assert result2 == True, "Expected True but got False"
# Test case 2: Boolean array [True, True] does not contain "false"
bool_array2 = [True, True]
result3 = _assert_contains(value=bool_array2, expected="false")
print(f"Test 3 - [True, True] contains 'false': {result3}")
assert result3 == False, "Expected False but got True"
# Test case 3: Test not_contains
result4 = _assert_not_contains(value=bool_array2, expected="false")
print(f"Test 4 - [True, True] not contains 'false': {result4}")
assert result4 == True, "Expected True but got False"
result5 = _assert_not_contains(value=bool_array, expected="true")
print(f"Test 5 - [True, False, True] not contains 'true': {result5}")
assert result5 == False, "Expected False but got True"
# Test case 4: Test with different string representations
result6 = _assert_contains(
value=bool_array, expected="1"
) # "1" should convert to True
print(f"Test 6 - [True, False, True] contains '1': {result6}")
assert result6 == True, "Expected True but got False"
result7 = _assert_contains(
value=bool_array, expected="0"
) # "0" should convert to False
print(f"Test 7 - [True, False, True] contains '0': {result7}")
assert result7 == True, "Expected True but got False"
print("\n✅ All boolean array comparison tests passed!")
if __name__ == "__main__":
test_boolean_array_contains()

99
test_boolean_factory.py Normal file
View File

@ -0,0 +1,99 @@
#!/usr/bin/env python3
"""
Simple test script to verify boolean type inference in variable factory.
"""
import sys
import os
# Add the api directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "api"))
try:
from factories.variable_factory import build_segment, segment_to_variable
from core.variables.segments import BooleanSegment, ArrayBooleanSegment
from core.variables.variables import BooleanVariable, ArrayBooleanVariable
from core.variables.types import SegmentType
def test_boolean_inference():
print("Testing boolean type inference...")
# Test single boolean values
true_segment = build_segment(True)
false_segment = build_segment(False)
print(f"True value: {true_segment}")
print(f"Type: {type(true_segment)}")
print(f"Value type: {true_segment.value_type}")
print(f"Is BooleanSegment: {isinstance(true_segment, BooleanSegment)}")
print(f"\nFalse value: {false_segment}")
print(f"Type: {type(false_segment)}")
print(f"Value type: {false_segment.value_type}")
print(f"Is BooleanSegment: {isinstance(false_segment, BooleanSegment)}")
# Test array of booleans
bool_array_segment = build_segment([True, False, True])
print(f"\nBoolean array: {bool_array_segment}")
print(f"Type: {type(bool_array_segment)}")
print(f"Value type: {bool_array_segment.value_type}")
print(
f"Is ArrayBooleanSegment: {isinstance(bool_array_segment, ArrayBooleanSegment)}"
)
# Test empty boolean array
empty_bool_array = build_segment([])
print(f"\nEmpty array: {empty_bool_array}")
print(f"Type: {type(empty_bool_array)}")
print(f"Value type: {empty_bool_array.value_type}")
# Test segment to variable conversion
bool_var = segment_to_variable(
segment=true_segment, selector=["test", "bool_var"], name="test_boolean"
)
print(f"\nBoolean variable: {bool_var}")
print(f"Type: {type(bool_var)}")
print(f"Is BooleanVariable: {isinstance(bool_var, BooleanVariable)}")
array_bool_var = segment_to_variable(
segment=bool_array_segment,
selector=["test", "array_bool_var"],
name="test_array_boolean",
)
print(f"\nArray boolean variable: {array_bool_var}")
print(f"Type: {type(array_bool_var)}")
print(
f"Is ArrayBooleanVariable: {isinstance(array_bool_var, ArrayBooleanVariable)}"
)
# Test that bool comes before int (critical ordering)
print(f"\nTesting bool vs int precedence:")
print(f"True is instance of bool: {isinstance(True, bool)}")
print(f"True is instance of int: {isinstance(True, int)}")
print(f"False is instance of bool: {isinstance(False, bool)}")
print(f"False is instance of int: {isinstance(False, int)}")
# Verify that boolean values are correctly inferred as boolean, not int
assert true_segment.value_type == SegmentType.BOOLEAN, (
"True should be inferred as BOOLEAN"
)
assert false_segment.value_type == SegmentType.BOOLEAN, (
"False should be inferred as BOOLEAN"
)
assert bool_array_segment.value_type == SegmentType.ARRAY_BOOLEAN, (
"Boolean array should be inferred as ARRAY_BOOLEAN"
)
print("\n✅ All boolean inference tests passed!")
if __name__ == "__main__":
test_boolean_inference()
except ImportError as e:
print(f"Import error: {e}")
print("Make sure you're running this from the correct directory")
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()

230
test_boolean_variable_assigner.py Normal file
View File

@ -0,0 +1,230 @@
#!/usr/bin/env python3
"""
Test script to verify boolean support in VariableAssigner node
"""
import sys
import os
# Add the api directory to the Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "api"))
from core.variables import SegmentType
from core.workflow.nodes.variable_assigner.v2.helpers import (
is_operation_supported,
is_constant_input_supported,
is_input_value_valid,
)
from core.workflow.nodes.variable_assigner.v2.enums import Operation
from core.workflow.nodes.variable_assigner.v2.constants import EMPTY_VALUE_MAPPING
def test_boolean_operation_support():
"""Test that boolean types support the correct operations"""
print("Testing boolean operation support...")
# Boolean should support SET, OVER_WRITE, and CLEAR
assert is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET
)
assert is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.OVER_WRITE
)
assert is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.CLEAR
)
# Boolean should NOT support arithmetic operations
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.ADD
)
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.SUBTRACT
)
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.MULTIPLY
)
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.DIVIDE
)
# Boolean should NOT support array operations
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.APPEND
)
assert not is_operation_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.EXTEND
)
print("✓ Boolean operation support tests passed")
def test_array_boolean_operation_support():
"""Test that array boolean types support the correct operations"""
print("Testing array boolean operation support...")
# Array boolean should support APPEND, EXTEND, SET, OVER_WRITE, CLEAR
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.APPEND
)
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.EXTEND
)
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.OVER_WRITE
)
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.CLEAR
)
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.REMOVE_FIRST
)
assert is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.REMOVE_LAST
)
# Array boolean should NOT support arithmetic operations
assert not is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.ADD
)
assert not is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.SUBTRACT
)
assert not is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.MULTIPLY
)
assert not is_operation_supported(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.DIVIDE
)
print("✓ Array boolean operation support tests passed")
def test_boolean_constant_input_support():
"""Test that boolean types support constant input for correct operations"""
print("Testing boolean constant input support...")
# Boolean should support constant input for SET and OVER_WRITE
assert is_constant_input_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET
)
assert is_constant_input_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.OVER_WRITE
)
# Boolean should NOT support constant input for arithmetic operations
assert not is_constant_input_supported(
variable_type=SegmentType.BOOLEAN, operation=Operation.ADD
)
print("✓ Boolean constant input support tests passed")
def test_boolean_input_validation():
"""Test that boolean input validation works correctly"""
print("Testing boolean input validation...")
# Boolean values should be valid for boolean type
assert is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET, value=True
)
assert is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET, value=False
)
assert is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.OVER_WRITE, value=True
)
# Non-boolean values should be invalid for boolean type
assert not is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET, value="true"
)
assert not is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET, value=1
)
assert not is_input_value_valid(
variable_type=SegmentType.BOOLEAN, operation=Operation.SET, value=0
)
print("✓ Boolean input validation tests passed")
def test_array_boolean_input_validation():
"""Test that array boolean input validation works correctly"""
print("Testing array boolean input validation...")
# Boolean values should be valid for array boolean append
assert is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.APPEND, value=True
)
assert is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN, operation=Operation.APPEND, value=False
)
# Boolean arrays should be valid for extend/overwrite
assert is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN,
operation=Operation.EXTEND,
value=[True, False, True],
)
assert is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN,
operation=Operation.OVER_WRITE,
value=[False, False],
)
# Non-boolean values should be invalid
assert not is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN,
operation=Operation.APPEND,
value="true",
)
assert not is_input_value_valid(
variable_type=SegmentType.ARRAY_BOOLEAN,
operation=Operation.EXTEND,
value=[True, "false"],
)
print("✓ Array boolean input validation tests passed")
def test_empty_value_mapping():
"""Test that empty value mapping includes boolean types"""
print("Testing empty value mapping...")
# Check that boolean types have correct empty values
assert SegmentType.BOOLEAN in EMPTY_VALUE_MAPPING
assert EMPTY_VALUE_MAPPING[SegmentType.BOOLEAN] is False
assert SegmentType.ARRAY_BOOLEAN in EMPTY_VALUE_MAPPING
assert EMPTY_VALUE_MAPPING[SegmentType.ARRAY_BOOLEAN] == []
print("✓ Empty value mapping tests passed")
def main():
"""Run all tests"""
print("Running VariableAssigner boolean support tests...\n")
try:
test_boolean_operation_support()
test_array_boolean_operation_support()
test_boolean_constant_input_support()
test_boolean_input_validation()
test_array_boolean_input_validation()
test_empty_value_mapping()
print(
"\n🎉 All tests passed! Boolean support has been successfully added to VariableAssigner."
)
except Exception as e:
print(f"\n❌ Test failed: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()