test(langgraph): Add missing PureGraph BDD and integration coverage
Add comprehensive test coverage for PureGraph module: - Created features/pure_graph_coverage.feature with BDD scenarios for topological ordering, function execution, missing function handling, and non-functional nodes - Added robot/langgraph/pure_graph.robot with Robot Framework integration tests - Created benchmarks/pure_graph_bench.py with ASV benchmarks for execution throughput and ordering performance This addresses the test infrastructure gap identified in issue #9531 where PureGraph had orphaned step definitions but no feature file, and lacked integration and performance test coverage. ISSUES CLOSED: #9531
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
"""Benchmarks for PureGraph execution throughput and ordering."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType
|
||||
from cleveragents.langgraph.pure_graph import PureGraph
|
||||
|
||||
|
||||
class PureGraphBench:
|
||||
"""Benchmark suite for PureGraph execution performance."""
|
||||
|
||||
params = [10, 50, 100, 500]
|
||||
param_names = ["node_count"]
|
||||
|
||||
def setup(self, node_count: int) -> None:
|
||||
"""Set up benchmark with varying node counts."""
|
||||
self.node_count = node_count
|
||||
self.nodes: dict[str, NodeConfig] = {
|
||||
"start": NodeConfig(name="start", type=NodeType.START),
|
||||
"end": NodeConfig(name="end", type=NodeType.END),
|
||||
}
|
||||
self.edges: list[Edge] = []
|
||||
|
||||
# Create a linear chain of nodes
|
||||
previous = "start"
|
||||
for i in range(node_count):
|
||||
node_name = f"node_{i}"
|
||||
self.nodes[node_name] = NodeConfig(
|
||||
name=node_name,
|
||||
type=NodeType.FUNCTION,
|
||||
function=f"fn_{i}"
|
||||
)
|
||||
self.edges.append(Edge(source=previous, target=node_name))
|
||||
previous = node_name
|
||||
|
||||
self.edges.append(Edge(source=previous, target="end"))
|
||||
|
||||
# Create function registry with identity functions
|
||||
self.fn_registry: dict[str, Any] = {
|
||||
f"fn_{i}": lambda x, i=i: x for i in range(node_count)
|
||||
}
|
||||
|
||||
self.graph = PureGraph(
|
||||
name=f"bench_graph_{node_count}",
|
||||
nodes=self.nodes,
|
||||
edges=self.edges
|
||||
)
|
||||
|
||||
def time_topological_order(self, node_count: int) -> None:
|
||||
"""Benchmark topological ordering."""
|
||||
self.graph.topological_order()
|
||||
|
||||
def time_execute(self, node_count: int) -> None:
|
||||
"""Benchmark graph execution."""
|
||||
self.graph.execute(self.fn_registry, initial=0)
|
||||
|
||||
def time_execute_with_result(self, node_count: int) -> None:
|
||||
"""Benchmark graph execution with result accumulation."""
|
||||
result = self.graph.execute(self.fn_registry, initial=42)
|
||||
assert result == 42, f"Expected 42, got {result}"
|
||||
Reference in New Issue
Block a user