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
321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""
|
|
BDD step definitions for unified routes testing.
|
|
"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveractors.core.application import ReactiveCleverAgentsApp
|
|
from cleveractors.reactive.route import RouteComplexityAnalyzer, RouteType
|
|
|
|
|
|
@given("I have a route configuration")
|
|
def step_route_config(context: Context):
|
|
"""Parse route configuration from docstring."""
|
|
context.config_data = yaml.safe_load(context.text)
|
|
context.temp_file = tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
)
|
|
yaml.dump(context.config_data, context.temp_file)
|
|
context.temp_file.close()
|
|
context.config_file = Path(context.temp_file.name)
|
|
context.config_files = [context.config_file]
|
|
|
|
|
|
@given("I have a route configuration without type")
|
|
def step_route_config_no_type(context: Context):
|
|
"""Parse invalid route configuration."""
|
|
context.config_data = yaml.safe_load(context.text)
|
|
context.temp_file = tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
)
|
|
yaml.dump(context.config_data, context.temp_file)
|
|
context.temp_file.close()
|
|
context.config_file = Path(context.temp_file.name)
|
|
context.config_files = [context.config_file]
|
|
|
|
|
|
@given("I have a route configuration with bridging")
|
|
def step_route_config_bridging(context: Context):
|
|
"""Parse route configuration with bridge settings."""
|
|
context.config_data = yaml.safe_load(context.text)
|
|
# Add missing agent for the example
|
|
if "agents" not in context.config_data:
|
|
context.config_data["agents"] = {
|
|
"processor": {
|
|
"type": "llm",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
}
|
|
|
|
context.temp_file = tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
)
|
|
yaml.dump(context.config_data, context.temp_file)
|
|
context.temp_file.close()
|
|
context.config_file = Path(context.temp_file.name)
|
|
context.config_files = [context.config_file]
|
|
|
|
|
|
@given("I have a configuration with both stream and graph routes")
|
|
def step_mixed_routes_config(context: Context):
|
|
"""Parse configuration with mixed route types."""
|
|
context.config_data = yaml.safe_load(context.text)
|
|
# Add missing agent
|
|
if "agents" not in context.config_data:
|
|
context.config_data["agents"] = {
|
|
"processor": {
|
|
"type": "llm",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
}
|
|
|
|
context.temp_file = tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
)
|
|
yaml.dump(context.config_data, context.temp_file)
|
|
context.temp_file.close()
|
|
context.config_file = Path(context.temp_file.name)
|
|
context.config_files = [context.config_file]
|
|
|
|
|
|
@given("I have a bridge-only route configuration")
|
|
def step_bridge_route_config(context: Context):
|
|
"""Parse bridge route configuration."""
|
|
context.config_data = yaml.safe_load(context.text)
|
|
context.temp_file = tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False
|
|
)
|
|
yaml.dump(context.config_data, context.temp_file)
|
|
context.temp_file.close()
|
|
context.config_file = Path(context.temp_file.name)
|
|
context.config_files = [context.config_file]
|
|
|
|
|
|
@given("I have routes with different complexity levels")
|
|
def step_complex_routes(context: Context):
|
|
"""Create routes with varying complexity."""
|
|
context.app = ReactiveCleverAgentsApp()
|
|
# Create sample routes for complexity analysis
|
|
context.test_routes = {
|
|
"simple_stream": {
|
|
"type": RouteType.STREAM,
|
|
"operators": [{"type": "map"}],
|
|
},
|
|
"complex_stream": {
|
|
"type": RouteType.STREAM,
|
|
"operators": [
|
|
{"type": "map"},
|
|
{"type": "filter"},
|
|
{"type": "debounce"},
|
|
],
|
|
"subscriptions": ["input1", "input2"],
|
|
"publications": ["output1", "output2"],
|
|
},
|
|
"simple_graph": {
|
|
"type": RouteType.GRAPH,
|
|
"nodes": {"node1": {}},
|
|
"edges": [{"source": "start", "target": "node1"}],
|
|
},
|
|
"complex_graph": {
|
|
"type": RouteType.GRAPH,
|
|
"nodes": {"node1": {}, "node2": {}, "node3": {}},
|
|
"edges": [
|
|
{"source": "start", "target": "node1"},
|
|
{"source": "node1", "target": "node2", "condition": "test"},
|
|
{"source": "node2", "target": "node3"},
|
|
],
|
|
"checkpointing": True,
|
|
},
|
|
}
|
|
|
|
|
|
@given("I have a stream route with bridge configuration")
|
|
def step_stream_with_bridge(context: Context):
|
|
"""Create a stream route with bridge config."""
|
|
# This would be set up with a proper route config
|
|
context.has_bridge_route = True
|
|
|
|
|
|
@then('the route "{route_name}" should be of type "{route_type}"')
|
|
def step_verify_route_type(context: Context, route_name: str, route_type: str):
|
|
"""Verify route type."""
|
|
route = context.app.config.routes.get(route_name)
|
|
assert route is not None, f"Route '{route_name}' not found"
|
|
assert route.type.value == route_type.lower()
|
|
|
|
|
|
@then('the route "{route_name}" should have bridge configuration')
|
|
def step_verify_bridge_config(context: Context, route_name: str):
|
|
"""Verify route has bridge configuration."""
|
|
route = context.app.config.routes.get(route_name)
|
|
assert route is not None
|
|
assert route.bridge is not None
|
|
|
|
|
|
@then("the bridge should have upgrade conditions")
|
|
def step_verify_upgrade_conditions(context: Context):
|
|
"""Verify bridge upgrade conditions."""
|
|
# Find route with bridge
|
|
bridge_route = None
|
|
for route in context.app.config.routes.values():
|
|
if route.bridge:
|
|
bridge_route = route
|
|
break
|
|
|
|
assert bridge_route is not None
|
|
assert bridge_route.bridge.upgrade_conditions
|
|
assert "needs_state" in bridge_route.bridge.upgrade_conditions
|
|
assert "message_count" in bridge_route.bridge.upgrade_conditions
|
|
|
|
|
|
@then("I should have {count:d} stream route")
|
|
def step_verify_stream_route_count(context: Context, count: int):
|
|
"""Count stream-type routes."""
|
|
stream_count = sum(
|
|
1
|
|
for route in context.app.config.routes.values()
|
|
if route.type == RouteType.STREAM
|
|
)
|
|
assert stream_count == count
|
|
|
|
|
|
@then("I should have {count:d} graph route")
|
|
def step_verify_graph_route_count(context: Context, count: int):
|
|
"""Count graph-type routes."""
|
|
graph_count = sum(
|
|
1
|
|
for route in context.app.config.routes.values()
|
|
if route.type == RouteType.GRAPH
|
|
)
|
|
assert graph_count == count
|
|
|
|
|
|
@then("I should have {count:d} routes")
|
|
def step_verify_total_route_count(context: Context, count: int):
|
|
"""Count total routes."""
|
|
total_count = len(context.app.config.routes)
|
|
assert total_count == count
|
|
|
|
|
|
@when("I analyze route complexity")
|
|
def step_analyze_complexity(context: Context):
|
|
"""Analyze complexity of test routes."""
|
|
context.complexity_results = {}
|
|
|
|
for name, route_data in context.test_routes.items():
|
|
from cleveractors.reactive.route import RouteConfig
|
|
|
|
# Create RouteConfig from test data
|
|
route_config = RouteConfig(
|
|
name=name,
|
|
type=route_data["type"],
|
|
operators=route_data.get("operators", []),
|
|
subscriptions=route_data.get("subscriptions", []),
|
|
publications=route_data.get("publications", []),
|
|
nodes=route_data.get("nodes", {}),
|
|
edges=route_data.get("edges", []),
|
|
checkpointing=route_data.get("checkpointing", False),
|
|
)
|
|
|
|
# Analyze complexity
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route_config)
|
|
context.complexity_results[name] = analysis
|
|
|
|
|
|
@then("stream routes should have lower complexity scores")
|
|
def step_verify_stream_complexity(context: Context):
|
|
"""Verify stream complexity is lower."""
|
|
stream_scores = [
|
|
result["score"]
|
|
for name, result in context.complexity_results.items()
|
|
if result["type"] == "stream"
|
|
]
|
|
|
|
assert all(score < 10 for score in stream_scores)
|
|
|
|
|
|
@then("graph routes should have higher complexity scores")
|
|
def step_verify_graph_complexity(context: Context):
|
|
"""Verify graph complexity is higher."""
|
|
graph_scores = [
|
|
result["score"]
|
|
for name, result in context.complexity_results.items()
|
|
if result["type"] == "graph"
|
|
]
|
|
|
|
assert all(score >= 5 for score in graph_scores)
|
|
|
|
|
|
@then("routes with more operators should have higher scores")
|
|
def step_verify_operator_complexity(context: Context):
|
|
"""Verify operator count affects complexity."""
|
|
simple_score = context.complexity_results["simple_stream"]["score"]
|
|
complex_score = context.complexity_results["complex_stream"]["score"]
|
|
|
|
assert complex_score > simple_score
|
|
|
|
|
|
@when("the upgrade conditions are met")
|
|
def step_trigger_upgrade(context: Context):
|
|
"""Simulate meeting upgrade conditions."""
|
|
# This would trigger actual upgrade in real scenario
|
|
context.upgrade_triggered = True
|
|
|
|
|
|
@then("the route should upgrade to a graph")
|
|
def step_verify_upgrade(context: Context):
|
|
"""Verify route upgraded to graph."""
|
|
assert context.upgrade_triggered
|
|
|
|
|
|
@then("state should be preserved during conversion")
|
|
def step_verify_state_preserved(context: Context):
|
|
"""Verify state preservation."""
|
|
# In real implementation, would check actual state
|
|
assert True
|
|
|
|
|
|
@then("subscriptions should be maintained")
|
|
def step_verify_subscriptions_maintained(context: Context):
|
|
"""Verify subscriptions maintained."""
|
|
# In real implementation, would check actual subscriptions
|
|
assert True
|
|
|
|
|
|
# Missing step definitions with colons - delegate to existing implementations
|
|
|
|
|
|
@given("I have a route configuration:")
|
|
def step_route_config_with_colon(context: Context):
|
|
"""Parse route configuration."""
|
|
return step_route_config(context)
|
|
|
|
|
|
@given("I have a route configuration without type:")
|
|
def step_route_config_no_type_with_colon(context: Context):
|
|
"""Parse route configuration without type."""
|
|
return step_route_config_no_type(context)
|
|
|
|
|
|
@given("I have a route configuration with bridging:")
|
|
def step_route_config_bridging_with_colon(context: Context):
|
|
"""Parse route configuration with bridging."""
|
|
return step_route_config_bridging(context)
|
|
|
|
|
|
@given("I have a bridge-only route configuration:")
|
|
def step_bridge_only_config_with_colon(context: Context):
|
|
"""Parse bridge-only route configuration."""
|
|
return step_bridge_route_config(context)
|
|
|
|
|
|
@given("I have a configuration with both stream and graph routes:")
|
|
def step_both_routes_config_with_colon(context: Context):
|
|
"""Parse configuration with both stream and graph routes."""
|
|
return step_mixed_routes_config(context)
|