forked from cleveragents/cleveragents-core
939 lines
31 KiB
Python
939 lines
31 KiB
Python
"""
|
|
Comprehensive unit tests for langgraph nodes module.
|
|
"""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from cleveragents.agents.tool import ToolAgent
|
|
from cleveragents.langgraph.nodes import Edge, Node, NodeConfig, NodeType
|
|
from cleveragents.langgraph.state import GraphState
|
|
|
|
|
|
@pytest.fixture
|
|
def agent_node_with_state():
|
|
"""Fixture for agent node with test state setup."""
|
|
mock_agent = AsyncMock()
|
|
mock_agent.name = "agent1"
|
|
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT, agent="agent1")
|
|
node = Node(config, {"agent1": mock_agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "Test"}]
|
|
|
|
return node, state, mock_agent
|
|
|
|
|
|
@pytest.fixture
|
|
def conditional_node_with_two_messages():
|
|
"""Fixture for conditional node with two messages in state."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "message_count", "operator": "eq", "value": 2}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "1"}, {"role": "assistant", "content": "2"}]
|
|
|
|
return node, state
|
|
|
|
|
|
class TestNodeType:
|
|
"""Test cases for NodeType enum."""
|
|
|
|
def test_node_type_values(self):
|
|
"""Test NodeType enum values."""
|
|
assert NodeType.AGENT.value == "agent"
|
|
assert NodeType.FUNCTION.value == "function"
|
|
assert NodeType.TOOL.value == "tool"
|
|
assert NodeType.CONDITIONAL.value == "conditional"
|
|
assert NodeType.SUBGRAPH.value == "subgraph"
|
|
assert NodeType.START.value == "start"
|
|
assert NodeType.END.value == "end"
|
|
|
|
|
|
class TestNodeConfig:
|
|
"""Test cases for NodeConfig dataclass."""
|
|
|
|
def test_node_config_creation(self):
|
|
"""Test NodeConfig creation."""
|
|
config = NodeConfig(name="test_node", type=NodeType.AGENT)
|
|
|
|
assert config.name == "test_node"
|
|
assert config.type == NodeType.AGENT
|
|
assert config.agent is None
|
|
assert config.function is None
|
|
assert config.tools == []
|
|
assert config.retry_policy is None
|
|
assert config.timeout is None
|
|
assert config.parallel is False
|
|
assert config.condition is None
|
|
assert config.subgraph is None
|
|
assert config.metadata == {}
|
|
|
|
def test_node_config_with_values(self):
|
|
"""Test NodeConfig with all values."""
|
|
config = NodeConfig(
|
|
name="test",
|
|
type=NodeType.AGENT,
|
|
agent="agent1",
|
|
function="func1",
|
|
tools=["tool1", "tool2"],
|
|
retry_policy={"max_retries": 3},
|
|
timeout=10.0,
|
|
parallel=True,
|
|
condition={"type": "always"},
|
|
subgraph="subgraph1",
|
|
metadata={"key": "value"}
|
|
)
|
|
|
|
assert config.name == "test"
|
|
assert config.agent == "agent1"
|
|
assert len(config.tools) == 2
|
|
assert config.timeout == 10.0
|
|
|
|
|
|
class TestEdge:
|
|
"""Test cases for Edge dataclass."""
|
|
|
|
def test_edge_creation(self):
|
|
"""Test Edge creation."""
|
|
edge = Edge(source="start", target="end")
|
|
|
|
assert edge.source == "start"
|
|
assert edge.target == "end"
|
|
assert edge.condition is None
|
|
assert edge.metadata == {}
|
|
|
|
def test_edge_with_condition(self):
|
|
"""Test Edge with condition."""
|
|
edge = Edge(
|
|
source="node1",
|
|
target="node2",
|
|
condition={"type": "always"},
|
|
metadata={"weight": 1}
|
|
)
|
|
|
|
assert edge.source == "node1"
|
|
assert edge.condition["type"] == "always"
|
|
assert edge.metadata["weight"] == 1
|
|
|
|
|
|
class TestNode:
|
|
"""Test cases for Node class."""
|
|
|
|
def test_node_init(self):
|
|
"""Test Node initialization."""
|
|
config = NodeConfig(name="test_node", type=NodeType.START)
|
|
node = Node(config)
|
|
|
|
assert node.name == "test_node"
|
|
assert node.type == NodeType.START
|
|
assert node.execution_count == 0
|
|
assert node.last_execution_time is None
|
|
assert node.last_error is None
|
|
|
|
def test_node_init_with_agents(self):
|
|
"""Test Node initialization with agents."""
|
|
config = NodeConfig(name="test", type=NodeType.AGENT)
|
|
mock_agent = Mock()
|
|
agents = {"agent1": mock_agent}
|
|
|
|
node = Node(config, agents)
|
|
|
|
assert node.agents == agents
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_start_node(self):
|
|
"""Test executing a START node."""
|
|
config = NodeConfig(name="start", type=NodeType.START)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["current_node"] == "start"
|
|
assert result["started"] is True
|
|
assert node.execution_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_end_node(self):
|
|
"""Test executing an END node."""
|
|
config = NodeConfig(name="end", type=NodeType.END)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["current_node"] == "end"
|
|
assert result["completed"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_node(self):
|
|
"""Test executing an AGENT node."""
|
|
mock_agent = AsyncMock()
|
|
mock_agent.process_message = AsyncMock(return_value="Agent response")
|
|
mock_agent.name = "agent1"
|
|
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT, agent="agent1")
|
|
node = Node(config, {"agent1": mock_agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "Hello"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "messages" in result
|
|
assert len(result["messages"]) == 1
|
|
assert result["messages"][0]["role"] == "assistant"
|
|
assert result["messages"][0]["content"] == "Agent response"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_node_no_agent_specified(self):
|
|
"""Test executing AGENT node without agent specified raises error."""
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should return error state
|
|
assert "error" in result
|
|
assert "failed_node" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_node_agent_not_found(self):
|
|
"""Test executing AGENT node when agent not found."""
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT, agent="missing")
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "error" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_tool_agent(self):
|
|
"""Test executing AGENT node with ToolAgent."""
|
|
mock_tool_agent = AsyncMock(spec=ToolAgent)
|
|
mock_tool_agent.process_message = AsyncMock(return_value="Tool response")
|
|
mock_tool_agent.name = "tool1"
|
|
|
|
config = NodeConfig(name="tool_node", type=NodeType.AGENT, agent="tool1")
|
|
node = Node(config, {"tool1": mock_tool_agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "assistant", "content": "Use tool"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "messages" in result
|
|
# ToolAgent should receive the last message
|
|
mock_tool_agent.process_message.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_no_messages(self):
|
|
"""Test executing AGENT node with no messages."""
|
|
mock_agent = AsyncMock()
|
|
mock_agent.process_message = AsyncMock(return_value="Response")
|
|
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT, agent="agent1")
|
|
node = Node(config, {"agent1": mock_agent})
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should call agent with empty string
|
|
mock_agent.process_message.assert_called_once()
|
|
assert "messages" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_with_metadata(self):
|
|
"""Test executing AGENT node with metadata."""
|
|
mock_agent = AsyncMock()
|
|
mock_agent.process_message = AsyncMock(return_value="Response")
|
|
|
|
config = NodeConfig(name="agent_node", type=NodeType.AGENT, agent="agent1")
|
|
node = Node(config, {"agent1": mock_agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "Test"}]
|
|
state.metadata = {"unsafe_mode": True}
|
|
|
|
_ = await node.execute(state)
|
|
|
|
# Metadata should be passed to agent
|
|
call_args = mock_agent.process_message.call_args
|
|
context = call_args[0][1]
|
|
assert "unsafe_mode" in context
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_agent_error_handling(self, agent_node_with_state):
|
|
"""Test AGENT node error handling."""
|
|
node, state, mock_agent = agent_node_with_state
|
|
mock_agent.process_message = AsyncMock(side_effect=Exception("Agent error"))
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should return error message in messages
|
|
assert "messages" in result
|
|
assert "Error processing message" in result["messages"][0]["content"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_summarize(self):
|
|
"""Test executing FUNCTION node with summarize function."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="summarize")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg1"}, {"role": "assistant", "content": "msg2"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "metadata" in result
|
|
assert "summary" in result["metadata"]
|
|
assert "2 messages" in result["metadata"]["summary"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_route_question(self):
|
|
"""Test executing FUNCTION node with route function for questions."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="route")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "What is this?"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["route"] == "question"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_route_statement(self):
|
|
"""Test executing FUNCTION node with route function for statements."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="route")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "This is a statement"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["route"] == "statement"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_route_default(self):
|
|
"""Test executing FUNCTION node with route function default."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="route")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["route"] == "default"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_validate(self):
|
|
"""Test executing FUNCTION node with validate function."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="validate")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["valid"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_no_function(self):
|
|
"""Test executing FUNCTION node without function specified."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "error" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_function_node_unknown_function(self):
|
|
"""Test executing FUNCTION node with unknown function."""
|
|
config = NodeConfig(name="func_node", type=NodeType.FUNCTION, function="unknown")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should return empty dict
|
|
assert result["current_node"] == "func_node"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_tool_node(self):
|
|
"""Test executing TOOL node."""
|
|
config = NodeConfig(name="tool_node", type=NodeType.TOOL, tools=["tool1", "tool2"])
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "metadata" in result
|
|
assert "tool_results" in result["metadata"]
|
|
assert len(result["metadata"]["tool_results"]) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_tool_node_no_tools(self):
|
|
"""Test executing TOOL node without tools."""
|
|
config = NodeConfig(name="tool_node", type=NodeType.TOOL)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["current_node"] == "tool_node"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_always(self):
|
|
"""Test executing CONDITIONAL node with always condition."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "always"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_never(self):
|
|
"""Test executing CONDITIONAL node with never condition."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "never"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_has_messages_true(self):
|
|
"""Test CONDITIONAL node with has_messages condition (true)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "has_messages"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_has_messages_false(self):
|
|
"""Test CONDITIONAL node with has_messages condition (false)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "has_messages"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_message_count_gt(self, conditional_node_with_two_messages):
|
|
"""Test CONDITIONAL node with message_count gt condition."""
|
|
node, state = conditional_node_with_two_messages
|
|
# Override condition for this test
|
|
node.config.condition = {"type": "message_count", "operator": "gt", "value": 1}
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_message_count_lt(self):
|
|
"""Test CONDITIONAL node with message_count lt condition."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "message_count", "operator": "lt", "value": 5}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_message_count_eq(self, conditional_node_with_two_messages):
|
|
"""Test CONDITIONAL node with message_count eq condition."""
|
|
node, state = conditional_node_with_two_messages
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_message_count_unknown_operator(self):
|
|
"""Test CONDITIONAL node with unknown operator."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "message_count", "operator": "unknown", "value": 1}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_metadata_check_true(self):
|
|
"""Test CONDITIONAL node with metadata_check (true)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "metadata_check", "key": "status", "value": "ready"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.metadata = {"status": "ready"}
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_metadata_check_false(self):
|
|
"""Test CONDITIONAL node with metadata_check (false)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "metadata_check", "key": "status", "value": "ready"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.metadata = {"status": "pending"}
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_contains_true(self):
|
|
"""Test CONDITIONAL node with content_contains (true)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_contains", "text": "hello"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "hello world"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_contains_false(self):
|
|
"""Test CONDITIONAL node with content_contains (false)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_contains", "text": "goodbye"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "hello"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_not_contains_true(self):
|
|
"""Test CONDITIONAL node with content_not_contains (true)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_not_contains", "text": "goodbye"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "hello"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_starts_with_true(self):
|
|
"""Test CONDITIONAL node with content_starts_with (true)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_starts_with", "text": "hello"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "hello world"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_starts_with_false(self):
|
|
"""Test CONDITIONAL node with content_starts_with (false)."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_starts_with", "text": "world"}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "hello world"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_custom_function(self):
|
|
"""Test CONDITIONAL node with custom function."""
|
|
def custom_func(state):
|
|
return len(state.messages) > 0
|
|
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "custom", "function": custom_func}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg"}]
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_no_condition(self):
|
|
"""Test CONDITIONAL node without condition."""
|
|
config = NodeConfig(name="cond_node", type=NodeType.CONDITIONAL)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_subgraph_node(self):
|
|
"""Test executing SUBGRAPH node."""
|
|
config = NodeConfig(name="sub_node", type=NodeType.SUBGRAPH, subgraph="subgraph1")
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["subgraph"] == "subgraph1"
|
|
assert result["metadata"]["subgraph_pending"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_subgraph_no_subgraph(self):
|
|
"""Test executing SUBGRAPH node without subgraph specified."""
|
|
config = NodeConfig(name="sub_node", type=NodeType.SUBGRAPH)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert "error" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_with_retry_policy(self):
|
|
"""Test node execution with retry policy for function nodes."""
|
|
# Use a function node that can actually raise an error
|
|
config = NodeConfig(
|
|
name="func_node",
|
|
type=NodeType.FUNCTION,
|
|
function=None, # This will cause an error
|
|
retry_policy={"max_retries": 1, "delay": 0.01}
|
|
)
|
|
node = Node(config)
|
|
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Retry policy with max_retries=1 means 1 initial attempt + 1 retry = 2 total
|
|
assert node.execution_count == 2
|
|
assert "error" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_retry_all_fail(self, agent_node_with_state):
|
|
"""Test node execution when all retries fail."""
|
|
# Agent errors are caught and returned as error messages, not error state
|
|
node, state, mock_agent = agent_node_with_state
|
|
mock_agent.process_message = AsyncMock(side_effect=Exception("Always fails"))
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Agent errors are caught and returned in messages
|
|
assert "messages" in result
|
|
assert "Error processing message" in result["messages"][0]["content"]
|
|
assert node.last_error is None # Error was handled
|
|
|
|
def test_can_execute_parallel_true(self):
|
|
"""Test can_execute_parallel returns True."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT, parallel=True)
|
|
node = Node(config)
|
|
|
|
assert node.can_execute_parallel() is True
|
|
|
|
def test_can_execute_parallel_false(self):
|
|
"""Test can_execute_parallel returns False."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT, parallel=False)
|
|
node = Node(config)
|
|
|
|
assert node.can_execute_parallel() is False
|
|
|
|
def test_get_timeout(self):
|
|
"""Test get_timeout returns configured timeout."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT, timeout=5.0)
|
|
node = Node(config)
|
|
|
|
assert node.get_timeout() == 5.0
|
|
|
|
def test_get_timeout_none(self):
|
|
"""Test get_timeout returns None when not set."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
assert node.get_timeout() is None
|
|
|
|
def test_get_edges(self):
|
|
"""Test get_edges returns outgoing edges."""
|
|
config = NodeConfig(name="node1", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
edges = [
|
|
Edge("node1", "node2"),
|
|
Edge("node2", "node3"),
|
|
Edge("node1", "node3"),
|
|
]
|
|
|
|
outgoing = node.get_edges(edges)
|
|
|
|
assert len(outgoing) == 2
|
|
assert all(e.source == "node1" for e in outgoing)
|
|
|
|
def test_evaluate_edge_condition_no_condition(self):
|
|
"""Test evaluate_edge_condition with no condition."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
edge = Edge("node1", "node2")
|
|
state = GraphState()
|
|
|
|
result = node.evaluate_edge_condition(edge, state)
|
|
|
|
assert result is True
|
|
|
|
def test_evaluate_edge_condition_with_condition(self):
|
|
"""Test evaluate_edge_condition with condition."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
edge = Edge("node1", "node2", condition={"type": "always"})
|
|
state = GraphState()
|
|
|
|
result = node.evaluate_edge_condition(edge, state)
|
|
|
|
assert result is True
|
|
|
|
def test_evaluate_edge_condition_in_event_loop(self):
|
|
"""Test evaluate_edge_condition when already in event loop."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
edge = Edge("node1", "node2", condition={"type": "has_messages"})
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "msg"}]
|
|
|
|
# Simulate being in an event loop
|
|
async def test_in_loop():
|
|
result = node.evaluate_edge_condition(edge, state)
|
|
return result
|
|
|
|
result = asyncio.run(test_in_loop())
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_not_contains_no_messages(self):
|
|
"""Test CONDITIONAL node with content_not_contains and no messages."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_not_contains", "text": "test"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# No messages means text is not contained
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_starts_with_no_messages(self):
|
|
"""Test CONDITIONAL node with content_starts_with and no messages."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_starts_with", "text": "test"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_content_contains_no_messages(self):
|
|
"""Test CONDITIONAL node with content_contains and no messages."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "content_contains", "text": "test"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
assert result["metadata"]["condition_result"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_custom_no_function(self):
|
|
"""Test CONDITIONAL node with custom type but no function."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "custom", "function": None}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should default to True
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_custom_non_callable(self):
|
|
"""Test CONDITIONAL node with custom type but non-callable function."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "custom", "function": "not_callable"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should default to True
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_conditional_unknown_type(self):
|
|
"""Test CONDITIONAL node with unknown condition type."""
|
|
config = NodeConfig(
|
|
name="cond_node",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"type": "unknown_type"}
|
|
)
|
|
node = Node(config)
|
|
state = GraphState()
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should default to True
|
|
assert result["metadata"]["condition_result"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_node_non_dict_result(self, agent_node_with_state):
|
|
"""Test execute when result is not a dict."""
|
|
# This tests the case where result is not a dict
|
|
node, state, mock_agent = agent_node_with_state
|
|
mock_agent.process_message = AsyncMock(return_value=None) # Returns None
|
|
|
|
result = await node.execute(state)
|
|
|
|
# Should still have current_node
|
|
assert "current_node" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_edge_condition_non_bool_result(self):
|
|
"""Test evaluate_edge_condition with non-boolean result."""
|
|
config = NodeConfig(name="node", type=NodeType.AGENT)
|
|
node = Node(config)
|
|
|
|
# Create a custom function that returns a non-bool
|
|
def custom_func(_state):
|
|
return 1 # Non-boolean
|
|
|
|
edge = Edge("node1", "node2", condition={"type": "custom", "function": custom_func})
|
|
state = GraphState()
|
|
|
|
result = node.evaluate_edge_condition(edge, state)
|
|
|
|
# Should convert to bool
|
|
assert result is True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|
|
|