CleverThis Engineering
5bbda89386
- Fix bare excepts (replace with except Exception:) - Remove wildcard star imports from test step files - Add explicit imports for ConfigurationError, AgentCreationError - Add 'from err' to raise statements in except blocks - Fix loop variable binding with default arguments (B023) - Organize imports, fix trailing whitespace and blank line whitespace - Bump Python to 3.13 and update tool configurations
147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
"""
|
|
Step definitions for LangGraph visualization feature tests.
|
|
These steps test the ReactiveCleverAgentsApp's visualize_network() method.
|
|
"""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveractors.core.application import ReactiveCleverAgentsApp
|
|
from cleveractors.reactive.route import RouteType
|
|
|
|
|
|
@given("a basic application with simple config")
|
|
def step_create_basic_app(context):
|
|
"""Create a basic application with simple configuration."""
|
|
context.app = ReactiveCleverAgentsApp()
|
|
context.app.agents = {}
|
|
context.app.config = Mock()
|
|
context.app.config.routes = {}
|
|
context.app.config.merges = []
|
|
context.app.langgraph_bridge = Mock()
|
|
context.app.langgraph_bridge.list_graphs = Mock(return_value=[])
|
|
|
|
|
|
@given("an application with agents and routes")
|
|
def step_create_app_with_agents_and_routes(context):
|
|
"""Create an application with agents and routes."""
|
|
context.app = ReactiveCleverAgentsApp()
|
|
|
|
# Add some agents
|
|
context.app.agents = {"analyzer": Mock(), "processor": Mock()}
|
|
|
|
# Add configuration with routes
|
|
context.app.config = Mock()
|
|
|
|
# Create a stream route with operators and publications
|
|
mock_stream_route = Mock()
|
|
mock_stream_route.type = RouteType.STREAM
|
|
mock_stream_route.operators = [{"type": "map", "params": {"agent": "analyzer"}}]
|
|
mock_stream_route.publications = ["output_stream"]
|
|
|
|
context.app.config.routes = {"input_stream": mock_stream_route}
|
|
context.app.config.merges = []
|
|
|
|
# Mock langgraph bridge
|
|
context.app.langgraph_bridge = Mock()
|
|
context.app.langgraph_bridge.list_graphs = Mock(return_value=[])
|
|
|
|
|
|
@given("an application with multi-node graph")
|
|
def step_create_app_with_multi_node_graph(context):
|
|
"""Create an application with a multi-node graph."""
|
|
context.app = ReactiveCleverAgentsApp()
|
|
|
|
# Add agents
|
|
context.app.agents = {"agent1": Mock(), "agent2": Mock(), "agent3": Mock()}
|
|
|
|
# Add configuration with routes including graph
|
|
context.app.config = Mock()
|
|
|
|
# Create multiple routes
|
|
mock_stream_route = Mock()
|
|
mock_stream_route.type = RouteType.STREAM
|
|
mock_stream_route.operators = [{"type": "map", "params": {"agent": "agent1"}}]
|
|
mock_stream_route.publications = ["stream2"]
|
|
|
|
mock_graph_route = Mock()
|
|
mock_graph_route.type = RouteType.GRAPH
|
|
|
|
context.app.config.routes = {
|
|
"stream1": mock_stream_route,
|
|
"graph1": mock_graph_route,
|
|
}
|
|
context.app.config.merges = [
|
|
{"sources": ["stream1", "stream2"], "target": "merged_stream"}
|
|
]
|
|
|
|
# Mock langgraph bridge with a graph
|
|
context.app.langgraph_bridge = Mock()
|
|
context.app.langgraph_bridge.list_graphs = Mock(return_value=["graph1"])
|
|
|
|
# Create a mock graph that returns visualization
|
|
mock_graph = Mock()
|
|
mock_graph.visualize = Mock(
|
|
return_value="""graph TD
|
|
start[Start] --> node1[Agent Node]
|
|
node1 --> node2[Process Node]
|
|
node2 --> end[End]"""
|
|
)
|
|
context.app.langgraph_bridge.get_graph = Mock(return_value=mock_graph)
|
|
|
|
|
|
@when('visualizing network with format "{format_type}"')
|
|
def step_visualize_network_with_format(context, format_type):
|
|
"""Visualize the network with the specified format."""
|
|
context.visualization_result = context.app.visualize_network(
|
|
output_format=format_type
|
|
)
|
|
|
|
|
|
@then('visualization returns "not supported" message')
|
|
def step_check_not_supported_message(context):
|
|
"""Check that visualization returns 'not supported' message."""
|
|
assert "not supported" in context.visualization_result.lower()
|
|
|
|
|
|
@then('visualization contains "graph TD"')
|
|
def step_check_contains_graph_td(context):
|
|
"""Check that visualization contains 'graph TD'."""
|
|
assert "graph TD" in context.visualization_result
|
|
|
|
|
|
@then("visualization contains agent references")
|
|
def step_check_contains_agent_references(context):
|
|
"""Check that visualization contains agent references."""
|
|
# Should contain at least one agent from our configuration
|
|
assert (
|
|
"analyzer" in context.visualization_result
|
|
or "Agent:" in context.visualization_result
|
|
)
|
|
|
|
|
|
@then("visualization includes all nodes")
|
|
def step_check_includes_all_nodes(context):
|
|
"""Check that visualization includes all nodes."""
|
|
# Should include agents
|
|
viz = context.visualization_result
|
|
assert "agent1" in viz or "agent2" in viz or "agent3" in viz or "Agent:" in viz
|
|
|
|
|
|
@then("visualization includes all edges")
|
|
def step_check_includes_all_edges(context):
|
|
"""Check that visualization includes all edges."""
|
|
# Should include arrow notation for edges
|
|
assert "-->" in context.visualization_result
|
|
|
|
|
|
@then("visualization is properly indented")
|
|
def step_check_properly_indented(context):
|
|
"""Check that visualization is properly indented."""
|
|
# Check that lines are indented (mermaid format uses indentation)
|
|
lines = context.visualization_result.split("\n")
|
|
# At least some lines should have indentation (4 spaces)
|
|
indented_lines = [line for line in lines if line.startswith(" ")]
|
|
assert len(indented_lines) > 0, "No indented lines found in visualization"
|