test: Add unit tests for End Node Template Engine
- test_end_node_template.py: Tests for execute_end_node() - Tests AUTO mode (backward compatible concatenation) - Tests TEMPLATE mode (Jinja2 rendering, conditionals) - Tests error handling (missing template, syntax errors) Note: Tests require Jinja2, run in Docker or CI/CD environment. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
fac76c28da
commit
7deca6c64d
244
backend/tests/test_end_node_template.py
Normal file
244
backend/tests/test_end_node_template.py
Normal file
|
|
@ -0,0 +1,244 @@
|
|||
"""
|
||||
Unit Tests: End Node Template Engine (Phase 4)
|
||||
|
||||
Tests execute_end_node() mit AUTO und TEMPLATE modes.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
from workflow_executor import execute_end_node
|
||||
from workflow_models import (
|
||||
WorkflowNode, EndNodeOutputMode, NodeStatus, NodeExecutionState
|
||||
)
|
||||
|
||||
|
||||
class TestEndNodeTemplateEngine:
|
||||
"""Tests für End Node Template Rendering"""
|
||||
|
||||
def test_auto_mode_concatenates_all_analyses(self):
|
||||
"""AUTO mode: Concatenates all analysis_core values"""
|
||||
# Setup: Node mit AUTO mode
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.AUTO
|
||||
)
|
||||
|
||||
# Setup: Context mit 2 executed nodes
|
||||
context = {
|
||||
"node_results": {
|
||||
"analysis1": NodeExecutionState(
|
||||
node_id="analysis1",
|
||||
status=NodeStatus.EXECUTED,
|
||||
analysis_core="Body analysis result",
|
||||
decision_signals={"relevanz": "hoch"}
|
||||
),
|
||||
"analysis2": NodeExecutionState(
|
||||
node_id="analysis2",
|
||||
status=NodeStatus.EXECUTED,
|
||||
analysis_core="Training recommendation",
|
||||
decision_signals={"prioritaet": "mittel"}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
# Execute
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
# Assert
|
||||
assert result.status == NodeStatus.EXECUTED
|
||||
assert "## analysis1" in result.analysis_core
|
||||
assert "Body analysis result" in result.analysis_core
|
||||
assert "## analysis2" in result.analysis_core
|
||||
assert "Training recommendation" in result.analysis_core
|
||||
print("✓ AUTO mode concatenation works")
|
||||
|
||||
|
||||
def test_auto_mode_skips_skipped_nodes(self):
|
||||
"""AUTO mode: Skips nodes with status SKIPPED"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.AUTO
|
||||
)
|
||||
|
||||
context = {
|
||||
"node_results": {
|
||||
"analysis1": NodeExecutionState(
|
||||
node_id="analysis1",
|
||||
status=NodeStatus.EXECUTED,
|
||||
analysis_core="Executed analysis"
|
||||
),
|
||||
"analysis2": NodeExecutionState(
|
||||
node_id="analysis2",
|
||||
status=NodeStatus.SKIPPED,
|
||||
analysis_core="This should not appear"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert "Executed analysis" in result.analysis_core
|
||||
assert "This should not appear" not in result.analysis_core
|
||||
print("✓ AUTO mode skips SKIPPED nodes")
|
||||
|
||||
|
||||
def test_template_mode_basic_rendering(self):
|
||||
"""TEMPLATE mode: Renders Jinja2 template with node placeholders"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.TEMPLATE,
|
||||
template=(
|
||||
"# Final Result\n\n"
|
||||
"## Analysis\n"
|
||||
"{{analysis1.analysis_core}}\n\n"
|
||||
"## Signals\n"
|
||||
"Relevanz: {{analysis1.relevanz}}"
|
||||
)
|
||||
)
|
||||
|
||||
context = {
|
||||
"node_results": {
|
||||
"analysis1": NodeExecutionState(
|
||||
node_id="analysis1",
|
||||
status=NodeStatus.EXECUTED,
|
||||
analysis_core="Test analysis content",
|
||||
decision_signals={"relevanz": "hoch"}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert result.status == NodeStatus.EXECUTED
|
||||
assert "# Final Result" in result.analysis_core
|
||||
assert "Test analysis content" in result.analysis_core
|
||||
assert "Relevanz: hoch" in result.analysis_core
|
||||
print("✓ TEMPLATE mode basic rendering works")
|
||||
|
||||
|
||||
def test_template_conditional_rendering(self):
|
||||
"""TEMPLATE mode: Conditional rendering with {% if node_id %}"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.TEMPLATE,
|
||||
template=(
|
||||
"# Result\n\n"
|
||||
"{% if analysis1 %}"
|
||||
"Analysis 1 executed\n"
|
||||
"{% endif %}"
|
||||
"{% if analysis2 %}"
|
||||
"Analysis 2 executed\n"
|
||||
"{% endif %}"
|
||||
)
|
||||
)
|
||||
|
||||
context = {
|
||||
"node_results": {
|
||||
"analysis1": NodeExecutionState(
|
||||
node_id="analysis1",
|
||||
status=NodeStatus.EXECUTED,
|
||||
analysis_core="Content 1"
|
||||
)
|
||||
# analysis2 not in results (optional path not taken)
|
||||
}
|
||||
}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert "Analysis 1 executed" in result.analysis_core
|
||||
assert "Analysis 2 executed" not in result.analysis_core
|
||||
print("✓ TEMPLATE mode conditional rendering works")
|
||||
|
||||
|
||||
def test_template_default_values(self):
|
||||
"""TEMPLATE mode: Default values for missing nodes"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.TEMPLATE,
|
||||
template=(
|
||||
"Result: {{missing_node.analysis_core|default('Nicht verfügbar')}}"
|
||||
)
|
||||
)
|
||||
|
||||
context = {"node_results": {}}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert "Nicht verfügbar" in result.analysis_core
|
||||
print("✓ TEMPLATE mode default values work")
|
||||
|
||||
|
||||
def test_template_missing_template_fails(self):
|
||||
"""TEMPLATE mode without template raises error"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.TEMPLATE,
|
||||
template=None # Missing template
|
||||
)
|
||||
|
||||
context = {"node_results": {}}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert result.status == NodeStatus.FAILED
|
||||
assert "no template defined" in result.error.lower()
|
||||
print("✓ TEMPLATE mode fails without template")
|
||||
|
||||
|
||||
def test_template_syntax_error_fails(self):
|
||||
"""TEMPLATE mode with invalid syntax fails gracefully"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.TEMPLATE,
|
||||
template="{% invalid syntax %}"
|
||||
)
|
||||
|
||||
context = {"node_results": {}}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert result.status == NodeStatus.FAILED
|
||||
assert "template rendering failed" in result.error.lower()
|
||||
print("✓ TEMPLATE mode handles syntax errors")
|
||||
|
||||
|
||||
def test_auto_mode_empty_results(self):
|
||||
"""AUTO mode with no executed nodes returns placeholder"""
|
||||
end_node = WorkflowNode(
|
||||
id="end",
|
||||
type="end",
|
||||
output_mode=EndNodeOutputMode.AUTO
|
||||
)
|
||||
|
||||
context = {"node_results": {}}
|
||||
|
||||
result = execute_end_node(end_node, context)
|
||||
|
||||
assert result.status == NodeStatus.EXECUTED
|
||||
assert "[No analysis generated]" in result.analysis_core
|
||||
print("✓ AUTO mode handles empty results")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run tests
|
||||
test = TestEndNodeTemplateEngine()
|
||||
|
||||
print("\n=== Testing End Node Template Engine ===\n")
|
||||
|
||||
test.test_auto_mode_concatenates_all_analyses()
|
||||
test.test_auto_mode_skips_skipped_nodes()
|
||||
test.test_template_mode_basic_rendering()
|
||||
test.test_template_conditional_rendering()
|
||||
test.test_template_default_values()
|
||||
test.test_template_missing_template_fails()
|
||||
test.test_template_syntax_error_fails()
|
||||
test.test_auto_mode_empty_results()
|
||||
|
||||
print("\n✅ All tests passed!\n")
|
||||
Loading…
Reference in New Issue
Block a user