Lars
ba04e0c0b6
Critical fix: Without extra='forbid', Pydantic accepted UI format
{operator: "and", operands: [...]} as valid Condition by ignoring
unknown fields, resulting in Condition(expression=None).
With extra='forbid':
- Condition rejects unknown fields → fails
- Union tries next type → LogicExpression → success
Test Results (9/9 passed):
- Simple comparisons (eq, neq, gt, lt, in) ✅
- AND/OR combinations ✅
- Deep nesting (3+ levels) ✅
- NOT operator ✅
- All operators (eq, neq, in, not_in, gt, lt, gte, lte, and, or, not) ✅
- Legacy format (Condition wrapper) ✅
- Complex real-world scenarios ✅
Added comprehensive test suite in:
- test_condition_parsing.py (9 test cases)
- test_condition_union.py (Union resolution verification)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""
|
|
Test Union[LogicExpression, Condition] type resolution
|
|
"""
|
|
import sys
|
|
import os
|
|
|
|
if sys.platform == 'win32':
|
|
import io
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
|
|
|
sys.path.insert(0, 'backend')
|
|
|
|
from workflow_models import LogicExpression, Condition, WorkflowNode
|
|
import json
|
|
|
|
# Test 1: UI format (should be LogicExpression)
|
|
print("\n" + "="*60)
|
|
print("TEST 1: UI Format (direct LogicExpression)")
|
|
print("="*60)
|
|
|
|
ui_format = {
|
|
"operator": "and",
|
|
"operands": [
|
|
{"operator": "eq", "ref": "node_5.qTiefananalyseRecovery", "value": "ja"},
|
|
{"operator": "neq", "ref": "node_6.qKonsistenz", "value": "nein"}
|
|
]
|
|
}
|
|
|
|
node = WorkflowNode(
|
|
id="test_node",
|
|
type="logic",
|
|
condition=ui_format
|
|
)
|
|
|
|
print(f"Input: {json.dumps(ui_format, indent=2)}")
|
|
print(f"node.condition type: {type(node.condition).__name__}")
|
|
print(f"Expected: LogicExpression")
|
|
|
|
if isinstance(node.condition, LogicExpression):
|
|
print("✅ CORRECT: Deserialized as LogicExpression")
|
|
print(f"Has operator: {hasattr(node.condition, 'operator')} = {node.condition.operator if hasattr(node.condition, 'operator') else 'N/A'}")
|
|
print(f"Has operands: {hasattr(node.condition, 'operands')} = {len(node.condition.operands) if hasattr(node.condition, 'operands') and node.condition.operands else 'N/A'}")
|
|
elif isinstance(node.condition, Condition):
|
|
print("❌ WRONG: Deserialized as Condition (should be LogicExpression)")
|
|
print(f"Has expression: {hasattr(node.condition, 'expression')} = {type(node.condition.expression).__name__ if hasattr(node.condition, 'expression') and node.condition.expression else 'N/A'}")
|
|
if hasattr(node.condition, 'expression') and node.condition.expression:
|
|
print(f"expression.operator: {node.condition.expression.operator}")
|
|
print(f"expression.operands: {len(node.condition.expression.operands) if node.condition.expression.operands else 0}")
|
|
else:
|
|
print(f"❌ UNEXPECTED TYPE: {type(node.condition)}")
|
|
|
|
# Test 2: Legacy format (should be Condition)
|
|
print("\n" + "="*60)
|
|
print("TEST 2: Legacy Format (Condition with expression)")
|
|
print("="*60)
|
|
|
|
legacy_format = {
|
|
"type": "if",
|
|
"expression": {
|
|
"operator": "eq",
|
|
"ref": "node_1.q1",
|
|
"value": "ja"
|
|
},
|
|
"then_path": "edge_1",
|
|
"else_path": "edge_2"
|
|
}
|
|
|
|
node2 = WorkflowNode(
|
|
id="test_node2",
|
|
type="logic",
|
|
condition=legacy_format
|
|
)
|
|
|
|
print(f"Input: {json.dumps(legacy_format, indent=2)}")
|
|
print(f"node.condition type: {type(node2.condition).__name__}")
|
|
print(f"Expected: Condition")
|
|
|
|
if isinstance(node2.condition, Condition):
|
|
print("✅ CORRECT: Deserialized as Condition")
|
|
print(f"Has expression: {hasattr(node2.condition, 'expression')}")
|
|
print(f"Has then_path: {hasattr(node2.condition, 'then_path')} = {node2.condition.then_path}")
|
|
elif isinstance(node2.condition, LogicExpression):
|
|
print("❌ WRONG: Deserialized as LogicExpression (should be Condition)")
|
|
else:
|
|
print(f"❌ UNEXPECTED TYPE: {type(node2.condition)}")
|
|
|
|
# Test 3: Check what executor would do
|
|
print("\n" + "="*60)
|
|
print("TEST 3: Executor Logic Simulation")
|
|
print("="*60)
|
|
|
|
from workflow_models import LogicExpression, Condition
|
|
|
|
for test_name, node in [("UI Format", node), ("Legacy Format", node2)]:
|
|
print(f"\n{test_name}:")
|
|
print(f" condition type: {type(node.condition).__name__}")
|
|
|
|
if isinstance(node.condition, LogicExpression):
|
|
print(" ✅ Executor would use: node.condition directly")
|
|
expression = node.condition
|
|
elif isinstance(node.condition, Condition):
|
|
print(" ✅ Executor would use: node.condition.expression")
|
|
expression = node.condition.expression if node.condition.expression else None
|
|
else:
|
|
print(f" ❌ Executor would fail: Unknown type {type(node.condition)}")
|
|
expression = None
|
|
|
|
if expression:
|
|
print(f" Expression type: {type(expression).__name__}")
|
|
print(f" Expression operator: {expression.operator}")
|
|
print(f" Expression has operands: {expression.operands is not None}")
|
|
else:
|
|
print(f" ❌ No expression found!")
|