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