diff --git a/backend/workflow_models.py b/backend/workflow_models.py index 76ace63..0b98ef2 100644 --- a/backend/workflow_models.py +++ b/backend/workflow_models.py @@ -162,7 +162,13 @@ class Condition(BaseModel): Bedingung für einen Logik-Knoten. Unterstützt if/else-if/else-Logik. + + Note: Uses extra='forbid' to ensure proper Union resolution with LogicExpression. + If unknown fields are present (like 'operator', 'operands'), deserialization fails + and Pydantic tries LogicExpression instead. """ + model_config = {'extra': 'forbid'} + type: str = Field(default="if", description="Bedingungstyp: if, else-if, else") expression: Optional[LogicExpression] = Field(None, description="Logischer Ausdruck (null bei 'else')") then_path: Optional[str] = Field(None, description="Edge-ID für 'then'-Pfad") diff --git a/test_condition_parsing.py b/test_condition_parsing.py new file mode 100644 index 0000000..70fa53c --- /dev/null +++ b/test_condition_parsing.py @@ -0,0 +1,246 @@ +""" +Test Condition Parsing - Alle Formate und Verschachtelungen + +Testet ob Pydantic die verschiedenen Condition-Formate korrekt deserialisiert. +""" +import sys +import os + +# Force UTF-8 encoding on Windows +if sys.platform == 'win32': + import io + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') + sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') + +sys.path.insert(0, 'backend') + +from workflow_models import LogicExpression, Condition, WorkflowNode, LogicOperator +from pydantic import ValidationError +import json + +def test_case(name, data, expected_type, should_fail=False): + """Test a single condition format""" + print(f"\n{'='*60}") + print(f"TEST: {name}") + print(f"{'='*60}") + print(f"Input: {json.dumps(data, indent=2)}") + + try: + # Test 1: Direct deserialization + if expected_type == LogicExpression: + result = LogicExpression(**data) + elif expected_type == Condition: + result = Condition(**data) + + if should_fail: + print("❌ FAILED: Should have raised ValidationError but didn't") + return False + + print(f"✅ PASSED: Deserialized as {type(result).__name__}") + print(f"Result: {result.model_dump()}") + + # Test 2: As part of WorkflowNode + node_data = { + "id": "test_node", + "type": "logic", + "condition": data + } + node = WorkflowNode(**node_data) + print(f"✅ PASSED: WorkflowNode.condition type: {type(node.condition).__name__}") + + return True + + except ValidationError as e: + if should_fail: + print(f"✅ PASSED: Correctly raised ValidationError") + return True + else: + print(f"❌ FAILED: {e}") + return False + except Exception as e: + print(f"❌ FAILED: Unexpected error: {e}") + import traceback + traceback.print_exc() + return False + + +# ============================================================================ +# Test Cases +# ============================================================================ + +test_results = [] + +# Test 1: Simple comparison (UI format - einfachster Fall) +test_results.append(test_case( + "Simple comparison (eq)", + { + "operator": "eq", + "ref": "node_1.q1", + "value": "ja" + }, + LogicExpression +)) + +# Test 2: Simple AND (UI format - wie im Workflow) +test_results.append(test_case( + "Simple AND with 2 operands", + { + "operator": "and", + "operands": [ + {"operator": "eq", "ref": "node_5.qTiefananalyseRecovery", "value": "ja"}, + {"operator": "neq", "ref": "node_6.qKonsistenz", "value": "nein"} + ] + }, + LogicExpression +)) + +# Test 3: Simple OR +test_results.append(test_case( + "Simple OR with 3 operands", + { + "operator": "or", + "operands": [ + {"operator": "eq", "ref": "node_1.q1", "value": "ja"}, + {"operator": "eq", "ref": "node_1.q2", "value": "nein"}, + {"operator": "eq", "ref": "node_1.q3", "value": "unklar"} + ] + }, + LogicExpression +)) + +# Test 4: Nested AND/OR +test_results.append(test_case( + "Nested: OR with nested AND", + { + "operator": "or", + "operands": [ + { + "operator": "and", + "operands": [ + {"operator": "eq", "ref": "node_1.q1", "value": "ja"}, + {"operator": "neq", "ref": "node_1.q2", "value": "nein"} + ] + }, + {"operator": "eq", "ref": "node_2.q1", "value": "hoch"} + ] + }, + LogicExpression +)) + +# Test 5: Deep nesting (3 levels) +test_results.append(test_case( + "Deep nesting (3 levels)", + { + "operator": "and", + "operands": [ + { + "operator": "or", + "operands": [ + {"operator": "eq", "ref": "node_1.q1", "value": "ja"}, + { + "operator": "and", + "operands": [ + {"operator": "eq", "ref": "node_2.q1", "value": "hoch"}, + {"operator": "neq", "ref": "node_2.q2", "value": "niedrig"} + ] + } + ] + }, + {"operator": "eq", "ref": "node_3.q1", "value": "aktiv"} + ] + }, + LogicExpression +)) + +# Test 6: Different operators +test_results.append(test_case( + "Different comparison operators (gt, lt, in)", + { + "operator": "and", + "operands": [ + {"operator": "gt", "ref": "node_1.score", "value": 50}, + {"operator": "lt", "ref": "node_1.score", "value": 100}, + {"operator": "in", "ref": "node_1.category", "value": ["high", "medium"]} + ] + }, + LogicExpression +)) + +# Test 7: Legacy format (wrapped in Condition) +test_results.append(test_case( + "Legacy format: Condition with expression", + { + "type": "if", + "expression": { + "operator": "eq", + "ref": "node_1.q1", + "value": "ja" + }, + "then_path": "edge_1", + "else_path": "edge_2" + }, + Condition +)) + +# Test 8: NOT operator +test_results.append(test_case( + "NOT operator", + { + "operator": "not", + "operands": [ + {"operator": "eq", "ref": "node_1.q1", "value": "nein"} + ] + }, + LogicExpression +)) + +# Test 9: Complex real-world scenario +test_results.append(test_case( + "Complex real-world: Multiple nested conditions", + { + "operator": "and", + "operands": [ + { + "operator": "or", + "operands": [ + {"operator": "eq", "ref": "node_1.relevance", "value": "high"}, + { + "operator": "and", + "operands": [ + {"operator": "eq", "ref": "node_1.relevance", "value": "medium"}, + {"operator": "gt", "ref": "node_1.priority", "value": 5} + ] + } + ] + }, + {"operator": "neq", "ref": "node_2.status", "value": "blocked"}, + { + "operator": "in", + "ref": "node_3.category", + "value": ["training", "nutrition", "recovery"] + } + ] + }, + LogicExpression +)) + +# ============================================================================ +# Results Summary +# ============================================================================ + +print("\n" + "="*60) +print("TEST RESULTS SUMMARY") +print("="*60) + +passed = sum(test_results) +total = len(test_results) +failed = total - passed + +print(f"\n✅ Passed: {passed}/{total}") +if failed > 0: + print(f"❌ Failed: {failed}/{total}") + print(f"\n⚠️ CRITICAL: Some tests failed! Do NOT deploy until fixed.") + sys.exit(1) +else: + print(f"\n🎉 All tests passed! Safe to deploy.") + sys.exit(0) diff --git a/test_condition_union.py b/test_condition_union.py new file mode 100644 index 0000000..1fd7fe5 --- /dev/null +++ b/test_condition_union.py @@ -0,0 +1,113 @@ +""" +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!")