CoreRasurae
9753b31c7e
CI / quality (push) Successful in 36s
CI / lint (push) Successful in 38s
CI / typecheck (push) Successful in 49s
CI / security (push) Successful in 51s
CI / integration_tests (push) Successful in 55s
CI / unit_tests (push) Successful in 3m35s
CI / build (push) Successful in 33s
CI / coverage (push) Successful in 3m36s
CI / status-check (push) Successful in 3s
Add 13 BDD scenarios covering previously uncovered code paths: - SAFE_BUILTINS validation (sandbox.py: 0% → 100%) - ConfigurationError re-raise path (config.py: 99.3% → 100%) - CLI hello/main functions (cli.py: 72.7% → 90.9%) - GraphState message truncation (state.py: 98.7% → 100%) - ProgressBarManager update/context rendering (progress.py: 0% → 87.7%) - MessageRouter regex/exact/contains routing (message_router.py: 0% → 59.4%) - RoutingAdapter parse_routing_command (routing_adapter.py) - DynamicRouterNode pattern-based routing (dynamic_router.py) - EnhancedTemplateRegistry unknown template type (enhanced_registry.py: 99.2%) - CompositeAgent null-graph error path (composite.py: 97.8% → 98.6%) Cover routing_adapter.py (17% → 100%): all GOTO/ROUTE patterns, create_routing_node, create_conditional_router, dynamic config conversion. Cover dynamic_router.py (21% → 87%): execute with empty/dict/string messages, extract_message with colon parsing, config creation, graph extension with edge generation. Cover message_router.py (59% → 78%): regex, exact, contains, prefix, suffix match types, invalid regex handling, non-string message, set_state. Exercise Node._prepare_conversation_history with invalid configs, _runtime error paths, _execute_message_router with rules, _execute_agent with current_message and metadata propagation, _execute_function with dynamic_router, and _execute_conditional with content_contains/content_not_contains/content_starts_with and custom condition types. Improves nodes.py from 71.3% to 72.5%. Exercise PureGraphConfig, PureLangGraph init with dict/config, RxPyLangGraphBridge registration/connection/lookup, ReactiveStreamRouter operator creation and condition functions, ReactiveConfigParser config/route/graph parsing, ToolAgent tool execution with JSON, space-separated, single, file_read, progress_bar invocations, and ReactiveCleverAgentsApp init/dispose/visualization.
1165 lines
39 KiB
Python
1165 lines
39 KiB
Python
"""
|
|
Step definitions for Route Upgrade and Downgrade Conversion BDD tests.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
from unittest.mock import Mock
|
|
|
|
from behave import given, then, when
|
|
from behave.api.async_step import async_run_until_complete
|
|
from rx.scheduler.eventloop import AsyncIOScheduler
|
|
|
|
from cleveractors.agents.base import Agent
|
|
from cleveractors.langgraph.graph import GraphConfig
|
|
from cleveractors.langgraph.nodes import NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
from cleveractors.reactive.route import BridgeConfig, RouteConfig, RouteType
|
|
from cleveractors.reactive.route_bridge import RouteBridge
|
|
from cleveractors.reactive.stream_router import (
|
|
ReactiveStreamRouter,
|
|
StreamConfig,
|
|
StreamMessage,
|
|
StreamType,
|
|
)
|
|
|
|
|
|
# Mock implementations for testing
|
|
class MockAgent(Agent):
|
|
"""Mock agent for testing."""
|
|
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
self.config = {}
|
|
# Initialize Agent with required parameters
|
|
super().__init__(name, {}, Mock())
|
|
|
|
async def process_message(self, message: str, context=None) -> str:
|
|
return f"processed_{message}_by_{self.name}"
|
|
|
|
def get_capabilities(self) -> List[str]:
|
|
return ["test_capability"]
|
|
|
|
|
|
class MockRouteComplexityAnalyzer:
|
|
"""Mock complexity analyzer."""
|
|
|
|
@staticmethod
|
|
def analyze_route(route_config: RouteConfig) -> Dict[str, Any]:
|
|
# Return high complexity for routes with many operators
|
|
operator_count = len(route_config.operators)
|
|
return {"score": operator_count * 10}
|
|
|
|
|
|
class MockGraph:
|
|
"""Mock LangGraph for testing."""
|
|
|
|
def __init__(self, config, **kwargs):
|
|
self.config = config
|
|
self.nodes = {}
|
|
self.state_manager = Mock()
|
|
self.state_manager.get_state.return_value = {"test": "state"}
|
|
self.state_manager.checkpoint_dir = None
|
|
self.state_manager._save_checkpoint = Mock()
|
|
self.state_manager.update_state = Mock()
|
|
|
|
# Create mock nodes based on config
|
|
if hasattr(config, "nodes"):
|
|
for node_name, node_config in config.nodes.items():
|
|
mock_node = Mock()
|
|
mock_node.config = node_config
|
|
self.nodes[node_name] = mock_node
|
|
|
|
def _topological_levels(self):
|
|
"""Mock topological sorting."""
|
|
return {
|
|
0: ["start"],
|
|
1: ["op_0_map", "op_1_filter"],
|
|
2: ["end"],
|
|
}
|
|
|
|
|
|
@given("I have a clean test environment for route bridge")
|
|
def step_clean_environment_route_bridge(context):
|
|
"""Set up clean test environment for route bridge."""
|
|
context.route_bridge = None
|
|
context.stream_router = None
|
|
context.agents = {}
|
|
context.scheduler = None
|
|
context.route_config = None
|
|
context.stream_message = None
|
|
context.graph_state = None
|
|
context.result = None
|
|
context.error = None
|
|
context.mock_graph = None
|
|
context.active_conversions = {}
|
|
context.mock_analyzer = None
|
|
context.original_analyzer = None
|
|
|
|
|
|
@given("I have a stream router and agents")
|
|
def step_stream_router_and_agents(context):
|
|
"""Create stream router and agents."""
|
|
context.stream_router = Mock(spec=ReactiveStreamRouter)
|
|
context.agents = {
|
|
"test_agent": MockAgent("test_agent"),
|
|
"another_agent": MockAgent("another_agent"),
|
|
}
|
|
|
|
|
|
@given("I have an AsyncIO scheduler")
|
|
def step_asyncio_scheduler(context):
|
|
"""Create AsyncIO scheduler."""
|
|
context.scheduler = Mock(spec=AsyncIOScheduler)
|
|
|
|
|
|
@when("I create a RouteBridge instance")
|
|
def step_create_route_bridge(context):
|
|
"""Create RouteBridge instance."""
|
|
try:
|
|
context.route_bridge = RouteBridge(
|
|
stream_router=context.stream_router, agents=context.agents
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I create a RouteBridge instance with scheduler")
|
|
def step_create_route_bridge_with_scheduler(context):
|
|
"""Create RouteBridge instance with scheduler."""
|
|
try:
|
|
context.route_bridge = RouteBridge(
|
|
stream_router=context.stream_router,
|
|
agents=context.agents,
|
|
scheduler=context.scheduler,
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the route bridge should be initialized correctly")
|
|
def step_verify_route_bridge_init(context):
|
|
"""Verify route bridge initialization."""
|
|
assert context.error is None
|
|
assert context.route_bridge is not None
|
|
assert context.route_bridge.stream_router == context.stream_router
|
|
assert context.route_bridge.agents == context.agents
|
|
assert context.route_bridge.scheduler is None
|
|
|
|
|
|
@then("the route bridge should be initialized with scheduler")
|
|
def step_verify_route_bridge_init_with_scheduler(context):
|
|
"""Verify route bridge initialization with scheduler."""
|
|
assert context.error is None
|
|
assert context.route_bridge is not None
|
|
assert context.route_bridge.scheduler == context.scheduler
|
|
|
|
|
|
@then("the logger should be set up")
|
|
def step_verify_logger_setup(context):
|
|
"""Verify logger setup."""
|
|
assert context.route_bridge.logger is not None
|
|
assert isinstance(context.route_bridge.logger, logging.Logger)
|
|
|
|
|
|
@then("active conversions should be empty")
|
|
def step_verify_empty_conversions(context):
|
|
"""Verify active conversions are empty."""
|
|
assert context.route_bridge._active_conversions == {}
|
|
|
|
|
|
@then("the scheduler should be stored correctly")
|
|
def step_verify_scheduler_stored(context):
|
|
"""Verify scheduler is stored correctly."""
|
|
assert context.route_bridge.scheduler == context.scheduler
|
|
|
|
|
|
@given("I have a graph route config")
|
|
def step_graph_route_config(context):
|
|
"""Create basic graph route config."""
|
|
context.route_config = RouteConfig(
|
|
name="basic_graph_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"start": {"type": "start"},
|
|
"process": {"type": "agent", "agent": "test_agent"},
|
|
"end": {"type": "end"},
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "process"},
|
|
{"source": "process", "target": "end"},
|
|
],
|
|
)
|
|
|
|
|
|
@given("I have a RouteBridge instance")
|
|
def step_create_default_route_bridge(context):
|
|
"""Create default RouteBridge instance."""
|
|
if not hasattr(context, "stream_router") or context.stream_router is None:
|
|
context.stream_router = Mock(spec=ReactiveStreamRouter)
|
|
if not hasattr(context, "agents") or context.agents is None:
|
|
context.agents = {"test_agent": MockAgent("test_agent")}
|
|
|
|
context.route_bridge = RouteBridge(
|
|
stream_router=context.stream_router, agents=context.agents
|
|
)
|
|
|
|
|
|
@given("I have a route config without bridge configuration")
|
|
def step_route_config_no_bridge(context):
|
|
"""Create route config without bridge."""
|
|
context.route_config = RouteConfig(
|
|
name="test_route", type=RouteType.STREAM, stream_type=StreamType.COLD
|
|
)
|
|
|
|
|
|
@given("I have a stream message")
|
|
def step_stream_message(context):
|
|
"""Create stream message."""
|
|
context.stream_message = StreamMessage(
|
|
content="test_data", metadata={}, timestamp=1234567890.0
|
|
)
|
|
|
|
|
|
@when("I check upgrade conditions")
|
|
@async_run_until_complete
|
|
async def step_check_upgrade_conditions(context):
|
|
"""Check upgrade conditions."""
|
|
try:
|
|
# Apply temporary mock if needed for complexity threshold test
|
|
if hasattr(context, "mock_analyzer") and context.mock_analyzer:
|
|
import cleveractors.reactive.route
|
|
|
|
original = cleveractors.reactive.route.RouteComplexityAnalyzer
|
|
cleveractors.reactive.route.RouteComplexityAnalyzer = context.mock_analyzer
|
|
|
|
try:
|
|
context.result = await context.route_bridge.check_upgrade_conditions(
|
|
context.route_config, context.stream_message
|
|
)
|
|
finally:
|
|
# Restore immediately
|
|
cleveractors.reactive.route.RouteComplexityAnalyzer = original
|
|
else:
|
|
context.result = await context.route_bridge.check_upgrade_conditions(
|
|
context.route_config, context.stream_message
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the result should be False")
|
|
def step_verify_result_false(context):
|
|
"""Verify result is False."""
|
|
assert context.error is None
|
|
assert context.result is False
|
|
|
|
|
|
@then("the result should be True")
|
|
def step_verify_result_true(context):
|
|
"""Verify result is True."""
|
|
assert context.error is None
|
|
assert context.result is True
|
|
|
|
|
|
@given("I have a graph route config with bridge configuration")
|
|
def step_graph_route_config_with_bridge(context):
|
|
"""Create graph route config with bridge."""
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"needs_state": True},
|
|
downgrade_conditions={"idle_time": 30.0},
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="test_graph_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with needs_state upgrade condition")
|
|
def step_stream_route_needs_state(context):
|
|
"""Create stream route with needs_state condition."""
|
|
bridge_config = BridgeConfig(upgrade_conditions={"needs_state": True})
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a stream message with requires_state metadata")
|
|
def step_stream_message_requires_state(context):
|
|
"""Create stream message with requires_state metadata."""
|
|
context.stream_message = StreamMessage(
|
|
content="test_data", metadata={"requires_state": True}, timestamp=1234567890.0
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with message_count upgrade condition")
|
|
def step_stream_route_message_count(context):
|
|
"""Create stream route with message_count condition."""
|
|
bridge_config = BridgeConfig(upgrade_conditions={"message_count": 5})
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have processed enough messages to trigger upgrade")
|
|
def step_processed_enough_messages(context):
|
|
"""Mock message count to trigger upgrade."""
|
|
# Mock the _get_message_count method to return high count
|
|
context.route_bridge._get_message_count = Mock(return_value=10)
|
|
|
|
|
|
@given("I have a stream route config with complexity_threshold upgrade condition")
|
|
def step_stream_route_complexity_threshold(context):
|
|
"""Create stream route with complexity_threshold condition."""
|
|
# Set up isolated mock for this test
|
|
import cleveractors.reactive.route_bridge
|
|
|
|
context.original_analyzer = getattr(
|
|
cleveractors.reactive.route_bridge, "RouteComplexityAnalyzer", None
|
|
)
|
|
context.mock_analyzer = MockRouteComplexityAnalyzer
|
|
|
|
bridge_config = BridgeConfig(upgrade_conditions={"complexity_threshold": 50})
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "test_agent"}},
|
|
{"type": "filter", "params": {"condition": "test"}},
|
|
{"type": "buffer", "params": {"count": 3}},
|
|
{"type": "debounce", "params": {"duration": 1.0}},
|
|
{"type": "throttle", "params": {"interval": 0.5}},
|
|
{"type": "distinct", "params": {}},
|
|
],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a complex route configuration")
|
|
def step_complex_route_configuration(context):
|
|
"""Ensure complex route configuration exists."""
|
|
assert context.route_config is not None, "Route config should be set"
|
|
assert context.route_config.name is not None, (
|
|
f"Route config should have a name, got {context.route_config.name}"
|
|
)
|
|
|
|
|
|
@given("I have a stream route config with custom_predicate upgrade condition")
|
|
def step_stream_route_custom_predicate(context):
|
|
"""Create stream route with custom_predicate condition."""
|
|
context.custom_predicate = Mock(return_value=True)
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"custom_predicate": context.custom_predicate}
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a custom predicate that returns True")
|
|
def step_custom_predicate_true(context):
|
|
"""Set custom predicate to return True."""
|
|
context.custom_predicate.return_value = True
|
|
|
|
|
|
@given("I have a custom predicate that returns False")
|
|
def step_custom_predicate_false(context):
|
|
"""Set custom predicate to return False."""
|
|
context.custom_predicate.return_value = False
|
|
|
|
|
|
@given("I have a stream route config with non-callable custom_predicate")
|
|
def step_stream_route_non_callable_predicate(context):
|
|
"""Create stream route with non-callable custom_predicate."""
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"custom_predicate": "not_callable"}
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state")
|
|
def step_graph_state(context):
|
|
"""Create graph state."""
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@when("I check downgrade conditions")
|
|
@async_run_until_complete
|
|
async def step_check_downgrade_conditions(context):
|
|
"""Check downgrade conditions."""
|
|
try:
|
|
context.result = await context.route_bridge.check_downgrade_conditions(
|
|
context.route_config, context.graph_state
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@given("I have a stream route config with bridge configuration")
|
|
def step_stream_route_config_with_bridge(context):
|
|
"""Create stream route config with bridge."""
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"needs_state": True},
|
|
downgrade_conditions={"idle_time": 30.0},
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="test_stream_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph route config with idle_time downgrade condition")
|
|
def step_graph_route_idle_time(context):
|
|
"""Create graph route with idle_time condition."""
|
|
bridge_config = BridgeConfig(downgrade_conditions={"idle_time": 5.0})
|
|
context.route_config = RouteConfig(
|
|
name="test_graph_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state that has been idle too long")
|
|
def step_graph_state_idle(context):
|
|
"""Create idle graph state."""
|
|
context.graph_state = GraphState()
|
|
# Set last_updated to be longer than idle_time threshold
|
|
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):
|
|
"""Create graph route with state_size condition."""
|
|
bridge_config = BridgeConfig(downgrade_conditions={"state_size": 2})
|
|
context.route_config = RouteConfig(
|
|
name="test_graph_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state with minimal messages")
|
|
def step_graph_state_minimal(context):
|
|
"""Create graph state with minimal messages."""
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {}
|
|
context.graph_state.messages = ["single_message"] # Less than threshold of 2
|
|
|
|
|
|
@given("I have a graph route config with no_conditionals_used downgrade condition")
|
|
def step_graph_route_no_conditionals(context):
|
|
"""Create graph route with no_conditionals_used condition."""
|
|
bridge_config = BridgeConfig(downgrade_conditions={"no_conditionals_used": True})
|
|
context.route_config = RouteConfig(
|
|
name="test_graph_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a graph state with no conditional edges used")
|
|
def step_graph_state_no_conditionals(context):
|
|
"""Create graph state with no conditional edges used."""
|
|
context.graph_state = GraphState()
|
|
context.graph_state.metadata = {"conditional_edges_used": False}
|
|
context.graph_state.messages = []
|
|
|
|
|
|
@given("I have a stream route config for upgrade")
|
|
def step_stream_route_for_upgrade(context):
|
|
"""Create stream route config for upgrade."""
|
|
context.route_config = RouteConfig(
|
|
name="upgrade_test_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "test_agent"}},
|
|
{"type": "filter", "params": {"condition": "test"}},
|
|
],
|
|
subscriptions=["input_stream"],
|
|
publications=["output_stream"],
|
|
)
|
|
|
|
|
|
@when("I upgrade stream to graph")
|
|
@async_run_until_complete
|
|
async def step_upgrade_stream_to_graph(context):
|
|
"""Upgrade stream to graph."""
|
|
try:
|
|
# Mock LangGraph to avoid complex dependencies
|
|
import cleveractors.reactive.route_bridge
|
|
|
|
original_langgraph = cleveractors.reactive.route_bridge.LangGraph
|
|
cleveractors.reactive.route_bridge.LangGraph = MockGraph
|
|
|
|
context.result = await context.route_bridge.upgrade_stream_to_graph(
|
|
context.route_config, context.stream_message
|
|
)
|
|
|
|
# Restore original
|
|
cleveractors.reactive.route_bridge.LangGraph = original_langgraph
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("a LangGraph should be created")
|
|
def step_verify_langgraph_created(context):
|
|
"""Verify LangGraph was created."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, MockGraph)
|
|
|
|
|
|
@then("the graph should be configured correctly")
|
|
def step_verify_graph_configured(context):
|
|
"""Verify graph configuration."""
|
|
graph = context.result
|
|
assert graph.config is not None
|
|
assert graph.config.name == "upgrade_test_route_graph"
|
|
|
|
|
|
@then("active conversions should be updated")
|
|
def step_verify_active_conversions_updated(context):
|
|
"""Verify active conversions are updated."""
|
|
conversions = context.route_bridge._active_conversions
|
|
route_name = context.route_config.name
|
|
assert route_name in conversions
|
|
# Type can be either "graph" or "stream" depending on operation
|
|
assert "type" in conversions[route_name]
|
|
|
|
|
|
@given("I have a stream route config with state extractor")
|
|
def step_stream_route_state_extractor(context):
|
|
"""Create stream route with state extractor."""
|
|
context.state_extractor = Mock(return_value={"extracted": "state"})
|
|
bridge_config = BridgeConfig(state_extractor=context.state_extractor)
|
|
context.route_config = RouteConfig(
|
|
name="extractor_test_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[{"type": "map", "params": {"agent": "test_agent"}}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@then("a LangGraph should be created with initial state")
|
|
def step_verify_langgraph_with_state(context):
|
|
"""Verify LangGraph was created with initial state."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
# Verify state_manager.update_state was called
|
|
context.result.state_manager.update_state.assert_called_once()
|
|
|
|
|
|
@then("the state should be extracted correctly")
|
|
def step_verify_state_extracted(context):
|
|
"""Verify state extraction."""
|
|
context.state_extractor.assert_called_once_with(context.stream_message)
|
|
|
|
|
|
@given("I have a stream route config with preserve subscriptions")
|
|
def step_stream_route_preserve_subscriptions(context):
|
|
"""Create stream route with preserve subscriptions."""
|
|
bridge_config = BridgeConfig(preserve_subscriptions=True)
|
|
context.route_config = RouteConfig(
|
|
name="preserve_subs_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[{"type": "map", "params": {"agent": "test_agent"}}],
|
|
subscriptions=["input1", "input2"],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@then("subscriptions should be preserved")
|
|
def step_verify_subscriptions_preserved(context):
|
|
"""Verify subscriptions are preserved."""
|
|
# This is a placeholder - the actual implementation would need to verify
|
|
# that subscriptions are handled properly
|
|
assert len(context.route_config.subscriptions) == 2
|
|
|
|
|
|
@given("I have a graph route config for downgrade")
|
|
def step_graph_route_for_downgrade(context):
|
|
"""Create graph route config for downgrade."""
|
|
context.route_config = RouteConfig(
|
|
name="downgrade_test_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"start": {"type": "start"},
|
|
"process": {"type": "agent", "agent": "test_agent"},
|
|
"end": {"type": "end"},
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "process"},
|
|
{"source": "process", "target": "end"},
|
|
],
|
|
subscriptions=["input_stream"],
|
|
publications=["output_stream"],
|
|
)
|
|
|
|
|
|
@given("I have a LangGraph instance")
|
|
def step_langgraph_instance(context):
|
|
"""Create LangGraph instance."""
|
|
context.mock_graph = MockGraph(context.route_config)
|
|
|
|
|
|
@when("I downgrade graph to stream")
|
|
@async_run_until_complete
|
|
async def step_downgrade_graph_to_stream(context):
|
|
"""Downgrade graph to stream."""
|
|
try:
|
|
context.result = await context.route_bridge.downgrade_graph_to_stream(
|
|
context.route_config, context.mock_graph
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("a StreamConfig should be created")
|
|
def step_verify_stream_config_created(context):
|
|
"""Verify StreamConfig was created."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, StreamConfig)
|
|
|
|
|
|
@then("the stream should be configured correctly")
|
|
def step_verify_stream_configured(context):
|
|
"""Verify stream configuration."""
|
|
stream_config = context.result
|
|
assert stream_config.name == "downgrade_test_route_stream"
|
|
assert stream_config.type == StreamType.COLD
|
|
|
|
|
|
@given("I have a graph route config with state flattener")
|
|
def step_graph_route_state_flattener(context):
|
|
"""Create graph route with state flattener."""
|
|
context.state_flattener = Mock(return_value={"flattened": "data"})
|
|
bridge_config = BridgeConfig(state_flattener=context.state_flattener)
|
|
context.route_config = RouteConfig(
|
|
name="flattener_test_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a LangGraph instance with state")
|
|
def step_langgraph_with_state(context):
|
|
"""Create LangGraph instance with state."""
|
|
context.mock_graph = MockGraph(context.route_config)
|
|
context.mock_graph.state_manager.get_state.return_value = {"test": "state_data"}
|
|
|
|
|
|
@then("the state should be flattened correctly")
|
|
def step_verify_state_flattened(context):
|
|
"""Verify state flattening."""
|
|
context.state_flattener.assert_called_once()
|
|
|
|
|
|
@given("I have a graph route config with preserve checkpointing")
|
|
def step_graph_route_preserve_checkpointing(context):
|
|
"""Create graph route with preserve checkpointing."""
|
|
bridge_config = BridgeConfig(preserve_checkpointing=True)
|
|
context.route_config = RouteConfig(
|
|
name="checkpoint_test_route",
|
|
type=RouteType.GRAPH,
|
|
nodes={"start": {}, "end": {}},
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@given("I have a LangGraph instance with checkpoint directory")
|
|
def step_langgraph_with_checkpoint_dir(context):
|
|
"""Create LangGraph instance with checkpoint directory."""
|
|
context.mock_graph = MockGraph(context.route_config)
|
|
context.mock_graph.state_manager.checkpoint_dir = "/tmp/checkpoints"
|
|
|
|
|
|
@then("checkpointing should be preserved")
|
|
def step_verify_checkpointing_preserved(context):
|
|
"""Verify checkpointing is preserved."""
|
|
context.mock_graph.state_manager._save_checkpoint.assert_called_once()
|
|
|
|
|
|
@given("I have a stream route config with basic operators")
|
|
def step_stream_route_basic_operators(context):
|
|
"""Create stream route with basic operators."""
|
|
context.route_config = RouteConfig(
|
|
name="basic_ops_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[
|
|
{"type": "filter", "params": {"condition": "test"}},
|
|
{"type": "buffer", "params": {"count": 3}},
|
|
],
|
|
)
|
|
|
|
|
|
@when("I create graph from stream")
|
|
def step_create_graph_from_stream(context):
|
|
"""Create graph from stream."""
|
|
try:
|
|
context.result = context.route_bridge._create_graph_from_stream(
|
|
context.route_config
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("a GraphConfig should be created")
|
|
def step_verify_graph_config_created(context):
|
|
"""Verify GraphConfig was created."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, GraphConfig)
|
|
|
|
|
|
@then("nodes should be created from operators")
|
|
def step_verify_nodes_from_operators(context):
|
|
"""Verify nodes were created from operators."""
|
|
graph_config = context.result
|
|
assert len(graph_config.nodes) == 2 # Two operators
|
|
node_names = list(graph_config.nodes.keys())
|
|
assert "op_0_filter" in node_names
|
|
assert "op_1_buffer" in node_names
|
|
|
|
|
|
@then("edges should connect the nodes properly")
|
|
def step_verify_edges_connect_nodes(context):
|
|
"""Verify edges connect nodes properly."""
|
|
graph_config = context.result
|
|
assert len(graph_config.edges) == 3 # start->op1, op1->op2, op2->end
|
|
edge_sources = [edge.source for edge in graph_config.edges]
|
|
edge_targets = [edge.target for edge in graph_config.edges]
|
|
assert "start" in edge_sources
|
|
assert "end" in edge_targets
|
|
|
|
|
|
@given("I have a stream route config with agent operators")
|
|
def step_stream_route_agent_operators(context):
|
|
"""Create stream route with agent operators."""
|
|
context.route_config = RouteConfig(
|
|
name="agent_ops_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "test_agent"}},
|
|
{"type": "map", "params": {"agent": "another_agent"}},
|
|
],
|
|
)
|
|
|
|
|
|
@then("a GraphConfig should be created with agent nodes")
|
|
def step_verify_graph_config_agent_nodes(context):
|
|
"""Verify GraphConfig with agent nodes."""
|
|
graph_config = context.result
|
|
agent_nodes = [
|
|
node for node in graph_config.nodes.values() if node.type == NodeType.AGENT
|
|
]
|
|
assert len(agent_nodes) == 2
|
|
|
|
|
|
@then("agent nodes should be configured correctly")
|
|
def step_verify_agent_nodes_configured(context):
|
|
"""Verify agent nodes are configured correctly."""
|
|
graph_config = context.result
|
|
for node in graph_config.nodes.values():
|
|
if node.type == NodeType.AGENT:
|
|
assert node.agent in ["test_agent", "another_agent"]
|
|
|
|
|
|
@given("I have a stream route config with mixed operators")
|
|
def step_stream_route_mixed_operators(context):
|
|
"""Create stream route with mixed operators."""
|
|
context.route_config = RouteConfig(
|
|
name="mixed_ops_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[
|
|
{"type": "map", "params": {"agent": "test_agent"}},
|
|
{"type": "filter", "params": {"condition": "test"}},
|
|
{"type": "buffer", "params": {"count": 3}},
|
|
],
|
|
)
|
|
|
|
|
|
@then("a GraphConfig should be created with mixed node types")
|
|
def step_verify_graph_config_mixed_nodes(context):
|
|
"""Verify GraphConfig with mixed node types."""
|
|
graph_config = context.result
|
|
node_types = [node.type for node in graph_config.nodes.values()]
|
|
assert NodeType.AGENT in node_types
|
|
assert NodeType.FUNCTION in node_types
|
|
|
|
|
|
@then("all operators should be converted to nodes")
|
|
def step_verify_all_operators_converted(context):
|
|
"""Verify all operators are converted to nodes."""
|
|
graph_config = context.result
|
|
assert len(graph_config.nodes) == 3 # Three operators
|
|
|
|
|
|
@given("I have a LangGraph with nodes")
|
|
def step_langgraph_with_nodes(context):
|
|
"""Create LangGraph with nodes."""
|
|
graph_config = GraphConfig(
|
|
name="test_graph",
|
|
nodes={
|
|
"op_0_map": NodeConfig(
|
|
name="op_0_map", type=NodeType.FUNCTION, function="test_function"
|
|
),
|
|
"op_1_filter": NodeConfig(
|
|
name="op_1_filter", type=NodeType.FUNCTION, function="filter_function"
|
|
),
|
|
},
|
|
edges=[],
|
|
entry_point="start",
|
|
)
|
|
context.mock_graph = MockGraph(graph_config)
|
|
|
|
|
|
@when("I create stream from graph")
|
|
def step_create_stream_from_graph(context):
|
|
"""Create stream from graph."""
|
|
try:
|
|
context.result = context.route_bridge._create_stream_from_graph(
|
|
context.route_config, context.mock_graph
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("operators should be extracted from nodes")
|
|
def step_verify_operators_extracted(context):
|
|
"""Verify operators are extracted from nodes."""
|
|
stream_config = context.result
|
|
assert len(stream_config.operators) >= 0 # May be empty if no valid conversions
|
|
|
|
|
|
@then("the stream should be configured properly")
|
|
def step_verify_stream_configured_properly(context):
|
|
"""Verify stream is configured properly."""
|
|
stream_config = context.result
|
|
assert stream_config.name.endswith("_stream")
|
|
assert stream_config.type == StreamType.COLD
|
|
|
|
|
|
@given("I have a LangGraph with agent nodes")
|
|
def step_langgraph_with_agent_nodes(context):
|
|
"""Create LangGraph with agent nodes."""
|
|
graph_config = GraphConfig(
|
|
name="agent_graph",
|
|
nodes={
|
|
"agent_node": NodeConfig(
|
|
name="agent_node", type=NodeType.AGENT, agent="test_agent"
|
|
),
|
|
},
|
|
edges=[],
|
|
entry_point="start",
|
|
)
|
|
context.mock_graph = MockGraph(graph_config)
|
|
|
|
|
|
@then("a StreamConfig should be created with agent operators")
|
|
def step_verify_stream_config_agent_operators(context):
|
|
"""Verify StreamConfig with agent operators."""
|
|
stream_config = context.result
|
|
agent_operators = [op for op in stream_config.operators if op.get("type") == "map"]
|
|
# May be empty due to filtering logic in implementation
|
|
assert isinstance(stream_config.operators, list)
|
|
|
|
|
|
@then("agents should be extracted correctly")
|
|
def step_verify_agents_extracted(context):
|
|
"""Verify agents are extracted correctly."""
|
|
stream_config = context.result
|
|
assert isinstance(stream_config.agents, list)
|
|
|
|
|
|
@given("I have a LangGraph with function nodes")
|
|
def step_langgraph_with_function_nodes(context):
|
|
"""Create LangGraph with function nodes."""
|
|
graph_config = GraphConfig(
|
|
name="function_graph",
|
|
nodes={
|
|
"func_node": NodeConfig(
|
|
name="func_node",
|
|
type=NodeType.FUNCTION,
|
|
function={"type": "original_operator", "params": {"test": "value"}},
|
|
),
|
|
},
|
|
edges=[],
|
|
entry_point="start",
|
|
)
|
|
context.mock_graph = MockGraph(graph_config)
|
|
|
|
|
|
@then("a StreamConfig should be created with function operators")
|
|
def step_verify_stream_config_function_operators(context):
|
|
"""Verify StreamConfig with function operators."""
|
|
stream_config = context.result
|
|
assert isinstance(stream_config.operators, list)
|
|
|
|
|
|
@then("function operators should be restored correctly")
|
|
def step_verify_function_operators_restored(context):
|
|
"""Verify function operators are restored correctly."""
|
|
stream_config = context.result
|
|
# Check if any operators have the expected structure
|
|
for op in stream_config.operators:
|
|
if isinstance(op, dict) and "type" in op:
|
|
assert "params" in op or op["type"] is not None
|
|
|
|
|
|
@when("I get message count for a route")
|
|
def step_get_message_count(context):
|
|
"""Get message count for a route."""
|
|
try:
|
|
context.result = context.route_bridge._get_message_count("test_route")
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the message count should be returned")
|
|
def step_verify_message_count_returned(context):
|
|
"""Verify message count is returned."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
|
|
|
|
@then("the result should be a valid integer")
|
|
def step_verify_result_integer(context):
|
|
"""Verify result is a valid integer."""
|
|
assert isinstance(context.result, int)
|
|
assert context.result >= 0
|
|
|
|
|
|
@given("I have an active conversion registered")
|
|
def step_active_conversion_registered(context):
|
|
"""Register an active conversion."""
|
|
context.route_bridge._active_conversions["test_route"] = {
|
|
"type": "graph",
|
|
"instance": Mock(),
|
|
"original_config": Mock(),
|
|
}
|
|
|
|
|
|
@when("I get active conversion info")
|
|
def step_get_active_conversion_info(context):
|
|
"""Get active conversion info."""
|
|
try:
|
|
context.result = context.route_bridge.get_active_conversion("test_route")
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the conversion info should be returned")
|
|
def step_verify_conversion_info_returned(context):
|
|
"""Verify conversion info is returned."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, dict)
|
|
|
|
|
|
@then("the info should contain the correct details")
|
|
def step_verify_info_correct_details(context):
|
|
"""Verify info contains correct details."""
|
|
info = context.result
|
|
assert "type" in info
|
|
assert "instance" in info
|
|
assert "original_config" in info
|
|
|
|
|
|
@when("I get active conversion info for non-existing route")
|
|
def step_get_conversion_info_nonexisting(context):
|
|
"""Get active conversion info for non-existing route."""
|
|
try:
|
|
context.result = context.route_bridge.get_active_conversion("nonexistent")
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("None should be returned for route bridge")
|
|
def step_verify_none_returned_route_bridge(context):
|
|
"""Verify None is returned."""
|
|
assert context.error is None
|
|
assert context.result is None
|
|
|
|
|
|
@given("I have an invalid stream route config")
|
|
def step_invalid_stream_route_config(context):
|
|
"""Create invalid stream route config."""
|
|
context.route_config = RouteConfig(
|
|
name="invalid_route",
|
|
type=RouteType.STREAM, # Missing required fields
|
|
)
|
|
|
|
|
|
@when("I attempt to upgrade stream to graph")
|
|
@async_run_until_complete
|
|
async def step_attempt_upgrade_stream_to_graph(context):
|
|
"""Attempt to upgrade stream to graph."""
|
|
try:
|
|
# Mock LangGraph to avoid complex dependencies
|
|
import cleveractors.reactive.route_bridge
|
|
|
|
original_langgraph = cleveractors.reactive.route_bridge.LangGraph
|
|
cleveractors.reactive.route_bridge.LangGraph = MockGraph
|
|
|
|
context.result = await context.route_bridge.upgrade_stream_to_graph(
|
|
context.route_config, context.stream_message
|
|
)
|
|
|
|
# Restore original
|
|
cleveractors.reactive.route_bridge.LangGraph = original_langgraph
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("appropriate error handling should occur")
|
|
def step_verify_error_handling(context):
|
|
"""Verify appropriate error handling."""
|
|
# Either an error occurred or the system handled it gracefully
|
|
assert context.error is not None or context.result is not None
|
|
|
|
|
|
@then("the system should remain stable")
|
|
def step_verify_system_stable(context):
|
|
"""Verify system remains stable."""
|
|
# The route bridge should still be functional
|
|
assert context.route_bridge is not None
|
|
assert hasattr(context.route_bridge, "_active_conversions")
|
|
|
|
|
|
@given("I have an invalid graph route config")
|
|
def step_invalid_graph_route_config(context):
|
|
"""Create invalid graph route config."""
|
|
# Create a valid config but with problematic content
|
|
context.route_config = RouteConfig(
|
|
name="invalid_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"invalid": {"type": "unknown"}
|
|
}, # Valid structure but problematic content
|
|
)
|
|
|
|
|
|
@given("I have a corrupted LangGraph instance")
|
|
def step_corrupted_langgraph_instance(context):
|
|
"""Create corrupted LangGraph instance."""
|
|
context.mock_graph = Mock()
|
|
# Simulate corruption by making methods raise exceptions
|
|
context.mock_graph._topological_levels.side_effect = Exception("Corrupted graph")
|
|
|
|
|
|
@when("I attempt to downgrade graph to stream")
|
|
@async_run_until_complete
|
|
async def step_attempt_downgrade_graph_to_stream(context):
|
|
"""Attempt to downgrade graph to stream."""
|
|
try:
|
|
context.result = await context.route_bridge.downgrade_graph_to_stream(
|
|
context.route_config, context.mock_graph
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@given("I have a route config that supports both upgrade and downgrade")
|
|
def step_route_config_both_upgrade_downgrade(context):
|
|
"""Create route config supporting both upgrade and downgrade."""
|
|
bridge_config = BridgeConfig(
|
|
upgrade_conditions={"needs_state": True},
|
|
downgrade_conditions={"idle_time": 30.0},
|
|
)
|
|
context.route_config = RouteConfig(
|
|
name="bidirectional_route",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[{"type": "map", "params": {"agent": "test_agent"}}],
|
|
bridge=bridge_config,
|
|
)
|
|
|
|
|
|
@when("I perform an upgrade and then downgrade cycle")
|
|
@async_run_until_complete
|
|
async def step_perform_upgrade_downgrade_cycle(context):
|
|
"""Perform upgrade and then downgrade cycle."""
|
|
try:
|
|
# Mock LangGraph
|
|
import cleveractors.reactive.route_bridge
|
|
|
|
original_langgraph = cleveractors.reactive.route_bridge.LangGraph
|
|
cleveractors.reactive.route_bridge.LangGraph = MockGraph
|
|
|
|
# First upgrade
|
|
upgraded_graph = await context.route_bridge.upgrade_stream_to_graph(
|
|
context.route_config, context.stream_message
|
|
)
|
|
|
|
# Then downgrade
|
|
context.result = await context.route_bridge.downgrade_graph_to_stream(
|
|
context.route_config, upgraded_graph
|
|
)
|
|
|
|
# Restore original
|
|
cleveractors.reactive.route_bridge.LangGraph = original_langgraph
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the cycle should complete successfully")
|
|
def step_verify_cycle_completed(context):
|
|
"""Verify cycle completed successfully."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
|
|
|
|
@then("the final configuration should be consistent")
|
|
def step_verify_final_config_consistent(context):
|
|
"""Verify final configuration is consistent."""
|
|
final_stream = context.result
|
|
assert isinstance(final_stream, StreamConfig)
|
|
assert final_stream.name.endswith("_stream")
|
|
|
|
|
|
@then("active conversions should be tracked correctly")
|
|
def step_verify_conversions_tracked(context):
|
|
"""Verify active conversions are tracked correctly."""
|
|
conversions = context.route_bridge._active_conversions
|
|
assert "bidirectional_route" in conversions
|
|
# Should have the final state (stream)
|
|
assert conversions["bidirectional_route"]["type"] == "stream"
|
|
|
|
|
|
# Note: Cleanup is now handled inline in step functions to avoid global state issues
|