Testing Nodes, Routers, Reducers, and Whole Graphs

Recall your LangChain course’s own testing module — fast, free, mocked tests for logic; slower, real integration tests for genuine model behavior. That exact same discipline applies here, across every real mechanism this course has taught: nodes, routers, reducers, loops, interrupts, recovery, and subgraphs.

flowchart TD
    A[Nodes] --> E[Fast, free, run constantly]
    B[Routers] --> E
    C[Reducers] --> E
    D[Loop termination] --> E
    F[Interrupts] --> G[Need a real checkpointer]
    H[Recovery] --> G
    I[Subgraphs] --> J[Test in complete isolation]
    K[Full graph] --> L["@pytest.mark.integration —\nreal model calls, run deliberately"]

Testing a node in isolation

def check_refund_policy(state: dict) -> dict:
    return {"eligible": state["days_since_purchase"] <= 30}

def test_check_refund_policy_eligible():
    result = check_refund_policy({"days_since_purchase": 10})
    assert result["eligible"] is True

def test_check_refund_policy_not_eligible():
    result = check_refund_policy({"days_since_purchase": 45})
    assert result["eligible"] is False

A node is genuinely just a Python function taking state and returning a dictionary — recall Module 4’s own honest definition. Test it exactly like any other function, with no graph involved at all.

Testing a router

def route_by_category(state: dict) -> str:
    return f"{state['category']}_response"

def test_route_by_category_billing():
    assert route_by_category({"category": "billing"}) == "billing_response"

def test_route_by_category_technical():
    assert route_by_category({"category": "technical"}) == "technical_response"

Recall Module 8’s own clean separation — since routing logic is a genuine, standalone function, it tests exactly this simply, with no need to run the surrounding graph at all.

Testing a reducer

def merge_dicts(existing: dict, new: dict) -> dict:
    return {**existing, **new}

def test_merge_dicts_combines_keys():
    result = merge_dicts({"a": 1}, {"b": 2})
    assert result == {"a": 1, "b": 2}

def test_merge_dicts_new_value_overwrites_shared_key():
    result = merge_dicts({"a": 1}, {"a": 2})
    assert result == {"a": 2}

Recall Module 11’s own core warning about reducer argument order — this second test specifically verifies that order is genuinely correct, catching exactly the class of silent, confusing bug that warning was written to prevent.

Testing loop termination

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    draft: str
    retry_count: int

def revise(state: State) -> dict:
    return {"draft": state["draft"] + " more.", "retry_count": state["retry_count"] + 1}

def route(state: State) -> str:
    return "END" if state["retry_count"] >= 3 else "revise"

def test_loop_terminates_within_bound():
    builder = StateGraph(State)
    builder.add_node("revise", revise)
    builder.add_conditional_edges("revise", route, {"END": END, "revise": "revise"})
    builder.add_edge(START, "revise")
    graph = builder.compile()

    result = graph.invoke({"draft": "", "retry_count": 0})
    assert result["retry_count"] == 3

Recall Module 13’s real, documented AutoGPT concern — this test exists specifically to catch a regression where a bounded loop’s exit condition genuinely stops bounding it, before that regression ever reaches production.

Testing an interrupt

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def request_approval(state: State) -> dict:
    approved = interrupt({"question": "Approve?"})
    return {"draft": "approved" if approved else "rejected"}

def test_interrupt_pauses_and_resumes_correctly():
    builder = StateGraph(State)
    builder.add_node("request_approval", request_approval)
    builder.add_edge(START, "request_approval")
    builder.add_edge("request_approval", END)
    graph = builder.compile(checkpointer=InMemorySaver())
    config = {"configurable": {"thread_id": "test-thread"}}

    first_result = graph.invoke({"draft": "", "retry_count": 0}, config=config)
    assert "__interrupt__" in first_result

    final_result = graph.invoke(Command(resume=True), config=config)
    assert final_result["draft"] == "approved"

Recall Module 19’s own two-call pattern directly — a real, genuine test of the interrupt mechanism has to mirror it exactly: one call that pauses, one call that resumes, checking both real, distinct outcomes.

Testing recovery from a crash

def test_graph_resumes_after_simulated_crash():
    call_count = {"n": 0}

    def flaky_step(state: State) -> dict:
        call_count["n"] += 1
        if call_count["n"] == 1:
            raise RuntimeError("Simulated crash")
        return {"draft": "completed"}

    builder = StateGraph(State)
    builder.add_node("flaky_step", flaky_step)
    builder.add_edge(START, "flaky_step")
    builder.add_edge("flaky_step", END)
    graph = builder.compile(checkpointer=InMemorySaver())
    config = {"configurable": {"thread_id": "crash-test"}}

    try:
        graph.invoke({"draft": "", "retry_count": 0}, config=config)
    except RuntimeError:
        pass

    result = graph.invoke({"draft": "", "retry_count": 0}, config=config)
    assert result["draft"] == "completed"

Recall Module 16’s real crash-and-recovery example — this is genuinely that same test, formalized: force a real failure once, then verify the graph, resumed against the same thread, actually produces the correct, completed result.

Testing a subgraph independently

def test_billing_subgraph_in_isolation():
    result = billing_subgraph.invoke({"query": "test", "billing_result": ""})
    assert "Credit applied" in result["billing_result"]

Recall Module 22’s own real payoff, directly — billing_subgraph tests completely on its own, with zero dependency on classify, technical logic, or refund logic, exactly the modularity subgraphs were built to provide.

Testing the whole graph, end to end

@pytest.mark.integration
def test_full_resolution_graph_handles_billing_query():
    result = resolution_graph.invoke({"query": "Why was I charged twice?", "category": "", "billing_result": ""})
    assert result["category"] == "billing"
    assert result["billing_result"] != ""

Recall @pytest.mark.integration directly from your LangChain course’s own testing discipline — this test genuinely calls a real model and exercises the complete, real graph, reserved for less frequent, deliberate runs rather than every single code change.

Common mistakes worth avoiding

Only ever testing the whole graph, never its individual pieces. Recall this entire module’s own structure — a bug in one specific reducer, buried inside a full end-to-end test, is genuinely hard to isolate. Testing each real piece separately, as this module did throughout, makes failures immediately traceable to their actual, specific cause.

Forgetting that interrupt and recovery tests genuinely need a real checkpointer. Recall the interrupt and recovery tests above — both require checkpointer=InMemorySaver() at compile time; without it, these specific tests can’t meaningfully exercise the real behavior they’re meant to verify.

Running expensive, real integration tests on every single save. Recall your LangChain course’s own honest cost lesson — reserve @pytest.mark.integration tests for deliberate, periodic runs, not continuous, automatic execution on every code change.

What you should take away from this module

• Nodes, routers, and reducers are all genuinely just plain Python functions — test them exactly that simply, with zero graph involved.
• Loop termination, interrupts, and crash recovery all have real, concrete, testable behaviors — verify them directly, don’t just assume they work.
• Subgraphs test in complete isolation, exactly the payoff Module 22 promised.
• The same fast-mocked/slow-integration split from your LangChain course applies identically here.

Where this goes next

The next module gathers every production-relevant lesson from this entire course into one place: Production LangGraph Architecture — the full, real system a genuine deployment actually needs.