forked from cleveragents/cleveragents-core
674 lines
23 KiB
Python
674 lines
23 KiB
Python
"""
|
|
Comprehensive unit tests for reactive route module.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from cleveragents.core.exceptions import ConfigurationError
|
|
from cleveragents.langgraph.graph import GraphConfig
|
|
from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveragents.reactive.route import (
|
|
BridgeConfig,
|
|
RouteComplexityAnalyzer,
|
|
RouteConfig,
|
|
RouteType,
|
|
)
|
|
from cleveragents.reactive.stream_router import StreamConfig, StreamType
|
|
|
|
|
|
class TestRouteType:
|
|
"""Test cases for RouteType enum."""
|
|
|
|
def test_route_type_values(self):
|
|
"""Test RouteType enum values."""
|
|
assert RouteType.STREAM.value == "stream"
|
|
assert RouteType.GRAPH.value == "graph"
|
|
assert RouteType.BRIDGE.value == "bridge"
|
|
|
|
|
|
class TestBridgeConfig:
|
|
"""Test cases for BridgeConfig dataclass."""
|
|
|
|
def test_bridge_config_defaults(self):
|
|
"""Test BridgeConfig with default values."""
|
|
config = BridgeConfig()
|
|
|
|
assert config.upgrade_conditions == {}
|
|
assert config.downgrade_conditions == {}
|
|
assert config.state_extractor is None
|
|
assert config.state_flattener is None
|
|
assert config.preserve_subscriptions is True
|
|
assert config.preserve_checkpointing is True
|
|
|
|
def test_bridge_config_with_values(self):
|
|
"""Test BridgeConfig with custom values."""
|
|
config = BridgeConfig(
|
|
upgrade_conditions={"complexity": "high"},
|
|
downgrade_conditions={"complexity": "low"},
|
|
state_extractor="extract_func",
|
|
state_flattener="flatten_func",
|
|
preserve_subscriptions=False,
|
|
preserve_checkpointing=False
|
|
)
|
|
|
|
assert config.upgrade_conditions == {"complexity": "high"}
|
|
assert config.downgrade_conditions == {"complexity": "low"}
|
|
assert config.state_extractor == "extract_func"
|
|
assert config.state_flattener == "flatten_func"
|
|
assert config.preserve_subscriptions is False
|
|
assert config.preserve_checkpointing is False
|
|
|
|
|
|
class TestRouteConfig:
|
|
"""Test cases for RouteConfig dataclass."""
|
|
|
|
def test_route_config_stream_minimal(self):
|
|
"""Test RouteConfig for stream with minimal config."""
|
|
config = RouteConfig(name="test_stream", type=RouteType.STREAM)
|
|
|
|
assert config.name == "test_stream"
|
|
assert config.type == RouteType.STREAM
|
|
assert config.stream_type == StreamType.COLD # Auto-set in __post_init__
|
|
assert config.operators == []
|
|
|
|
def test_route_config_stream_full(self):
|
|
"""Test RouteConfig for stream with full config."""
|
|
config = RouteConfig(
|
|
name="test_stream",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.HOT,
|
|
operators=[{"type": "map", "function": "transform"}],
|
|
subscriptions=["input"],
|
|
publications=["output"],
|
|
agents=["agent1"],
|
|
initial_value="init",
|
|
buffer_size=10,
|
|
metadata={"key": "value"}
|
|
)
|
|
|
|
assert config.stream_type == StreamType.HOT
|
|
assert len(config.operators) == 1
|
|
assert config.subscriptions == ["input"]
|
|
assert config.publications == ["output"]
|
|
assert config.agents == ["agent1"]
|
|
assert config.initial_value == "init"
|
|
assert config.buffer_size == 10
|
|
assert config.metadata["key"] == "value"
|
|
|
|
def test_route_config_graph_minimal(self):
|
|
"""Test RouteConfig for graph with minimal config."""
|
|
config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={"node1": {"type": "agent"}}
|
|
)
|
|
|
|
assert config.name == "test_graph"
|
|
assert config.type == RouteType.GRAPH
|
|
assert "node1" in config.nodes
|
|
assert config.entry_point == "start"
|
|
assert config.checkpointing is False
|
|
|
|
def test_route_config_graph_no_nodes_error(self):
|
|
"""Test that graph without nodes raises error."""
|
|
with pytest.raises(ConfigurationError, match="must have nodes defined"):
|
|
RouteConfig(name="test_graph", type=RouteType.GRAPH)
|
|
|
|
def test_route_config_graph_with_empty_nodes_error(self):
|
|
"""Test that graph with empty nodes dict raises error."""
|
|
with pytest.raises(ConfigurationError):
|
|
RouteConfig(name="test_graph", type=RouteType.GRAPH, nodes={})
|
|
|
|
def test_route_config_graph_full(self):
|
|
"""Test RouteConfig for graph with full config."""
|
|
config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"start": {"type": "start"},
|
|
"agent": {"type": "agent", "agent": "my_agent"},
|
|
"end": {"type": "end"}
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "agent"},
|
|
{"source": "agent", "target": "end"}
|
|
],
|
|
entry_point="start",
|
|
checkpointing=True,
|
|
checkpoint_dir="/tmp/checkpoints",
|
|
enable_time_travel=True,
|
|
parallel_execution=False,
|
|
state_class="CustomState",
|
|
metadata={"version": "1.0"}
|
|
)
|
|
|
|
assert len(config.nodes) == 3
|
|
assert len(config.edges) == 2
|
|
assert config.checkpointing is True
|
|
assert config.checkpoint_dir == "/tmp/checkpoints"
|
|
assert config.enable_time_travel is True
|
|
assert config.parallel_execution is False
|
|
assert config.state_class == "CustomState"
|
|
|
|
def test_route_config_with_bridge(self):
|
|
"""Test RouteConfig with bridge configuration."""
|
|
bridge = BridgeConfig(
|
|
upgrade_conditions={"type": "complex"}
|
|
)
|
|
config = RouteConfig(
|
|
name="test_route",
|
|
type=RouteType.STREAM,
|
|
bridge=bridge
|
|
)
|
|
|
|
assert config.bridge is not None
|
|
assert config.bridge.upgrade_conditions == {"type": "complex"}
|
|
|
|
def test_route_config_with_template(self):
|
|
"""Test RouteConfig with template configuration."""
|
|
config = RouteConfig(
|
|
name="test_route",
|
|
type=RouteType.STREAM,
|
|
template_config={"template": "my_template", "params": {}}
|
|
)
|
|
|
|
assert config.template_config is not None
|
|
assert config.template_config["template"] == "my_template"
|
|
|
|
def test_to_stream_config_success(self):
|
|
"""Test converting RouteConfig to StreamConfig."""
|
|
route_config = RouteConfig(
|
|
name="test_stream",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.HOT,
|
|
operators=[{"type": "map"}],
|
|
subscriptions=["input"],
|
|
publications=["output"],
|
|
agents=["agent1"],
|
|
initial_value="init",
|
|
buffer_size=5,
|
|
template_config={"key": "value"}
|
|
)
|
|
|
|
stream_config = route_config.to_stream_config()
|
|
|
|
assert stream_config.name == "test_stream"
|
|
assert stream_config.type == StreamType.HOT
|
|
assert len(stream_config.operators) == 1
|
|
assert stream_config.subscriptions == ["input"]
|
|
assert stream_config.publications == ["output"]
|
|
assert stream_config.agents == ["agent1"]
|
|
assert stream_config.initial_value == "init"
|
|
assert stream_config.buffer_size == 5
|
|
assert stream_config.template_config == {"key": "value"}
|
|
|
|
def test_to_stream_config_from_graph_error(self):
|
|
"""Test that converting graph to stream raises error."""
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={"node1": {"type": "agent"}}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Cannot convert.*to StreamConfig"):
|
|
route_config.to_stream_config()
|
|
|
|
def test_to_graph_config_success(self):
|
|
"""Test converting RouteConfig to GraphConfig."""
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"start": {"type": "start"},
|
|
"agent": {"type": "agent", "agent": "my_agent"},
|
|
"end": {"type": "end"}
|
|
},
|
|
edges=[
|
|
{"source": "start", "target": "agent"},
|
|
{"source": "agent", "target": "end"}
|
|
],
|
|
entry_point="start",
|
|
checkpointing=True,
|
|
checkpoint_dir="/tmp/ckpt",
|
|
enable_time_travel=True,
|
|
parallel_execution=False,
|
|
metadata={"key": "value"}
|
|
)
|
|
|
|
graph_config = route_config.to_graph_config()
|
|
|
|
assert isinstance(graph_config, GraphConfig)
|
|
assert graph_config.name == "test_graph"
|
|
assert len(graph_config.nodes) == 3
|
|
assert len(graph_config.edges) == 2
|
|
assert graph_config.entry_point == "start"
|
|
assert graph_config.checkpointing is True
|
|
assert graph_config.enable_time_travel is True
|
|
assert graph_config.parallel_execution is False
|
|
assert graph_config.metadata == {"key": "value"}
|
|
|
|
def test_to_graph_config_with_node_details(self):
|
|
"""Test graph config conversion with detailed node config."""
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"agent_node": {
|
|
"type": "agent",
|
|
"agent": "my_agent",
|
|
"retry_policy": {"max_retries": 3},
|
|
"timeout": 30.0,
|
|
"parallel": True,
|
|
"metadata": {"priority": "high"}
|
|
},
|
|
"function_node": {
|
|
"type": "function",
|
|
"function": "process",
|
|
"tools": ["tool1", "tool2"],
|
|
"condition": {"type": "always"},
|
|
"subgraph": "sub_graph"
|
|
}
|
|
},
|
|
edges=[{"source": "agent_node", "target": "function_node"}]
|
|
)
|
|
|
|
graph_config = route_config.to_graph_config()
|
|
|
|
# Verify node configs
|
|
assert "agent_node" in graph_config.nodes
|
|
assert "function_node" in graph_config.nodes
|
|
|
|
agent_node = graph_config.nodes["agent_node"]
|
|
assert agent_node.type == NodeType.AGENT
|
|
assert agent_node.agent == "my_agent"
|
|
assert agent_node.retry_policy == {"max_retries": 3}
|
|
assert agent_node.timeout == 30.0
|
|
assert agent_node.parallel is True
|
|
|
|
function_node = graph_config.nodes["function_node"]
|
|
assert function_node.type == NodeType.FUNCTION
|
|
assert function_node.function == "process"
|
|
assert function_node.tools == ["tool1", "tool2"]
|
|
|
|
def test_to_graph_config_with_edge_details(self):
|
|
"""Test graph config conversion with edge conditions and metadata."""
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
"node1": {"type": "agent"},
|
|
"node2": {"type": "agent"}
|
|
},
|
|
edges=[{
|
|
"source": "node1",
|
|
"target": "node2",
|
|
"condition": {"type": "has_messages"},
|
|
"metadata": {"priority": "high"}
|
|
}]
|
|
)
|
|
|
|
graph_config = route_config.to_graph_config()
|
|
|
|
assert len(graph_config.edges) == 1
|
|
edge = graph_config.edges[0]
|
|
assert edge.source == "node1"
|
|
assert edge.target == "node2"
|
|
assert edge.condition == {"type": "has_messages"}
|
|
assert edge.metadata == {"priority": "high"}
|
|
|
|
def test_to_graph_config_from_stream_error(self):
|
|
"""Test that converting stream to graph raises error."""
|
|
route_config = RouteConfig(
|
|
name="test_stream",
|
|
type=RouteType.STREAM
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Cannot convert.*to GraphConfig"):
|
|
route_config.to_graph_config()
|
|
|
|
def test_to_graph_config_with_checkpoint_path(self):
|
|
"""Test graph config conversion with checkpoint path."""
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={"node1": {"type": "agent"}},
|
|
checkpointing=True,
|
|
checkpoint_dir="/my/checkpoint/dir"
|
|
)
|
|
|
|
graph_config = route_config.to_graph_config()
|
|
|
|
assert graph_config.checkpointing is True
|
|
assert str(graph_config.checkpoint_dir) == "/my/checkpoint/dir"
|
|
|
|
def test_route_config_bridge_type(self):
|
|
"""Test RouteConfig with BRIDGE type."""
|
|
config = RouteConfig(
|
|
name="test_bridge",
|
|
type=RouteType.BRIDGE,
|
|
bridge=BridgeConfig(
|
|
upgrade_conditions={"complexity": "high"}
|
|
)
|
|
)
|
|
|
|
assert config.type == RouteType.BRIDGE
|
|
assert config.bridge is not None
|
|
|
|
def test_node_type_conversion_all_types(self):
|
|
"""Test that all node types are properly converted."""
|
|
for node_type_str in ["start", "end", "agent", "function", "tool", "conditional", "subgraph"]:
|
|
route_config = RouteConfig(
|
|
name="test_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={"node": {"type": node_type_str}},
|
|
edges=[]
|
|
)
|
|
|
|
graph_config = route_config.to_graph_config()
|
|
node = graph_config.nodes["node"]
|
|
|
|
expected_type = NodeType[node_type_str.upper()]
|
|
assert node.type == expected_type
|
|
|
|
|
|
class TestRouteConfigFromStreamConfig:
|
|
"""Test cases for from_stream_config classmethod."""
|
|
|
|
def test_from_stream_config_basic(self):
|
|
"""Test creating RouteConfig from StreamConfig."""
|
|
stream_config = StreamConfig(
|
|
name="test_stream",
|
|
type=StreamType.COLD,
|
|
operators=[{"type": "map"}],
|
|
subscriptions=["input"],
|
|
publications=["output"],
|
|
agents=["agent1"]
|
|
)
|
|
|
|
route = RouteConfig.from_stream_config(stream_config)
|
|
|
|
assert route.name == "test_stream"
|
|
assert route.type == RouteType.STREAM
|
|
assert route.stream_type == StreamType.COLD
|
|
assert len(route.operators) == 1
|
|
assert route.subscriptions == ["input"]
|
|
assert route.agents == ["agent1"]
|
|
|
|
def test_from_stream_config_with_initial_value(self):
|
|
"""Test from_stream_config with initial value."""
|
|
stream_config = StreamConfig(
|
|
name="hot_stream",
|
|
type=StreamType.HOT,
|
|
initial_value="initial"
|
|
)
|
|
|
|
route = RouteConfig.from_stream_config(stream_config)
|
|
|
|
assert route.stream_type == StreamType.HOT
|
|
assert route.initial_value == "initial"
|
|
|
|
|
|
class TestRouteConfigFromGraphConfig:
|
|
"""Test cases for from_graph_config classmethod."""
|
|
|
|
def test_from_graph_config_basic(self):
|
|
"""Test creating RouteConfig from GraphConfig."""
|
|
nodes = {
|
|
"node1": NodeConfig(name="node1", type=NodeType.AGENT, agent="agent1")
|
|
}
|
|
edges = [Edge(source="start", target="node1")]
|
|
|
|
graph_config = GraphConfig(
|
|
name="test_graph",
|
|
nodes=nodes,
|
|
edges=edges
|
|
)
|
|
|
|
route = RouteConfig.from_graph_config(graph_config)
|
|
|
|
assert route.name == "test_graph"
|
|
assert route.type == RouteType.GRAPH
|
|
assert "node1" in route.nodes
|
|
assert route.nodes["node1"]["type"] == "agent"
|
|
assert len(route.edges) == 1
|
|
|
|
def test_from_graph_config_with_all_node_fields(self):
|
|
"""Test from_graph_config with all node fields."""
|
|
nodes = {
|
|
"node1": NodeConfig(
|
|
name="node1",
|
|
type=NodeType.FUNCTION,
|
|
function="my_function",
|
|
tools=["tool1"],
|
|
retry_policy={"max_retries": 3},
|
|
timeout=30,
|
|
parallel=True
|
|
)
|
|
}
|
|
|
|
graph_config = GraphConfig(name="test_graph", nodes=nodes, edges=[])
|
|
|
|
route = RouteConfig.from_graph_config(graph_config)
|
|
|
|
node_dict = route.nodes["node1"]
|
|
assert node_dict["function"] == "my_function"
|
|
assert node_dict["tools"] == ["tool1"]
|
|
assert node_dict["retry_policy"]["max_retries"] == 3
|
|
assert node_dict["timeout"] == 30
|
|
assert node_dict["parallel"] is True
|
|
|
|
def test_from_graph_config_with_conditional_edges(self):
|
|
"""Test from_graph_config with conditional edges."""
|
|
nodes = {
|
|
"node1": NodeConfig(name="node1", type=NodeType.AGENT)
|
|
}
|
|
edges = [
|
|
Edge(source="start", target="node1", condition="check_condition")
|
|
]
|
|
|
|
graph_config = GraphConfig(name="test_graph", nodes=nodes, edges=edges)
|
|
|
|
route = RouteConfig.from_graph_config(graph_config)
|
|
|
|
assert route.edges[0]["condition"] == "check_condition"
|
|
|
|
|
|
class TestRouteComplexityAnalyzer:
|
|
"""Test cases for RouteComplexityAnalyzer."""
|
|
|
|
def test_analyze_simple_stream(self):
|
|
"""Test analyzing a simple stream route."""
|
|
route = RouteConfig(
|
|
name="simple_stream",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.COLD,
|
|
operators=[]
|
|
)
|
|
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route)
|
|
|
|
assert analysis["type"] == "stream"
|
|
assert analysis["complexity"] == "simple"
|
|
assert analysis["score"] == 1
|
|
|
|
def test_analyze_complex_stream(self):
|
|
"""Test analyzing a complex stream route."""
|
|
route = RouteConfig(
|
|
name="complex_stream",
|
|
type=RouteType.STREAM,
|
|
stream_type=StreamType.HOT,
|
|
operators=[
|
|
{"type": "map"},
|
|
{"type": "filter"},
|
|
{"type": "buffer"},
|
|
{"type": "merge"}
|
|
],
|
|
subscriptions=["input1", "input2"],
|
|
publications=["output"]
|
|
)
|
|
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route)
|
|
|
|
assert analysis["type"] == "stream"
|
|
assert analysis["complexity"] == "complex"
|
|
assert analysis["score"] > 5
|
|
assert "operators" in analysis["features"][0]
|
|
assert "recommendation" in analysis
|
|
|
|
def test_analyze_simple_graph(self):
|
|
"""Test analyzing a simple graph route."""
|
|
|
|
route = RouteConfig(
|
|
name="simple_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={"node1": {"type": "agent"}},
|
|
edges=[{"source": "start", "target": "node1"}]
|
|
)
|
|
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route)
|
|
|
|
assert analysis["type"] == "graph"
|
|
assert analysis["complexity"] == "moderate"
|
|
assert "recommendation" in analysis
|
|
|
|
def test_analyze_complex_graph(self):
|
|
"""Test analyzing a complex graph route."""
|
|
|
|
route = RouteConfig(
|
|
name="complex_graph",
|
|
type=RouteType.GRAPH,
|
|
nodes={
|
|
f"node{i}": {"type": "agent"}
|
|
for i in range(10)
|
|
},
|
|
edges=[
|
|
{"source": f"node{i}", "target": f"node{i+1}", "condition": f"check_{i}"}
|
|
for i in range(9)
|
|
],
|
|
checkpointing=True,
|
|
enable_time_travel=True
|
|
)
|
|
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route)
|
|
|
|
assert analysis["type"] == "graph"
|
|
assert analysis["complexity"] in ["complex", "advanced"]
|
|
assert "checkpointing enabled" in analysis["features"]
|
|
assert "time travel enabled" in analysis["features"]
|
|
assert "conditional edges" in analysis["features"][-1]
|
|
|
|
def test_analyze_bridge_route(self):
|
|
"""Test analyzing a bridge route."""
|
|
|
|
route = RouteConfig(
|
|
name="bridge_route",
|
|
type=RouteType.BRIDGE,
|
|
bridge=BridgeConfig()
|
|
)
|
|
|
|
analysis = RouteComplexityAnalyzer.analyze_route(route)
|
|
|
|
assert analysis["complexity"] == "bridge"
|
|
assert analysis["score"] == 0
|
|
|
|
|
|
class TestRouteComplexityRecommendations:
|
|
"""Test cases for complexity recommendation methods."""
|
|
|
|
def test_stream_recommendation_simple(self):
|
|
"""Test stream recommendation for simple complexity."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_stream_recommendation(1)
|
|
|
|
assert "simple transformations" in recommendation
|
|
|
|
def test_stream_recommendation_moderate(self):
|
|
"""Test stream recommendation for moderate complexity."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_stream_recommendation(4)
|
|
|
|
assert "multi-step processing" in recommendation
|
|
|
|
def test_stream_recommendation_complex(self):
|
|
"""Test stream recommendation for complex streams."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_stream_recommendation(7)
|
|
|
|
assert "graph" in recommendation.lower()
|
|
|
|
def test_graph_recommendation_moderate(self):
|
|
"""Test graph recommendation for moderate complexity."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_graph_recommendation(8)
|
|
|
|
assert "conditional logic" in recommendation
|
|
|
|
def test_graph_recommendation_complex(self):
|
|
"""Test graph recommendation for complex graphs."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_graph_recommendation(12)
|
|
|
|
assert "stateful workflows" in recommendation
|
|
|
|
def test_graph_recommendation_advanced(self):
|
|
"""Test graph recommendation for advanced graphs."""
|
|
|
|
recommendation = RouteComplexityAnalyzer._get_graph_recommendation(20)
|
|
|
|
assert "Advanced" in recommendation
|
|
|
|
|
|
class TestSuggestRouteType:
|
|
"""Test cases for suggest_route_type method."""
|
|
|
|
def test_suggest_graph_for_persistence(self):
|
|
"""Test suggesting graph for persistence needs."""
|
|
|
|
result = RouteComplexityAnalyzer.suggest_route_type({
|
|
"needs_persistence": True
|
|
})
|
|
|
|
assert result == RouteType.GRAPH
|
|
|
|
def test_suggest_graph_for_state_and_conditionals(self):
|
|
"""Test suggesting graph for state + conditionals."""
|
|
|
|
result = RouteComplexityAnalyzer.suggest_route_type({
|
|
"needs_state": True,
|
|
"needs_conditionals": True
|
|
})
|
|
|
|
assert result == RouteType.GRAPH
|
|
|
|
def test_suggest_graph_for_conditionals_not_continuous(self):
|
|
"""Test suggesting graph for conditionals without continuous flow."""
|
|
|
|
result = RouteComplexityAnalyzer.suggest_route_type({
|
|
"needs_conditionals": True,
|
|
"is_continuous": False
|
|
})
|
|
|
|
assert result == RouteType.GRAPH
|
|
|
|
def test_suggest_stream_for_stateless_continuous(self):
|
|
"""Test suggesting stream for stateless continuous."""
|
|
|
|
result = RouteComplexityAnalyzer.suggest_route_type({
|
|
"is_stateless": True,
|
|
"is_continuous": True
|
|
})
|
|
|
|
assert result == RouteType.STREAM
|
|
|
|
def test_suggest_stream_default(self):
|
|
"""Test suggesting stream as default."""
|
|
|
|
result = RouteComplexityAnalyzer.suggest_route_type({})
|
|
|
|
assert result == RouteType.STREAM
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|
|
|