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!")
|