79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
"""Behave steps for routing_adapter coverage."""
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.langgraph.nodes import NodeType
|
|
from cleveragents.langgraph.routing_adapter import build_route_nodes_from_stream
|
|
|
|
|
|
@given("route operators include a map agent and function")
|
|
def step_setup_map_agent_and_function(context):
|
|
context.route_ops = [
|
|
{"type": "map", "params": {"agent": "agent_alpha"}},
|
|
{"type": "custom_fn"},
|
|
]
|
|
|
|
|
|
@given("route operators include a map without agent")
|
|
def step_setup_map_without_agent(context):
|
|
context.route_ops = [
|
|
{"type": "map", "params": {"other": "value"}},
|
|
]
|
|
|
|
|
|
@when("I build route nodes from that stream")
|
|
def step_build_route_nodes(context):
|
|
context.nodes, context.edges = build_route_nodes_from_stream(
|
|
"test_route", context.route_ops
|
|
)
|
|
|
|
|
|
@then("it creates an agent node and chained edges to end")
|
|
def step_assert_agent_nodes(context):
|
|
nodes = context.nodes
|
|
edges = context.edges
|
|
|
|
assert nodes["start"].type is NodeType.START
|
|
assert nodes["end"].type is NodeType.END
|
|
|
|
agent_node = nodes.get("op_0_map")
|
|
assert agent_node is not None
|
|
assert agent_node.type is NodeType.AGENT
|
|
assert agent_node.agent == "agent_alpha"
|
|
|
|
func_node = nodes.get("op_1_custom_fn")
|
|
assert func_node is not None
|
|
assert func_node.type is NodeType.FUNCTION
|
|
assert func_node.function == "custom_fn"
|
|
|
|
edge_pairs = [(e.source, e.target) for e in edges]
|
|
assert edge_pairs == [
|
|
("start", "op_0_map"),
|
|
("op_0_map", "op_1_custom_fn"),
|
|
("op_1_custom_fn", "end"),
|
|
]
|
|
|
|
|
|
@then("the function node is created with its type")
|
|
def step_assert_function_type(context):
|
|
func_node = context.nodes.get("op_1_custom_fn")
|
|
assert func_node.function == "custom_fn"
|
|
assert func_node.type is NodeType.FUNCTION
|
|
|
|
|
|
@then("it treats the map as function and terminates chain with end node")
|
|
def step_assert_map_falls_back_to_function(context):
|
|
nodes = context.nodes
|
|
edges = context.edges
|
|
|
|
map_node = nodes.get("op_0_map")
|
|
assert map_node is not None
|
|
assert map_node.type is NodeType.FUNCTION
|
|
assert map_node.function == "map"
|
|
|
|
assert nodes["start"].type is NodeType.START
|
|
assert nodes["end"].type is NodeType.END
|
|
|
|
edge_pairs = [(e.source, e.target) for e in edges]
|
|
assert edge_pairs == [("start", "op_0_map"), ("op_0_map", "end")]
|