test_workflow_app_config.py
1 """ 2 Unit tests for WorkflowAppConfig class. 3 4 Tests that WorkflowAppConfig validates correctly and the 'name' field 5 is properly used as the primary workflow identifier. 6 """ 7 8 import pytest 9 from pydantic import ValidationError 10 11 from solace_agent_mesh.workflow.app import ( 12 WorkflowAppConfig, 13 WorkflowDefinition, 14 AgentNode, 15 ) 16 17 18 class TestWorkflowAppConfigName: 19 """Tests for WorkflowAppConfig 'name' field.""" 20 21 def test_workflow_app_config_with_name_field(self): 22 """WorkflowAppConfig accepts 'name' field as the primary identifier.""" 23 config = WorkflowAppConfig( 24 namespace="test_namespace", 25 name="TestWorkflow", 26 model="gemini-1.5-pro", 27 workflow=WorkflowDefinition( 28 description="Test workflow", 29 nodes=[ 30 AgentNode(id="step1", type="agent", agent_name="Agent1"), 31 ], 32 output_mapping={"result": "{{step1.output}}"}, 33 ), 34 ) 35 assert config.name == "TestWorkflow" 36 assert config.agent_name == "TestWorkflow" 37 38 def test_name_field_is_required(self): 39 """WorkflowAppConfig requires 'name' field.""" 40 with pytest.raises(ValidationError) as excinfo: 41 WorkflowAppConfig( 42 namespace="test_namespace", 43 model="gemini-1.5-pro", 44 workflow=WorkflowDefinition( 45 description="Test workflow", 46 nodes=[ 47 AgentNode(id="step1", type="agent", agent_name="Agent1"), 48 ], 49 output_mapping={"result": "{{step1.output}}"}, 50 ), 51 ) 52 assert "name" in str(excinfo.value).lower() 53 54 def test_agent_name_auto_populated_from_name(self): 55 """agent_name is automatically populated from name via validator.""" 56 config = WorkflowAppConfig( 57 namespace="test_namespace", 58 name="MyWorkflow", 59 model="gemini-1.5-pro", 60 workflow=WorkflowDefinition( 61 description="Test workflow", 62 nodes=[ 63 AgentNode(id="step1", type="agent", agent_name="Agent1"), 64 ], 65 output_mapping={"result": "{{step1.output}}"}, 66 ), 67 ) 68 # Both should be equal after validator runs 69 assert config.name == "MyWorkflow" 70 assert config.agent_name == "MyWorkflow" 71 72 73 class TestWorkflowAppConfigValidation: 74 """Tests for WorkflowAppConfig validation of workflow definitions.""" 75 76 def test_workflow_definition_validation(self): 77 """Workflow definition is validated during config creation.""" 78 # Invalid workflow should fail validation 79 with pytest.raises(ValidationError): 80 WorkflowAppConfig( 81 namespace="test_namespace", 82 name="TestWorkflow", 83 model="gemini-1.5-pro", 84 workflow=WorkflowDefinition( 85 description="Test workflow", 86 nodes=[ 87 AgentNode(id="step1", type="agent", agent_name="Agent1"), 88 AgentNode( 89 id="step2", 90 type="agent", 91 agent_name="Agent2", 92 depends_on=["nonexistent"], # Invalid dependency 93 ), 94 ], 95 output_mapping={"result": "{{step2.output}}"}, 96 ), 97 ) 98 99 def test_workflow_config_with_custom_timeouts(self): 100 """WorkflowAppConfig accepts custom timeout values.""" 101 config = WorkflowAppConfig( 102 namespace="test_namespace", 103 name="TimeoutWorkflow", 104 model="gemini-1.5-pro", 105 workflow=WorkflowDefinition( 106 description="Complete workflow", 107 nodes=[ 108 AgentNode(id="step1", type="agent", agent_name="Agent1"), 109 ], 110 output_mapping={"result": "{{step1.output}}"}, 111 ), 112 max_workflow_execution_time_seconds=3600, 113 default_node_timeout_seconds=600, 114 node_cancellation_timeout_seconds=60, 115 default_max_map_items=200, 116 ) 117 assert config.max_workflow_execution_time_seconds == 3600 118 assert config.default_node_timeout_seconds == 600 119 assert config.node_cancellation_timeout_seconds == 60 120 assert config.default_max_map_items == 200 121 122 def test_workflow_config_with_default_timeouts(self): 123 """WorkflowAppConfig uses sensible timeout defaults.""" 124 config = WorkflowAppConfig( 125 namespace="test_namespace", 126 name="DefaultTimeoutWorkflow", 127 model="gemini-1.5-pro", 128 workflow=WorkflowDefinition( 129 description="Test workflow", 130 nodes=[ 131 AgentNode(id="step1", type="agent", agent_name="Agent1"), 132 ], 133 output_mapping={"result": "{{step1.output}}"}, 134 ), 135 ) 136 # Check defaults 137 assert config.max_workflow_execution_time_seconds == 1800 # 30 minutes 138 assert config.default_node_timeout_seconds == 300 # 5 minutes 139 assert config.node_cancellation_timeout_seconds == 30 140 assert config.default_max_map_items == 100 141 142 143 class TestWorkflowAppConfigFromDict: 144 """Tests for creating WorkflowAppConfig from dictionaries.""" 145 146 def test_model_validate_with_dict(self): 147 """WorkflowAppConfig can be created from dict via model_validate.""" 148 config_dict = { 149 "namespace": "test_namespace", 150 "name": "DictCreatedWorkflow", 151 "model": "gemini-1.5-pro", 152 "workflow": { 153 "description": "Test workflow", 154 "nodes": [ 155 {"id": "step1", "type": "agent", "agent_name": "Agent1"}, 156 ], 157 "output_mapping": {"result": "{{step1.output}}"}, 158 }, 159 } 160 config = WorkflowAppConfig.model_validate(config_dict) 161 assert config.name == "DictCreatedWorkflow" 162 assert config.agent_name == "DictCreatedWorkflow" 163 assert config.namespace == "test_namespace" 164 165 def test_model_validate_requires_name(self): 166 """model_validate also requires 'name' field.""" 167 config_dict = { 168 "namespace": "test_namespace", 169 "workflow": { 170 "description": "Test workflow", 171 "nodes": [ 172 {"id": "step1", "type": "agent", "agent_name": "Agent1"}, 173 ], 174 "output_mapping": {"result": "{{step1.output}}"}, 175 }, 176 } 177 with pytest.raises(ValidationError) as excinfo: 178 WorkflowAppConfig.model_validate(config_dict) 179 assert "name" in str(excinfo.value).lower()