2. Unit Testing for Agents

Overview

In the previous capsule you saw that agents without tests are time bombs. Non-determinism, silent regression, and the complexity accumulated over 8 modules of development make "trust and deploy" irresponsible. Now for the practical part: where do you start testing a system like the Research Agent?

The answer is unit testing — testing individual components in isolation. Not the full graph, not the agent loop, not the interaction between agents. First the atomic pieces: does this tool return the right thing? Does this routing function decide well? Does this node update the state the way it should? If the individual pieces don't work, the complete system can't work. And the best part: unit tests are fast, deterministic, and cheap. You don't need an LLM running to verify that web_search handles a timeout correctly.

Connection with the module: This capsule covers the first level of the testing pyramid. In capsule 03, you'll climb to the second level: integration testing with a real LLM. In 04, trajectory evaluation — assessing whether the path was correct, not just the result. But everything starts here: solid unit tests are the foundation confidence is built on.

Testing Individual Tools

Tools are the easiest functions to test in an agent because they're pure Python with defined inputs and outputs. Every @tool you created in previous modules has an input type, an output type, and an expected behavior. That's exactly what pytest needs.

Happy path

import pytest
from langchain_core.tools import tool

@tool
def calculate(expression: str) -> str:
    """Evaluate a simple math expression."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return f"Result: {result}"
    except Exception as e:
        return f"Error in expression: {e}"


def test_calculate_basic_operations():
    assert calculate.invoke({"expression": "2 + 3"}) == "Result: 5"
    assert calculate.invoke({"expression": "10 * 4"}) == "Result: 40"


def test_calculate_complex_expression():
    result = calculate.invoke({"expression": "(15 + 5) * 3"})
    assert result == "Result: 60"

Notice how we invoke the tool: calculate.invoke({"expression": "..."}). We don't call the Python function directly — we use .invoke() because that's how LangGraph calls it at runtime. If there's an error in the tool's schema, .invoke() reveals it.

Edge cases and parametrize

An agent's tools receive inputs from the LLM, and the LLM can send anything. @pytest.mark.parametrize lets you cover multiple scenarios in a few lines:

@pytest.mark.parametrize("expression, expected_substring", [
    ("2 + 2", "4"),
    ("10 - 3", "7"),
    ("6 * 7", "42"),
    ("2 ** 10", "1024"),
    ("invalid", "Error"),
    ("", "Error"),
    ("import os", "Error"),
    ("10 / 0", "Error"),
])
def test_calculate_parametrized(expression, expected_substring):
    result = calculate.invoke({"expression": expression})
    assert expected_substring in result

A table of 8 tests in 10 lines. If you add a new edge case, you just add a row.

Error handling with mocks

When a tool makes external calls (APIs, databases), you mock those dependencies to isolate it:

from unittest.mock import patch, MagicMock

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    import requests
    try:
        response = requests.get(
            "https://api.search.com/search",
            params={"q": query}, timeout=10
        )
        response.raise_for_status()
        data = response.json()
        return "\n".join(r["snippet"] for r in data["results"][:3])
    except requests.Timeout:
        return f"Error: timeout searching for '{query}'"
    except requests.HTTPError as e:
        return f"HTTP error: {e.response.status_code}"
    except Exception as e:
        return f"Unexpected error: {e}"


@patch("requests.get")
def test_web_search_timeout(mock_get):
    import requests
    mock_get.side_effect = requests.Timeout("Connection timed out")
    result = web_search.invoke({"query": "langchain agents"})
    assert "timeout" in result.lower()
    assert "langchain agents" in result


@patch("requests.get")
def test_web_search_success(mock_get):
    mock_response = MagicMock()
    mock_response.json.return_value = {
        "results": [{"snippet": "LangGraph is a framework for agents"}]
    }
    mock_response.raise_for_status = MagicMock()
    mock_get.return_value = mock_response
    result = web_search.invoke({"query": "langgraph"})
    assert "LangGraph" in result

The pattern is always the same: mock the external dependency, verify that the tool handles every scenario — success, timeout, HTTP error — and returns a readable string instead of propagating exceptions. An agent that propagates exceptions breaks the agent loop.

Testing Graph Nodes

A node in LangGraph is a function that receives the state and returns a dict with updates. For unit testing, you isolate the node from the LLM with a mock that returns predictable responses:

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

def researcher_node(state):
    """Node that uses the LLM to analyze a query."""
    messages = state["messages"]
    response = model.invoke(
        [SystemMessage(content="You are a researcher. Analyze the query.")] + messages
    )
    return {
        "messages": [response],
        "research_count": state.get("research_count", 0) + 1,
    }


def test_researcher_node_updates_state():
    mock_response = AIMessage(
        content="I need to search for papers on transformers.",
        tool_calls=[{
            "name": "web_search",
            "args": {"query": "transformer architecture papers"},
            "id": "call_123",
        }]
    )
    with patch("__main__.model") as mock_model:
        mock_model.invoke.return_value = mock_response
        result = researcher_node({
            "messages": [HumanMessage(content="Research transformers")],
            "research_count": 0,
        })

    assert len(result["messages"]) == 1
    assert result["research_count"] == 1
    assert result["messages"][0].tool_calls[0]["name"] == "web_search"

The test verifies: (1) the node adds a message to the state, (2) it increments research_count, (3) the message includes the expected tool call. The real LLM is never involved.

Pure logic nodes

Some nodes don't use an LLM — they transform state directly. They're even easier to test:

def quality_gate_node(state):
    """Evaluate whether the research has sufficient quality."""
    sources_count = len(state.get("sources", []))
    has_analysis = bool(state.get("analysis", ""))
    word_count = len(state.get("draft", "").split())

    quality_score = 0.0
    if sources_count >= 3: quality_score += 0.4
    if has_analysis: quality_score += 0.3
    if word_count >= 200: quality_score += 0.3

    return {"quality_score": quality_score, "quality_passed": quality_score >= 0.7}


def test_quality_gate_passes_with_sufficient_data():
    state = {
        "sources": ["paper1", "paper2", "paper3"],
        "analysis": "Transformers revolutionized NLP...",
        "draft": " ".join(["word"] * 250),
    }
    result = quality_gate_node(state)
    assert result["quality_score"] == 1.0
    assert result["quality_passed"] is True


def test_quality_gate_fails_with_insufficient_sources():
    state = {"sources": ["paper1"], "analysis": "Partial", "draft": " ".join(["word"] * 250)}
    result = quality_gate_node(state)
    assert result["quality_score"] == 0.6
    assert result["quality_passed"] is False


def test_quality_gate_handles_empty_state():
    state = {"sources": [], "analysis": "", "draft": ""}
    result = quality_gate_node(state)
    assert result["quality_score"] == 0.0
    assert result["quality_passed"] is False

No mocks, no LLM. State in → state update out.

Testing State Transitions

Conditional edges are the functions that decide the graph's flow: does the agent keep researching or move to synthesis? They receive state, they return a string. They're perfect for unit testing.

Direct routing functions

def route_after_research(state) -> str:
    """Decide whether to keep researching or synthesize."""
    if state.get("iteration_count", 0) >= state.get("max_iterations", 5):
        return "synthesize"
    if state.get("quality_passed", False):
        return "synthesize"
    return "research"


@pytest.mark.parametrize("state, expected_route", [
    ({"iteration_count": 0, "max_iterations": 5, "quality_passed": False}, "research"),
    ({"iteration_count": 5, "max_iterations": 5, "quality_passed": False}, "synthesize"),
    ({"iteration_count": 2, "max_iterations": 5, "quality_passed": True}, "synthesize"),
    ({"iteration_count": 0, "quality_passed": False}, "research"),
    ({}, "research"),
])
def test_route_after_research(state, expected_route):
    assert route_after_research(state) == expected_route

Routing with tool calls

def route_after_llm(state) -> str:
    """Route based on whether the LLM made tool calls."""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "execute_tools"
    return "respond"


def test_route_to_tools_when_tool_calls_present():
    state = {"messages": [
        AIMessage(content="", tool_calls=[{"name": "web_search", "args": {"query": "test"}, "id": "1"}])
    ]}
    assert route_after_llm(state) == "execute_tools"


def test_route_to_respond_when_no_tool_calls():
    state = {"messages": [AIMessage(content="Here's your answer.", tool_calls=[])]}
    assert route_after_llm(state) == "respond"

Supervisor routing

In a multi-agent system, the supervisor decides which worker gets the task. Test it exhaustively, including the case of invalid workers:

def route_supervisor(state) -> str:
    if state.get("task_complete", False):
        return "synthesize"
    next_worker = state.get("next_worker", "researcher")
    valid_workers = {"researcher", "analyst", "writer"}
    if next_worker not in valid_workers:
        return "researcher"
    return next_worker


@pytest.mark.parametrize("state, expected", [
    ({"task_complete": True, "next_worker": "writer"}, "synthesize"),
    ({"task_complete": False, "next_worker": "analyst"}, "analyst"),
    ({"task_complete": False, "next_worker": "invalid_worker"}, "researcher"),
    ({}, "researcher"),
])
def test_route_supervisor(state, expected):
    assert route_supervisor(state) == expected

What happens if the LLM generates a worker name that doesn't exist? Your routing function needs a fallback. Without this test, that bug would reach production.

Testing Stop Conditions

Agents that don't stop are dangerous. An infinite loop burns tokens, time, and money. Stop conditions are your safety net.

def check_should_stop(state) -> str:
    iteration = state.get("iteration_count", 0)
    max_iter = state.get("max_iterations", 10)

    if iteration >= max_iter:
        return "stop"
    if state.get("quality_passed", False):
        return "stop"
    if state.get("consecutive_errors", 0) >= 3:
        return "stop"
    return "continue"


class TestStopConditions:

    def test_stops_at_max_iterations(self):
        assert check_should_stop({"iteration_count": 10, "max_iterations": 10}) == "stop"

    def test_continues_below_max(self):
        assert check_should_stop({"iteration_count": 3, "max_iterations": 10}) == "continue"

    def test_stops_when_quality_passed(self):
        assert check_should_stop({"iteration_count": 2, "quality_passed": True}) == "stop"

    def test_stops_after_consecutive_errors(self):
        assert check_should_stop({"iteration_count": 1, "consecutive_errors": 3}) == "stop"

    def test_continues_with_some_errors(self):
        assert check_should_stop({"iteration_count": 1, "consecutive_errors": 2}) == "continue"

    def test_defaults_when_state_empty(self):
        assert check_should_stop({}) == "continue"

    def test_max_iterations_takes_priority(self):
        state = {"iteration_count": 10, "max_iterations": 10,
                 "quality_passed": False, "consecutive_errors": 0}
        assert check_should_stop(state) == "stop"

The class groups related tests. The last test verifies priority: if you hit max_iterations, you stop even if nothing else says you should.

Quality thresholds

Reuse the evaluate_quality function you defined in the nodes section, and test the three zones: clearly above the threshold, clearly below, and right at the edge:

def test_high_quality_passes():
    state = {
        "sources": [{"url": "a.com", "domain": "a.com"}, {"url": "b.org", "domain": "b.org"},
                    {"url": "c.edu", "domain": "c.edu"}],
        "draft": " ".join(["content"] * 250),
        "analysis": "Deep analysis.",
    }
    result = evaluate_quality(state)
    assert result["quality_passed"] is True

def test_borderline_quality_fails():
    state = {
        "sources": [{"url": "a.com", "domain": "a.com"}, {"url": "b.org", "domain": "b.org"},
                    {"url": "c.edu", "domain": "a.com"}],
        "draft": " ".join(["word"] * 201), "analysis": "",
    }
    result = evaluate_quality(state)
    assert result["quality_score"] == 0.6
    assert result["quality_passed"] is False

The borderline case is the most valuable — it reveals whether your threshold is well calibrated.

Fixtures and Helpers for Agent Testing

When your tests grow, you start repeating setup. pytest fixtures remove that repetition.

State fixtures

@pytest.fixture
def empty_state():
    return {
        "messages": [], "iteration_count": 0, "max_iterations": 10,
        "sources": [], "draft": "", "analysis": "",
        "quality_score": 0.0, "quality_passed": False,
    }

@pytest.fixture
def researched_state():
    return {
        "messages": [
            HumanMessage(content="Research transformers in NLP"),
            AIMessage(content="I'm going to research transformers."),
        ],
        "iteration_count": 2, "max_iterations": 10,
        "sources": [
            {"url": "https://arxiv.org/abs/1706.03762", "domain": "arxiv.org"},
            {"url": "https://jalammar.github.io/illustrated-transformer/", "domain": "jalammar.github.io"},
            {"url": "https://arxiv.org/abs/1810.04805", "domain": "arxiv.org"},
        ],
        "draft": "", "analysis": "", "quality_score": 0.0, "quality_passed": False,
    }

@pytest.fixture
def completed_state(researched_state):
    researched_state.update({
        "draft": " ".join(["deep analysis"] * 150),
        "analysis": "Transformers revolutionized NLP.",
        "quality_score": 1.0, "quality_passed": True,
    })
    return researched_state

Usage:

def test_route_continues_from_empty(empty_state):
    assert route_after_research(empty_state) == "research"

def test_route_synthesizes_when_complete(completed_state):
    assert route_after_research(completed_state) == "synthesize"

Mock LLM and conftest.py

Put shared fixtures in conftest.py so they're available across all test files:

# conftest.py
import pytest
from unittest.mock import MagicMock
from langchain_core.messages import AIMessage, HumanMessage

@pytest.fixture
def mock_llm():
    """Configurable mock LLM."""
    mock = MagicMock()
    def configure(content="Default response", tool_calls=None):
        mock.invoke.return_value = AIMessage(content=content, tool_calls=tool_calls or [])
        return mock
    mock.configure = configure
    return mock

@pytest.fixture
def make_tool_call_message():
    """Factory to create AIMessages with tool calls."""
    def _make(tool_name, tool_args, content=""):
        return AIMessage(content=content,
            tool_calls=[{"name": tool_name, "args": tool_args, "id": f"call_{tool_name}"}])
    return _make

tests/
├── conftest.py              # shared fixtures
├── test_tools.py
├── test_nodes.py
├── test_routing.py
└── test_stop_conditions.py

What to Mock and What NOT to Mock

This section is crucial. The most common mistake in agent testing is mocking everything and testing nothing. If you mock the LLM, mock the tools, and mock the state, what are you testing? That your mocks work. That has no value.

┌─────────────────────────────────────────────────────────────┐
│                  WHAT TO MOCK IN UNIT TESTS                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ✓ MOCK                             ✗ DON'T MOCK            │
│  ────                               ──────────              │
│  • LLM (costly, non-deterministic)  • The tool itself       │
│  • External APIs (network)          • Routing functions     │
│  • Databases                        • State transformations │
│  • File system                      • Stop conditions       │
│  • Third-party services             • Quality evaluations   │
│                                                             │
│  Mock external DEPENDENCIES.   Don't mock your own LOGIC.   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

LLM: always mock in unit tests

# ✓ Correct: mock the LLM, test the node's logic
def test_analyst_node_extracts_findings(mock_llm):
    mock_llm.configure(content="Finding 1: Transformers are efficient.")
    with patch("__main__.model", mock_llm):
        result = analyst_node({"messages": [HumanMessage(content="Analyze")], "findings": []})
    assert "Finding" in result["messages"][0].content

# ✗ Wrong: mocking the logic you're supposed to be testing
def test_routing_mocked():
    mock_route = MagicMock(return_value="research")
    assert mock_route({}) == "research"
    # You only verified that MagicMock works. Zero value.

Tools: mock only the external dependencies

# ✓ Correct: mock requests, test the tool
@patch("requests.get")
def test_web_search_parses_results(mock_get):
    mock_get.return_value.json.return_value = {"results": [{"snippet": "LangGraph tutorial"}]}
    mock_get.return_value.raise_for_status = MagicMock()
    result = web_search.invoke({"query": "langgraph"})
    assert "LangGraph" in result

# ✗ Wrong: mocking the tool itself
def test_web_search_mocked():
    mock_tool = MagicMock(return_value="Mock result")
    assert mock_tool({"query": "test"}) == "Mock result"
    # You only tested MagicMock, not web_search.

When NOT to mock the LLM

There's one case: integration tests (capsule 03). The distinction is clear:

If your test takes more than 1 second, it's probably not a unit test.

Connection with the Project

In the module 9 project, you'll apply all of this to the Research Agent:

1. Tests for each tool: web_search, paper_reader, calculator, and MCP tools. Happy path, edge cases, error handling.
2. Tests for nodes: researcher_node, analyst_node, writer_node, quality_gate_node. LLM mocked, verifying state updates.
3. Tests for routing: route_after_research, route_supervisor, route_after_llm. Each routing function with parametrize covering every branch.
4. Tests for stop conditions: max_iterations, quality_passed, consecutive_errors.
5. conftest.py with reusable fixtures: empty_state, researched_state, mock_llm, make_tool_call_message.

research_agent/
├── agent/
│   ├── tools.py
│   ├── nodes.py
│   ├── routing.py
│   └── graph.py
└── tests/
    ├── conftest.py
    ├── test_tools.py
    ├── test_nodes.py
    ├── test_routing.py
    └── test_stop_conditions.py

Troubleshooting

Problem 1: "The test passes locally but fails in CI"

Cause: An environment dependency — API keys, local files, or residual state. Solution: Verify that every external dependency is mocked. Use @pytest.fixture(autouse=True) to clean state between tests.

Problem 2: "patch doesn't intercept the function"

Cause: The patch path is wrong. patch replaces the object where it's used, not where it's defined. Solution: If tools.py imports requests, the patch is @patch("agent.tools.requests.get"), not @patch("requests.get"). Check the exact import path.

Problem 3: "The LLM mock doesn't return tool_calls"

Cause: You forgot to include tool_calls in the mock's AIMessage. Solution: Always create the mock with AIMessage(content="...", tool_calls=[...]). Use the mock_llm.configure() fixture, which handles this automatically.

Problem 4: "TypeError when invoking a tool in a test"

Cause: You're calling tool({"arg": "value"}) instead of tool.invoke({"arg": "value"}). LangChain tools need .invoke() to go through schema validation. Solution: Always use .invoke() in your tests.

Problem 5: "The tests take >10 seconds"

Cause: Some test is making real calls to the LLM or external APIs. Solution: Run pytest --durations=10 to see the 10 slowest tests. If a unit test takes more than 100ms, investigate which dependency isn't mocked.

Exercises

Exercise 1: Test a tool with multiple scenarios (Easy)

Create a tool format_citation(title, authors, year) that generates an APA citation. Write tests with @pytest.mark.parametrize for: valid inputs, negative year, empty authors, and empty title.

@tool
def format_citation(title: str, authors: str, year: int) -> str:
    """Format a citation in APA style."""
    if not title: return "Error: title is required"
    if not authors: return "Error: authors is required"
    if not isinstance(year, int) or year < 0: return f"Error: invalid year ({year})"
    return f"{authors} ({year}). {title}."


@pytest.mark.parametrize("title, authors, year, expected_substring", [
    ("Attention Is All You Need", "Vaswani et al.", 2017, "Vaswani et al. (2017)"),
    ("BERT", "Devlin et al.", 2018, "Devlin et al. (2018). BERT."),
    ("", "Author", 2020, "Error: title"),
    ("Paper", "", 2020, "Error: authors"),
    ("Paper", "Author", -1, "Error: invalid year"),
])
def test_format_citation(title, authors, year, expected_substring):
    result = format_citation.invoke({"title": title, "authors": authors, "year": year})
    assert expected_substring in result

Exercise 2: Test a node with a mocked LLM (Easy)

Write a node summarize_node(state) that generates a summary via the LLM and increments summary_count. Test that the node adds the message, increments the counter, and that the LLM was called with a SystemMessage.

def summarize_node(state):
    response = model.invoke(
        [SystemMessage(content="Summarize the conversation in 2-3 sentences.")]
        + state["messages"]
    )
    return {"messages": [response], "summary_count": state.get("summary_count", 0) + 1}


def test_summarize_node():
    mock_response = AIMessage(content="Summary: transformers were researched and 3 papers were found.")
    with patch("__main__.model") as mock_model:
        mock_model.invoke.return_value = mock_response
        result = summarize_node({
            "messages": [HumanMessage(content="Research transformers")],
            "summary_count": 0,
        })

    assert len(result["messages"]) == 1
    assert "Summary" in result["messages"][0].content
    assert result["summary_count"] == 1
    call_args = mock_model.invoke.call_args[0][0]
    assert isinstance(call_args[0], SystemMessage)

Exercise 3: Test a complex routing function (Medium)

Implement route_after_evaluation(state): quality_score >= 0.8 → "publish", >= 0.5 and revision_count < 3 → "revise", >= 0.5 and revision_count >= 3 → "publish", < 0.5 → "restart". Tests with parametrize for the 4 branches + boundary values.

def route_after_evaluation(state) -> str:
    score = state.get("quality_score", 0.0)
    revisions = state.get("revision_count", 0)
    if score >= 0.8: return "publish"
    if score >= 0.5:
        return "revise" if revisions < 3 else "publish"
    return "restart"


@pytest.mark.parametrize("score, revisions, expected", [
    (0.9, 0, "publish"),
    (0.8, 0, "publish"),       # boundary: exactly 0.8
    (0.8, 5, "publish"),
    (0.7, 0, "revise"),
    (0.5, 2, "revise"),        # boundary: exactly 0.5
    (0.6, 3, "publish"),       # >= 0.5 but no revisions left
    (0.49, 0, "restart"),      # boundary: just below 0.5
    (0.0, 0, "restart"),
])
def test_route_after_evaluation(score, revisions, expected):
    state = {"quality_score": score, "revision_count": revisions}
    assert route_after_evaluation(state) == expected

def test_route_defaults():
    assert route_after_evaluation({}) == "restart"

Exercise 4: Create a conftest.py with fixtures (Medium)

Create a conftest.py with: (1) a research_state fixture with 3 sources and a draft, (2) a mock_llm_with_tools fixture pre-configured with a tool call to web_search, (3) a make_state fixture as a factory with defaults + overrides. Demonstrate its use in 3 tests.

# conftest.py
@pytest.fixture
def research_state():
    return {
        "messages": [HumanMessage(content="Research deep learning")],
        "iteration_count": 2, "max_iterations": 10,
        "sources": [
            {"url": "https://arxiv.org/1", "domain": "arxiv.org"},
            {"url": "https://blog.com/2", "domain": "blog.com"},
            {"url": "https://arxiv.org/3", "domain": "arxiv.org"},
        ],
        "draft": " ".join(["deep learning"] * 120),
        "analysis": "Preliminary analysis.", "quality_score": 0.6, "quality_passed": False,
    }

@pytest.fixture
def mock_llm_with_tools():
    mock = MagicMock()
    mock.invoke.return_value = AIMessage(content="I need more information.",
        tool_calls=[{"name": "web_search", "args": {"query": "deep learning 2025"}, "id": "call_1"}])
    return mock

@pytest.fixture
def make_state():
    def _factory(**overrides):
        base = {"messages": [], "iteration_count": 0, "max_iterations": 10,
                "sources": [], "draft": "", "analysis": "",
                "quality_score": 0.0, "quality_passed": False}
        base.update(overrides)
        return base
    return _factory

# tests
def test_quality_not_passed(research_state):
    result = evaluate_quality(research_state)
    assert result["quality_passed"] is False

def test_factory_with_overrides(make_state):
    state = make_state(iteration_count=5, quality_passed=True)
    assert state["iteration_count"] == 5
    assert state["max_iterations"] == 10  # default preserved

Exercise 5: Complete suite for one component (Hard)

Implement a planner_node that: uses the LLM to generate a plan of sub-questions (JSON), updates state["plan"], deduplicates sub-questions, and increments plan_version. Write a complete suite: happy path, empty plan, invalid JSON, duplicates, version increment, verification of the SystemMessage.

import json

PLANNER_PROMPT = "Generate a plan. Respond with JSON: {\"sub_questions\": [\"q1\", \"q2\", \"q3\"]}"

def planner_node(state):
    response = model.invoke([SystemMessage(content=PLANNER_PROMPT)] + state["messages"])
    try:
        sub_questions = json.loads(response.content).get("sub_questions", [])
    except (json.JSONDecodeError, AttributeError):
        sub_questions = []
    unique_questions = list(dict.fromkeys(sub_questions))
    return {
        "messages": [response],
        "plan": [{"question": q, "status": "pending"} for q in unique_questions],
        "plan_version": state.get("plan_version", 0) + 1,
    }

class TestPlannerNode:
    @pytest.fixture
    def base_state(self):
        return {"messages": [HumanMessage(content="Research LLMs")], "plan": [], "plan_version": 0}

    def test_happy_path(self, base_state):
        plan_json = json.dumps({"sub_questions": ["What are they?", "Applications?", "Limitations?"]})
        with patch("__main__.model") as m:
            m.invoke.return_value = AIMessage(content=plan_json)
            result = planner_node(base_state)
        assert len(result["plan"]) == 3
        assert all(item["status"] == "pending" for item in result["plan"])

    def test_invalid_json(self, base_state):
        with patch("__main__.model") as m:
            m.invoke.return_value = AIMessage(content="I cannot generate a plan.")
            assert planner_node(base_state)["plan"] == []

    def test_deduplicates(self, base_state):
        with patch("__main__.model") as m:
            m.invoke.return_value = AIMessage(
                content=json.dumps({"sub_questions": ["What are they?", "What are they?", "Apps?"]}))
            assert len(planner_node(base_state)["plan"]) == 2

    def test_version_increments(self, base_state):
        base_state["plan_version"] = 3
        with patch("__main__.model") as m:
            m.invoke.return_value = AIMessage(content='{"sub_questions": ["q1"]}')
            assert planner_node(base_state)["plan_version"] == 4

    def test_system_message_instructions(self, base_state):
        with patch("__main__.model") as m:
            m.invoke.return_value = AIMessage(content='{"sub_questions": ["q1"]}')
            planner_node(base_state)
        assert isinstance(m.invoke.call_args[0][0][0], SystemMessage)
        assert "sub_questions" in m.invoke.call_args[0][0][0].content

Summary

In this capsule you learned to write unit tests for every atomic component of an agent:

• Individual tools are tested with .invoke(), covering happy path, edge cases, and error handling. External dependencies get mocked with @patch.
• Graph nodes are tested by mocking the LLM and verifying state updates. The mock returns a predictable AIMessage — the test verifies the node's logic, not the LLM's.
• Routing functions are tested directly: state in → string out. @pytest.mark.parametrize covers dozens of scenarios in a few lines.
• Stop conditions (max iterations, quality thresholds, error counts) need explicit tests because they're the safety net against infinite loops.
• Fixtures in conftest.py remove repetition: empty_state, researched_state, mock_llm, make_state factory.
• What to mock: the LLM and external dependencies. What NOT to mock: the logic you're testing.

Next capsule: Integration Testing — running the full agent loop with a real LLM, snapshot testing of trajectories, and testing the interaction between components you already verified in isolation.

Additional Resources

1. pytest Documentation — Official pytest documentation, including fixtures, parametrize, and plugins
2. unittest.mock — Python Docs — Complete reference for Mock, MagicMock, patch, and side_effect
3. Testing LangChain Applications — Official testing guide for LangChain applications
4. LangGraph — Testing Agents — Testing patterns specific to LangGraph agents
5. Effective Python Testing with pytest — Real Python — A practical pytest tutorial with fixtures and parametrize