988 lines
33 KiB
Python
988 lines
33 KiB
Python
"""Route bridge coverage steps (ported and adapted)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined]
|
|
|
|
from cleveragents.core.exceptions import ConfigurationError
|
|
from cleveragents.langgraph.graph import GraphConfig
|
|
from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveragents.langgraph.state import GraphState
|
|
from cleveragents.reactive.route import BridgeConfig, RouteConfig, RouteType
|
|
from cleveragents.reactive.route_bridge import RouteBridge
|
|
from cleveragents.reactive.stream_router import (
|
|
ReactiveStreamRouter,
|
|
StreamMessage,
|
|
)
|
|
|
|
_DEF_CHECKPOINT_DIR = Path("/tmp/langgraph_checkpoints")
|
|
|
|
# Reuse a single event loop for async helpers in tests
|
|
_TEST_LOOP = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(_TEST_LOOP)
|
|
|
|
|
|
def _standard_topological_levels(nodes: dict[str, object]) -> dict[int, set[str]]:
|
|
middle_nodes = {name for name in nodes if name not in {"start", "end"}}
|
|
return {0: {"start"}, 1: middle_nodes, 2: {"end"}}
|
|
|
|
|
|
def _build_standard_langgraph_stub(corrupt_state_manager: bool = False):
|
|
start_cfg = NodeConfig(name="start", type=NodeType.START)
|
|
agent_node_cfg = NodeConfig(
|
|
name="agent_node", type=NodeType.AGENT, agent="dummy", parallel=False
|
|
)
|
|
function_node_cfg = NodeConfig(
|
|
name="function_node",
|
|
type=NodeType.FUNCTION,
|
|
function="stub_function",
|
|
parallel=False,
|
|
)
|
|
end_cfg = NodeConfig(name="end", type=NodeType.END)
|
|
|
|
def _stub_node(cfg):
|
|
return type("StubNode", (), {"config": cfg})()
|
|
|
|
nodes = {
|
|
"start": _stub_node(start_cfg),
|
|
"agent_node": _stub_node(agent_node_cfg),
|
|
"function_node": _stub_node(function_node_cfg),
|
|
"end": _stub_node(end_cfg),
|
|
}
|
|
|
|
edges = [
|
|
Edge(source="start", target="agent_node"),
|
|
Edge(source="agent_node", target="function_node"),
|
|
Edge(source="function_node", target="end"),
|
|
]
|
|
|
|
graph_config = GraphConfig(
|
|
name="stub_graph",
|
|
nodes={name: node.config for name, node in nodes.items()},
|
|
edges=edges,
|
|
entry_point="start",
|
|
checkpointing=True,
|
|
checkpoint_dir=_DEF_CHECKPOINT_DIR,
|
|
parallel_execution=False,
|
|
)
|
|
|
|
class StubStateManager:
|
|
def __init__(self, checkpoint_dir: Path):
|
|
self.checkpoint_dir = checkpoint_dir
|
|
self.update_count = 0
|
|
self.updated_state = None
|
|
self.update_state_called = False
|
|
self.checkpoint_saved = False
|
|
|
|
def _save_checkpoint(self): # pylint: disable=protected-access
|
|
self.checkpoint_saved = True
|
|
return None
|
|
|
|
def update_state(self, state=None):
|
|
self.update_state_called = True
|
|
self.updated_state = state or {}
|
|
self.update_count += 1
|
|
|
|
def get_state(self):
|
|
return self.updated_state or {}
|
|
|
|
langgraph = type("StubLangGraph", (), {})()
|
|
langgraph.nodes = nodes
|
|
langgraph.config = graph_config
|
|
langgraph.state_manager = (
|
|
None if corrupt_state_manager else StubStateManager(_DEF_CHECKPOINT_DIR)
|
|
)
|
|
langgraph._topological_levels = lambda: _standard_topological_levels(nodes) # type: ignore[attr-defined]
|
|
return langgraph
|
|
|
|
|
|
def _build_stub_langgraph_from_config(
|
|
graph_config: GraphConfig, checkpoint_dir: Path | None = None
|
|
):
|
|
class StubStateManager:
|
|
def __init__(self, ckp: Path | None):
|
|
self.checkpoint_dir = ckp
|
|
self.update_count = 0
|
|
self.update_state_called = False
|
|
self.updated_state = None
|
|
|
|
def update_state(self, state=None):
|
|
self.update_state_called = True
|
|
self.updated_state = state or {}
|
|
self.update_count += 1
|
|
|
|
def get_state(self):
|
|
return self.updated_state or {}
|
|
|
|
def _save_checkpoint(self): # pylint: disable=protected-access
|
|
return None
|
|
|
|
nodes = {
|
|
name: type("StubNode", (), {"config": cfg})()
|
|
for name, cfg in graph_config.nodes.items()
|
|
}
|
|
if "start" not in nodes:
|
|
nodes["start"] = type(
|
|
"StubNode", (), {"config": NodeConfig(name="start", type=NodeType.START)}
|
|
)()
|
|
if "end" not in nodes:
|
|
nodes["end"] = type(
|
|
"StubNode", (), {"config": NodeConfig(name="end", type=NodeType.END)}
|
|
)()
|
|
|
|
langgraph = type("StubLangGraph", (), {})()
|
|
langgraph.config = graph_config
|
|
langgraph.nodes = nodes
|
|
langgraph.state_manager = StubStateManager(
|
|
checkpoint_dir or graph_config.checkpoint_dir or _DEF_CHECKPOINT_DIR
|
|
)
|
|
langgraph._topological_levels = lambda: _standard_topological_levels(nodes) # type: ignore[attr-defined]
|
|
return langgraph
|
|
|
|
|
|
def _ensure_topological_levels(graph):
|
|
if not hasattr(graph, "_topological_levels"):
|
|
graph._topological_levels = lambda: _standard_topological_levels( # type: ignore[attr-defined]
|
|
getattr(graph, "nodes", {})
|
|
)
|
|
|
|
|
|
def _run_async(coro):
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_closed():
|
|
raise RuntimeError
|
|
if loop.is_running():
|
|
return asyncio.run(coro)
|
|
except Exception: # pylint: disable=broad-except
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
return loop.run_until_complete(coro)
|
|
|
|
|
|
@given("I have a clean test environment for route bridge")
|
|
def step_clean_env(context):
|
|
context.stream_router = ReactiveStreamRouter()
|
|
context.agents = {"agentA": DummyAgent("agentA")}
|
|
context.route_bridge = None
|
|
context.stream_message = None
|
|
context.langgraph = None
|
|
context.graph_result = None
|
|
context.stream_config_result = None
|
|
context.graph_config_result = None
|
|
context.route_bridge_error = None
|
|
context.state_flattener_called = False
|
|
|
|
|
|
class DummyAgent:
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
|
|
def process_message_sync(self, message, metadata=None):
|
|
return f"processed:{message}"
|
|
|
|
def get_capabilities(self):
|
|
return []
|
|
|
|
|
|
@given("I have a route bridge stream router and agents")
|
|
def step_stream_router_agents(context):
|
|
context.stream_router = ReactiveStreamRouter()
|
|
context.agents = {"agentA": DummyAgent("agentA"), "dummy": DummyAgent("dummy")}
|
|
for name, agent in context.agents.items():
|
|
context.stream_router.register_agent(name, agent)
|
|
|
|
|
|
@given("I have an AsyncIO scheduler")
|
|
def step_asyncio_scheduler(context):
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
context.scheduler = AsyncIOScheduler(loop)
|
|
|
|
|
|
@when("I create the route bridge instance")
|
|
def step_create_route_bridge(context):
|
|
context.route_bridge = RouteBridge(
|
|
context.stream_router,
|
|
context.agents,
|
|
scheduler=getattr(context, "scheduler", None),
|
|
)
|
|
|
|
|
|
@when("I create a RouteBridge instance with scheduler")
|
|
def step_create_route_bridge_scheduler(context):
|
|
step_create_route_bridge(context)
|
|
|
|
|
|
@then("the route bridge should be initialized correctly")
|
|
def step_rb_initialized(context):
|
|
assert context.route_bridge is not None
|
|
assert hasattr(context.route_bridge, "stream_router")
|
|
|
|
|
|
@then("the route bridge logger should be set up")
|
|
def step_logger_set(context):
|
|
assert context.route_bridge.logger is not None
|
|
|
|
|
|
@then("route bridge active conversions should be empty")
|
|
def step_active_conversions_empty(context):
|
|
assert context.route_bridge._active_conversions == {}
|
|
|
|
|
|
@then("the route bridge should be initialized with scheduler")
|
|
def step_scheduler_stored(context):
|
|
assert context.route_bridge.scheduler is not None
|
|
|
|
|
|
@then("the route bridge scheduler should be stored correctly")
|
|
def step_scheduler_correct(context):
|
|
assert context.route_bridge.scheduler == getattr(context, "scheduler", None)
|
|
|
|
|
|
@given("I have a RouteBridge instance")
|
|
def step_have_route_bridge(context):
|
|
if not getattr(context, "route_bridge", None):
|
|
step_create_route_bridge(context)
|
|
# Reset results for clarity
|
|
context.upgrade_result = False
|
|
context.downgrade_result = False
|
|
context.upgrade_error = None
|
|
context.downgrade_error = None
|
|
context.route_bridge._active_conversions.clear()
|
|
|
|
|
|
@given("I have a route config without bridge configuration")
|
|
def step_route_no_bridge(context):
|
|
context.route_config = RouteConfig(
|
|
name="r1",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"agent_node": {"type": "agent", "agent": "dummy", "parallel": False},
|
|
"function_node": {
|
|
"type": "function",
|
|
"function": "stub_function",
|
|
"parallel": False,
|
|
},
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "agent_node"},
|
|
{"source": "agent_node", "target": "function_node"},
|
|
{"source": "function_node", "target": "end"},
|
|
],
|
|
entry_point="start",
|
|
checkpointing=True,
|
|
checkpoint_dir=str(_DEF_CHECKPOINT_DIR),
|
|
)
|
|
context.downgrade_result = False
|
|
context.upgrade_result = False
|
|
|
|
|
|
@given("I have a graph route config with bridge configuration")
|
|
def step_graph_route_with_bridge(context):
|
|
context.route_config = RouteConfig(
|
|
name="g1",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"agent_node": {"type": "agent", "agent": "dummy", "parallel": False},
|
|
"function_node": {
|
|
"type": "function",
|
|
"function": "stub_function",
|
|
"parallel": False,
|
|
},
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "agent_node"},
|
|
{"source": "agent_node", "target": "function_node"},
|
|
{"source": "function_node", "target": "end"},
|
|
],
|
|
entry_point="start",
|
|
bridge=BridgeConfig(),
|
|
checkpointing=True,
|
|
checkpoint_dir=str(_DEF_CHECKPOINT_DIR),
|
|
parallel_execution=False,
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with needs_state upgrade condition")
|
|
def step_stream_route_needs_state(context):
|
|
context.route_config = RouteConfig(
|
|
name="needs_state",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"needs_state": True}),
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with message_count upgrade condition")
|
|
def step_stream_route_message_count(context):
|
|
context.route_config = RouteConfig(
|
|
name="message_count",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"message_count": 5}),
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with complexity_threshold upgrade condition")
|
|
def step_stream_route_complexity_threshold(context):
|
|
context.route_config = RouteConfig(
|
|
name="complexity_threshold",
|
|
type=RouteType.STREAM,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "dummy"}},
|
|
{"type": "map", "params": {"function": "f"}},
|
|
],
|
|
bridge=BridgeConfig(upgrade_conditions={"complexity_threshold": 6}),
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with custom_predicate upgrade condition")
|
|
def step_stream_route_custom_predicate(context):
|
|
context.route_config = RouteConfig(
|
|
name="custom_predicate",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"custom_predicate": None}),
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with non-callable custom_predicate")
|
|
def step_stream_route_non_callable_predicate(context):
|
|
context.route_config = RouteConfig(
|
|
name="non_callable_predicate",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"custom_predicate": "not_callable"}),
|
|
)
|
|
|
|
|
|
@given("I have a stream message")
|
|
def step_stream_message(context):
|
|
context.stream_message = StreamMessage(content="data", metadata={})
|
|
|
|
|
|
@given("I have a stream message with requires_state metadata")
|
|
def step_stream_message_requires_state(context):
|
|
context.stream_message = StreamMessage(
|
|
content="data", metadata={"requires_state": True}
|
|
)
|
|
|
|
|
|
@given("I have a graph route config for downgrade")
|
|
def step_graph_route_for_downgrade(context):
|
|
context.route_config = RouteConfig(
|
|
name="graph_to_stream",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"agent_node": {"type": "agent", "agent": "dummy", "parallel": False},
|
|
"function_node": {
|
|
"type": "function",
|
|
"function": "stub_function",
|
|
"parallel": False,
|
|
},
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "agent_node"},
|
|
{"source": "agent_node", "target": "function_node"},
|
|
{"source": "function_node", "target": "end"},
|
|
],
|
|
entry_point="start",
|
|
checkpointing=True,
|
|
checkpoint_dir=str(_DEF_CHECKPOINT_DIR),
|
|
parallel_execution=False,
|
|
bridge=BridgeConfig(),
|
|
)
|
|
|
|
|
|
@given("I have a LangGraph instance")
|
|
def step_have_langgraph_instance(context):
|
|
context.langgraph = _build_standard_langgraph_stub()
|
|
_ensure_topological_levels(context.langgraph)
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "graph",
|
|
"instance": context.langgraph,
|
|
"original_config": context.route_config,
|
|
}
|
|
|
|
|
|
@given("I have a graph state")
|
|
def step_graph_state(context):
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@given("I have a graph route config with idle_time downgrade condition")
|
|
def step_graph_route_idle_time(context):
|
|
bridge_config = BridgeConfig(downgrade_conditions={"idle_time": 5.0})
|
|
context.route_config = RouteConfig(
|
|
name="graph_idle",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state that has been idle too long")
|
|
def step_graph_state_idle(context):
|
|
context.graph_state = GraphState()
|
|
current_time = asyncio.get_event_loop().time()
|
|
context.graph_state.metadata = {"last_updated": current_time - 10.0}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@given("I have a graph route config with state_size downgrade condition")
|
|
def step_graph_route_state_size(context):
|
|
bridge_config = BridgeConfig(downgrade_conditions={"state_size": 1})
|
|
context.route_config = RouteConfig(
|
|
name="graph_state_size",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state with minimal messages")
|
|
def step_graph_state_minimal(context):
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = [{"content": "only_one"}]
|
|
|
|
|
|
@given("I have a graph route config with no_conditionals_used downgrade condition")
|
|
def step_graph_route_no_conditionals(context):
|
|
bridge_config = BridgeConfig(downgrade_conditions={"no_conditionals_used": True})
|
|
context.route_config = RouteConfig(
|
|
name="graph_no_conditionals",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state with no conditional edges used")
|
|
def step_graph_state_no_conditionals(context):
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {"conditional_edges_used": False}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@given("I have a stream route config with bridge configuration")
|
|
def step_stream_route_with_bridge_config(context):
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"needs_state": True},
|
|
downgrade_conditions={"idle_time": 30.0},
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="stream_with_bridge",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@when("I check downgrade conditions")
|
|
def step_check_downgrade(context):
|
|
if not getattr(context, "graph_state", None):
|
|
step_graph_state(context)
|
|
try:
|
|
context.downgrade_result = _run_async(
|
|
context.route_bridge.check_downgrade_conditions(
|
|
context.route_config, context.graph_state
|
|
)
|
|
)
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.downgrade_result = False
|
|
context.downgrade_error = exc
|
|
|
|
|
|
@then("the downgrade result should be True")
|
|
def step_downgrade_result_true(context):
|
|
assert context.downgrade_result is True
|
|
|
|
|
|
@then("the downgrade result should be False")
|
|
def step_downgrade_result_false(context):
|
|
assert context.downgrade_result is False
|
|
|
|
|
|
@given("I have processed enough messages to trigger upgrade")
|
|
def step_processed_enough_messages(context):
|
|
dummy_instance = type("DummyConversion", (), {"update_count": 5})()
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "stream",
|
|
"instance": dummy_instance,
|
|
"original_config": context.route_config,
|
|
}
|
|
|
|
|
|
@given("I have a complex route configuration")
|
|
def step_complex_route_configuration(context):
|
|
# Ensure the route has enough operators to exceed the threshold
|
|
if (
|
|
not getattr(context, "route_config", None)
|
|
or context.route_config.type != RouteType.STREAM
|
|
):
|
|
context.route_config = RouteConfig(
|
|
name="complex_route",
|
|
type=RouteType.STREAM,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "dummy"}},
|
|
{"type": "map", "params": {"function": "f1"}},
|
|
{"type": "map", "params": {"function": "f2"}},
|
|
{"type": "map", "params": {"function": "f3"}},
|
|
{"type": "map", "params": {"function": "f4"}},
|
|
],
|
|
bridge=BridgeConfig(upgrade_conditions={"complexity_threshold": 6}),
|
|
)
|
|
else:
|
|
# Enrich existing operators to make the route more complex
|
|
context.route_config.operators.extend(
|
|
[
|
|
{"type": "map", "params": {"function": "extra_f1"}},
|
|
{"type": "map", "params": {"function": "extra_f2"}},
|
|
{"type": "map", "params": {"function": "extra_f3"}},
|
|
]
|
|
)
|
|
|
|
|
|
@given("I seed a message count for the route")
|
|
def step_seed_message_count(context):
|
|
if not getattr(context, "route_config", None):
|
|
context.route_config = RouteConfig(
|
|
name="message_count_route",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"message_count": 5}),
|
|
)
|
|
state_manager = type("DummyStateManager", (), {"update_count": 7})()
|
|
dummy_graph = type("DummyGraph", (), {"state_manager": state_manager})()
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "graph",
|
|
"instance": dummy_graph,
|
|
"original_config": context.route_config,
|
|
}
|
|
|
|
|
|
@when("I get message count for a route")
|
|
def step_get_message_count(context):
|
|
if not getattr(context, "route_config", None):
|
|
step_seed_message_count(context)
|
|
context.message_count = context.route_bridge._get_message_count(
|
|
context.route_config.name
|
|
)
|
|
|
|
|
|
@when("I check upgrade conditions")
|
|
def step_check_upgrade(context):
|
|
if not getattr(context, "stream_message", None):
|
|
context.stream_message = StreamMessage(content="data", metadata={})
|
|
try:
|
|
context.upgrade_result = _run_async(
|
|
context.route_bridge.check_upgrade_conditions(
|
|
context.route_config, context.stream_message
|
|
)
|
|
)
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.upgrade_result = False
|
|
context.upgrade_error = exc
|
|
|
|
|
|
@when("I check upgrade conditions synchronously")
|
|
def step_check_upgrade_sync(context):
|
|
if not getattr(context, "stream_message", None):
|
|
context.stream_message = StreamMessage(content="data", metadata={})
|
|
coro = context.route_bridge.check_upgrade_conditions(
|
|
context.route_config, context.stream_message
|
|
)
|
|
context.upgrade_result = _run_async(coro)
|
|
|
|
|
|
@then("the route bridge upgrade result should be True")
|
|
def step_upgrade_result_true(context):
|
|
assert context.upgrade_result is True
|
|
|
|
|
|
@then("the route bridge upgrade result should be False")
|
|
def step_upgrade_result_false(context):
|
|
assert context.upgrade_result is False
|
|
|
|
|
|
@then("the route bridge message count should be at least 5")
|
|
def step_message_count_at_least(context):
|
|
assert context.route_bridge._get_message_count(context.route_config.name) >= 5
|
|
|
|
|
|
@then("the route bridge message count should be returned")
|
|
def step_message_count_returned(context):
|
|
assert context.message_count is not None
|
|
|
|
|
|
@then("the route bridge message count result should be a valid integer")
|
|
def step_message_count_valid_integer(context):
|
|
assert isinstance(context.message_count, int)
|
|
|
|
|
|
@then("route bridge error handling should occur")
|
|
def step_error_handling(context):
|
|
assert context.upgrade_error is None or isinstance(context.upgrade_error, Exception)
|
|
|
|
|
|
@then("the route bridge system should remain stable")
|
|
def step_system_stable(context):
|
|
assert context.route_bridge is not None
|
|
|
|
|
|
@given("I have an invalid graph route config")
|
|
def step_invalid_graph_route_config(context):
|
|
# Build an invalid placeholder but catch validation so test can proceed
|
|
try:
|
|
context.route_config = RouteConfig(
|
|
name="bad_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={},
|
|
edges=[],
|
|
)
|
|
except ConfigurationError as exc:
|
|
context.route_config = None
|
|
context.downgrade_error = exc
|
|
|
|
|
|
@given("I have a corrupted LangGraph instance")
|
|
def step_corrupted_langgraph(context):
|
|
context.langgraph = _build_standard_langgraph_stub(corrupt_state_manager=True)
|
|
|
|
|
|
@when("I attempt to downgrade graph to stream")
|
|
def step_attempt_downgrade(context):
|
|
if context.route_config is None:
|
|
context.downgrade_result = False
|
|
context.downgrade_error = context.downgrade_error or ConfigurationError(
|
|
"Invalid graph route"
|
|
)
|
|
return
|
|
try:
|
|
context.downgrade_result = asyncio.get_event_loop().run_until_complete(
|
|
context.route_bridge.downgrade_graph_to_stream(
|
|
context.route_config, context.langgraph
|
|
)
|
|
)
|
|
context.downgrade_error = None
|
|
except ConfigurationError as exc:
|
|
context.downgrade_result = False
|
|
context.downgrade_error = exc
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.downgrade_error = exc
|
|
|
|
|
|
@then("And the system should remain stable")
|
|
def step_system_stable_downgrade(context):
|
|
assert context.route_bridge is not None
|
|
|
|
|
|
@given("I have a route config that supports both upgrade and downgrade")
|
|
def step_route_upgrade_downgrade(context):
|
|
context.route_config = RouteConfig(
|
|
name="cycle",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(
|
|
upgrade_conditions={"needs_state": True},
|
|
downgrade_conditions={"state_size": 0},
|
|
),
|
|
)
|
|
context.stream_message = StreamMessage(
|
|
content="x", metadata={"requires_state": True}
|
|
)
|
|
|
|
|
|
@when("I perform an upgrade and then downgrade cycle")
|
|
def step_cycle_upgrade_downgrade(context):
|
|
context.langgraph = asyncio.get_event_loop().run_until_complete(
|
|
context.route_bridge.upgrade_stream_to_graph(
|
|
context.route_config, context.stream_message
|
|
)
|
|
)
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "graph",
|
|
"instance": context.langgraph,
|
|
"original_config": context.route_config,
|
|
}
|
|
_ensure_topological_levels(context.langgraph)
|
|
# Convert Node objects to NodeConfig dict
|
|
node_configs = {}
|
|
for name, node in context.langgraph.nodes.items():
|
|
if hasattr(node, "config") and node.config:
|
|
node_configs[name] = node.config
|
|
else:
|
|
# Create a minimal NodeConfig for nodes without config
|
|
node_configs[name] = NodeConfig(
|
|
name=name, type=getattr(node, "type", NodeType.FUNCTION)
|
|
)
|
|
|
|
context.stream_config = asyncio.get_event_loop().run_until_complete(
|
|
context.route_bridge.downgrade_graph_to_stream(
|
|
RouteConfig.from_graph_config(
|
|
GraphConfig(
|
|
name="cycle_g",
|
|
nodes=node_configs,
|
|
edges=context.langgraph.config.edges,
|
|
entry_point="start",
|
|
)
|
|
),
|
|
context.langgraph,
|
|
)
|
|
)
|
|
|
|
|
|
@then("the cycle should complete successfully")
|
|
def step_cycle_success(context):
|
|
assert context.stream_config is not None
|
|
|
|
|
|
@then("the final configuration should be consistent")
|
|
def step_final_config_consistent(context):
|
|
assert context.stream_config.name
|
|
|
|
|
|
@then("active conversions should be tracked correctly")
|
|
def step_active_conversions_tracked(context):
|
|
assert context.route_bridge._active_conversions
|
|
|
|
|
|
# Additional coverage steps
|
|
|
|
|
|
class _FailingRegistry:
|
|
def list_actors(self):
|
|
raise RuntimeError("registry failure")
|
|
|
|
|
|
@given("I have a failing agent registry for route bridge")
|
|
def step_failing_registry(context):
|
|
context.stream_router = ReactiveStreamRouter()
|
|
context.agents = _FailingRegistry()
|
|
|
|
|
|
@then("the route bridge agents should be empty")
|
|
def step_agents_empty(context):
|
|
assert context.route_bridge.agents == {}
|
|
|
|
|
|
@given("I have a pending message count below threshold")
|
|
def step_message_count_below_threshold(context):
|
|
dummy_instance = type("DummyConversion", (), {"update_count": 2})()
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "stream",
|
|
"instance": dummy_instance,
|
|
"original_config": context.route_config,
|
|
}
|
|
|
|
|
|
@given("I have a simple stream route config with high complexity threshold")
|
|
def step_simple_route_high_complexity(context):
|
|
context.route_config = RouteConfig(
|
|
name="simple_high_threshold",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"complexity_threshold": 999}),
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with callable custom_predicate")
|
|
def step_route_callable_predicate(context):
|
|
def _predicate(message, _cfg):
|
|
return True
|
|
|
|
bridge_config = BridgeConfig(upgrade_conditions={"custom_predicate": None})
|
|
object.__setattr__(
|
|
bridge_config, "upgrade_conditions", {"custom_predicate": _predicate}
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="callable_predicate",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with state extractor and subscriptions")
|
|
def step_route_with_state_extractor(context):
|
|
def _extractor(_msg):
|
|
return {"metadata": {"extracted": True}}
|
|
|
|
bridge_config = BridgeConfig(preserve_subscriptions=True)
|
|
object.__setattr__(bridge_config, "state_extractor", _extractor)
|
|
|
|
context.route_config = RouteConfig(
|
|
name="extractor_route",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
subscriptions=["a", "b"],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@when("I create a graph config from a stream without agents")
|
|
def step_create_graph_config_without_agents(context):
|
|
context.route_config = RouteConfig(
|
|
name="function_only_stream",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"function": "noop"}}],
|
|
)
|
|
context.graph_config_result = context.route_bridge._create_graph_from_stream(
|
|
context.route_config
|
|
)
|
|
|
|
|
|
@given("I seed a fallback update_count conversion")
|
|
def step_seed_fallback_update_count(context):
|
|
context.route_config = RouteConfig(
|
|
name="update_count_route",
|
|
type=RouteType.STREAM,
|
|
operators=[{"type": "map", "params": {"agent": "dummy"}}],
|
|
bridge=BridgeConfig(upgrade_conditions={"message_count": 5}),
|
|
)
|
|
instance = type("DummyConversion", (), {"update_count": 3})()
|
|
context.route_bridge._active_conversions[context.route_config.name] = {
|
|
"type": "graph",
|
|
"instance": instance,
|
|
"original_config": context.route_config,
|
|
}
|
|
|
|
|
|
@when("I upgrade the stream route to graph")
|
|
def step_upgrade_stream_with_extractor(context):
|
|
if not getattr(context, "stream_message", None):
|
|
context.stream_message = StreamMessage(content="data", metadata={})
|
|
context.graph_result = _run_async(
|
|
context.route_bridge.upgrade_stream_to_graph(
|
|
context.route_config, context.stream_message
|
|
)
|
|
)
|
|
|
|
|
|
@given("I have a graph route config with flattener and checkpointing")
|
|
def step_graph_route_with_flattener(context):
|
|
def _flattener(state):
|
|
context.state_flattener_called = True
|
|
return state
|
|
|
|
bridge_config = BridgeConfig(preserve_checkpointing=True)
|
|
object.__setattr__(bridge_config, "state_flattener", _flattener)
|
|
|
|
context.route_config = RouteConfig(
|
|
name="graph_flat",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
checkpointing=True,
|
|
checkpoint_dir=str(_DEF_CHECKPOINT_DIR),
|
|
parallel_execution=False,
|
|
)
|
|
|
|
|
|
@given("I have a LangGraph instance with checkpointing")
|
|
def step_langgraph_with_checkpointing(context):
|
|
context.langgraph = _build_standard_langgraph_stub()
|
|
# ensure checkpoint_dir persists for test assertions
|
|
if hasattr(context.langgraph, "state_manager"):
|
|
context.langgraph.state_manager.checkpoint_dir = _DEF_CHECKPOINT_DIR
|
|
|
|
|
|
@when("I downgrade the graph with flattener")
|
|
def step_downgrade_with_flattener(context):
|
|
context.stream_config_result = _run_async(
|
|
context.route_bridge.downgrade_graph_to_stream(
|
|
context.route_config, context.langgraph
|
|
)
|
|
)
|
|
|
|
|
|
@then("the state flattener should run")
|
|
def step_state_flattener_run(context):
|
|
assert context.state_flattener_called is True
|
|
|
|
|
|
@then("the checkpoint should be saved")
|
|
def step_checkpoint_saved_flag(context):
|
|
assert getattr(context.langgraph.state_manager, "checkpoint_saved", False) is True
|
|
|
|
|
|
@then("the graph initial state should include extracted data")
|
|
def step_graph_state_includes_extracted(context):
|
|
state_mgr = getattr(context.graph_result, "state_manager", None)
|
|
assert state_mgr is not None
|
|
state = state_mgr.get_state()
|
|
assert isinstance(state, GraphState)
|
|
assert state.metadata.get("extracted") is True
|
|
|
|
|
|
@given("I have a graph state with multiple messages")
|
|
def step_graph_state_multiple_messages(context):
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = [
|
|
{"content": "a"},
|
|
{"content": "b"},
|
|
{"content": "c"},
|
|
]
|
|
|
|
|
|
@given("I have a graph route config with idle_time not exceeded")
|
|
def step_graph_route_idle_time_not_exceeded(context):
|
|
bridge_config = BridgeConfig(downgrade_conditions={"idle_time": 1000.0})
|
|
context.route_config = RouteConfig(
|
|
name="graph_idle_safe",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
)
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {"last_updated": _TEST_LOOP.time()}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@then("the route bridge fallback message count should be returned")
|
|
def step_fallback_message_count(context):
|
|
assert context.route_bridge._get_message_count(context.route_config.name) == 3
|
|
|
|
|
|
@given("I have a graph route config with large state size")
|
|
def step_graph_route_large_state(context):
|
|
bridge_config = BridgeConfig(downgrade_conditions={"state_size": 1})
|
|
context.route_config = RouteConfig(
|
|
name="graph_state_large",
|
|
type=RouteType.GRAPH,
|
|
nodes={"n1": {"type": "agent", "agent": "dummy"}},
|
|
edges=[{"source": "start", "target": "n1"}],
|
|
bridge=bridge_config,
|
|
)
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = [
|
|
{"content": "a"},
|
|
{"content": "b"},
|
|
{"content": "c"},
|
|
]
|
|
|
|
|
|
@then("the resulting graph should use function nodes")
|
|
def step_result_graph_function_nodes(context):
|
|
node_cfg = context.graph_config_result.nodes.get("op_0_map")
|
|
assert node_cfg is not None
|
|
assert node_cfg.type == NodeType.FUNCTION
|