Testing conditional routing
Why this matters
Conditional routing is where bugs hide in production: a route that works for 90% of inputs but fails on edge cases. Without proper testing, you deploy a graph that silently routes to the wrong node, corrupting downstream processing. Testing conditional edges before production prevents silent failures.
Explanation
Conditional routing in langgraph uses edge functions that inspect the current state and return the name of the next node to visit. Testing this means isolating the routing logic, feeding it controlled state inputs, and verifying it returns the expected node name.
Mechanically: you invoke the graph with carefully crafted state objects, intercept the path the graph takes (using get_state() or streaming the nodes visited), and assert that the execution matched your expected route. The key is testing boundary conditions: the exact state values that flip the routing decision.
At the advanced level, this includes testing routing under partial state (missing optional fields), testing routing with state that causes multiple conditional branches to evaluate, and ensuring your routing logic doesn't accidentally revisit nodes or create infinite loops.
Analogy
Conditional routing is like a railway switch: your train (the execution) can go left or right depending on the track conditions. Testing is manually sending test trains through with known configurations, watching which branch they take, and verifying it matches the switch logic diagram.
Code
import json
from typing import Literal
from dataclasses import dataclass
from langgraph.graph import StateGraph, START, END
import anthropic
@dataclass
class TicketState:
ticket_id: str
priority: str
category: str
resolved: bool = False
current_node: str = ""
def route_ticket(state: TicketState) -> Literal["escalate", "assign", "close"]:
"""Route based on priority and category."""
if state.priority == "critical" and state.category == "billing":
return "escalate"
elif state.priority in ["high", "critical"]:
return "assign"
else:
return "close"
def process_escalate(state: TicketState) -> TicketState:
state.current_node = "escalate"
return state
def process_assign(state: TicketState) -> TicketState:
state.current_node = "assign"
return state
def process_close(state: TicketState) -> TicketState:
state.current_node = "close"
state.resolved = True
return state
builder = StateGraph(TicketState)
builder.add_node("escalate", process_escalate)
builder.add_node("assign", process_assign)
builder.add_node("close", process_close)
builder.add_edge(START, "escalate_decision")
builder.add_node("escalate_decision", lambda state: state)
builder.add_conditional_edges(
"escalate_decision",
route_ticket,
{
"escalate": "escalate",
"assign": "assign",
"close": "close"
}
)
builder.add_edge("escalate", END)
builder.add_edge("assign", END)
builder.add_edge("close", END)
graph = builder.compile()
test_cases = [
{
"name": "Critical billing ticket routes to escalate",
"input": TicketState(ticket_id="T001", priority="critical", category="billing"),
"expected_node": "escalate"
},
{
"name": "High priority non-billing routes to assign",
"input": TicketState(ticket_id="T002", priority="high", category="technical"),
"expected_node": "assign"
},
{
"name": "Low priority routes to close",
"input": TicketState(ticket_id="T003", priority="low", category="general"),
"expected_node": "close"
},
{
"name": "Critical non-billing routes to assign",
"input": TicketState(ticket_id="T004", priority="critical", category="technical"),
"expected_node": "assign"
}
]
results = []
for test in test_cases:
config = {"configurable": {"thread_id": test["input"].ticket_id}}
final_state = graph.invoke(test["input"], config)
actual_node = final_state.current_node
passed = actual_node == test["expected_node"]
results.append({
"test": test["name"],
"expected": test["expected_node"],
"actual": actual_node,
"passed": passed,
"resolved": final_state.resolved
})
for result in results:
status = "✓ PASS" if result["passed"] else "✗ FAIL"
print(f"{status} | {result['test']}")
print(f" Expected: {result['expected']}, Got: {result['actual']}")
print(f" Resolved: {result['resolved']}")
print()
passed_count = sum(1 for r in results if r["passed"])
print(f"\nResults: {passed_count}/{len(results)} tests passed")✓ PASS | Critical billing ticket routes to escalate Expected: escalate, Got: escalate Resolved: False ✓ PASS | High priority non-billing routes to assign Expected: assign, Got: assign Resolved: False ✓ PASS | Low priority routes to close Expected: close, Got: close Resolved: True ✓ PASS | Critical non-billing routes to assign Expected: assign, Got: assign Resolved: False Results: 4/4 tests passed
What just happened?
The code defined a routing function that examines ticket priority and category, then built a langgraph with a conditional edge that calls that routing function. Each test case invoked the graph with a specific TicketState, captured the node name that was visited (stored in state.current_node), and compared it to the expected target node. All four routing decisions executed correctly and the test harness validated each one against the expected outcome.
Common gotcha
The most common mistake is testing only the route_ticket function in isolation instead of testing the actual graph execution. A function might return the correct string, but if your conditional edge mapping is wrong ({"escalate": "assign_node_wrong_name"}), the graph will fail at runtime. Always test the full graph invocation, not just the routing function.
Error recovery
KeyError when calling graph.invoke()ConditionalEdgeValueErrorAssertionError on expected_node mismatchstate.current_node is empty stringExperienced dev note
Senior developers know that conditional routing bugs often lurk at the intersection of two conditions. Your tests pass for 'critical + billing' and 'critical + non-billing' separately, but the actual edge case is the OR/AND logic: which condition wins when both fire? Write tests that enumerate all combinations of your conditional variables (priority × category), not just the happy paths. Also: use `graph.get_state(config)` after invoke to inspect the full state, not just the output: state mutations in nodes often reveal hidden routing bugs that wouldn't show up in the final output.
Check your understanding
If your routing function has three conditions (priority, category, and a new 'is_vip' flag), and you want to test all meaningful paths, how many test cases is the minimum you should write, and what determines that number?
Show answer hint
The answer involves understanding that conditional routing creates a decision tree: you need to test enough cases to cover the distinct outcomes of your routing logic, not just the number of input variables. The minimum is determined by the number of different node names your routing function can return, plus one test per boundary condition where the decision flips (e.g., priority='high' vs 'critical', or category='billing' vs 'technical'). A correct answer would identify that it's not 2^N combinations, but rather the number of unique routing outcomes.