Testing

@threadplane/langgraph ships two purpose-built test doubles, plus an advanced manual-scripting transport for the rare cases that need it:

provideFakeAgent() — a one-call fake backend that runs the real adapter pipeline and streams canned tokens. No server, no LLM. Reach for this first.
mockLangGraphAgent() — a writable-signal mock for component and unit tests, where you set messages(), status(), and the rest directly.
MockAgentTransport — an advanced escape hatch for hand-scripting exact wire event sequences (tool calls, interrupts, multi-batch lifecycles).

Tip: No flaky tests

None of these touch the network, the LLM, or server state. Every test run produces identical results.

See Choosing an adapter → Testing for when to use each test double.

Every spec on this page is written for Vitest with TestBed, and each one runs as written against the adapter in this repository. Swap describe/it/expect for Jasmine if that is your runner — the TestBed wiring is identical.

Fake backend: `provideFakeAgent()`

provideFakeAgent({ tokens, reasoningTokens, delayMs }) wires a fake backend into Angular DI in one call. It exercises the real adapter pipeline — injectAgent(), status transitions, message accumulation — but the wire events are canned, so there is no server and no LLM. Use it for adapter-integration tests, in-browser demos, and offline development.

It drops in exactly where provideAgent() would go:

import { ApplicationConfig } from '@angular/core';
import { provideFakeAgent } from '@threadplane/langgraph';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideFakeAgent({ tokens: ['Hello', ' world'] }),
  ],
};

The component is byte-identical to production — only the provider changes. FakeAgentConfig ({ tokens?, reasoningTokens?, delayMs? }) lives in @threadplane/chat/testing:

Option	Type	Notes
`tokens`	`string[]`	Emitted as streamed text deltas, in order.
`reasoningTokens`	`string[]`	Emitted before text deltas to exercise reasoning UI.
`delayMs`	`number`	Delay between streamed events.

A runnable spec

Wire provideFakeAgent() into TestBed, submit, and assert the streamed tokens accumulate onto the assistant message. Set delayMs: 0 so the stream drains without wall-clock waiting, and await the promise submit() returns — it resolves when the run ends:

import { describe, it, expect, beforeEach } from 'vitest';
import { Component } from '@angular/core';
import { TestBed } from '@angular/core/testing';
import { injectAgent, provideFakeAgent } from '@threadplane/langgraph';
 
@Component({ template: '' })
class ChatHost {
  readonly chat = injectAgent();
}
 
describe('ChatHost with provideFakeAgent', () => {
  beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [ChatHost],
      providers: [provideFakeAgent({ tokens: ['Hello', ' world'], delayMs: 0 })],
    });
  });
 
  it('accumulates streamed tokens onto the assistant message', async () => {
    const fixture = TestBed.createComponent(ChatHost);
    fixture.detectChanges();
 
    await fixture.componentInstance.chat.submit({ message: 'Hi' });
    fixture.detectChanges();
 
    const assistant = fixture.componentInstance.chat.messages().at(-1);
    expect(assistant?.role).toBe('assistant');
    expect(assistant?.content).toBe('Hello world');
    expect(fixture.componentInstance.chat.status()).toBe('idle');
  });
});

Contract mock: `mockLangGraphAgent()`

For component and unit tests where you do not need a streaming pipeline at all, mockLangGraphAgent(initial) returns a LangGraphAgent whose state surface is exposed as writable signals. Set state directly and assert your component reacts — nothing is real.

Every Message carries a delivery field the adapter normally owns, so build the messages you set with staticDelivery(id), which marks a message complete and successful:

import { describe, it, expect } from 'vitest';
import { staticDelivery } from '@threadplane/chat';
import { mockLangGraphAgent } from '@threadplane/langgraph';
 
describe('mockLangGraphAgent', () => {
  it('exposes the agent surface as writable signals', () => {
    const agent = mockLangGraphAgent({ status: 'running' });
    agent.messages.set([
      { id: 'm1', role: 'assistant', content: 'Hello!', delivery: staticDelivery('m1') },
    ]);
 
    expect(agent.messages()[0].content).toBe('Hello!');
    expect(agent.status()).toBe('running');
  });
});

mockLangGraphAgent() extends the neutral mockAgent() from @threadplane/chat — it supplies the neutral Agent-contract writable signals (messages, status, isLoading, error, interrupt, and the rest) plus submit/stop call tracking, then layers the LangGraph-specific signals (langGraphMessages, branch, queue, and so on) on top.

Python: testing the graph

Before testing the Angular side, make sure your agent logic is correct. LangGraph agents are plain Python functions — test them directly with pytest.

import pytest
from langchain_core.messages import HumanMessage
from my_agent.agent import graph
 
@pytest.mark.asyncio
async def test_agent_responds():
    result = await graph.ainvoke(
        {"messages": [HumanMessage(content="Hello")]},
        config={"configurable": {"thread_id": "test_1"}},
    )
    assert len(result["messages"]) >= 2
    assert result["messages"][-1].type == "ai"
 
@pytest.mark.asyncio
async def test_agent_uses_tools():
    result = await graph.ainvoke(
        {"messages": [HumanMessage(content="Search for LangGraph docs")]},
        config={"configurable": {"thread_id": "test_2"}},
    )
    # Verify the agent called the search tool
    tool_messages = [m for m in result["messages"] if m.type == "tool"]
    assert len(tool_messages) > 0

Note: Agent tests are fast

With MemorySaver and a mocked LLM, agent tests run in milliseconds. Use langchain_core.language_models.FakeListChatModel to remove the LLM dependency entirely.

Advanced: `MockAgentTransport`

Lead with provideFakeAgent() for the simple case. Reach for MockAgentTransport only when you need to hand-script exact wire event sequences — tool calls, interrupts, or multi-batch streaming lifecycles that the canned-token fake cannot express.

Register it through provideAgent({ transport }) in the TestBed providers array, then resolve the agent with TestBed.runInInjectionContext(() => injectAgent()).

Warning: Await every emit

MockAgentTransport.stream() is an async generator: emit() only wakes the suspended stream loop, so nothing has reached the signals at the moment it is called. emit(), emitError() and close() therefore return a promise that resolves once the adapter has consumed the batch — await each one before asserting, and keep the promise submit() returns so the spec can await the end of the run. Specs that assert synchronously read stale signals and pass or fail for the wrong reason. flush() waits the same way without emitting anything, for the cases where something other than an emit drove the agent.

These examples also pass throttle: false. The adapter throttles signal writes by 16 ms by default, and the awaited emit does not cover that window.

Basic setup

One messages event carries an assistant chunk; the adapter projects it into the neutral messages() transcript. Message payloads use the LangChain wire shape — { id, type, content }, where type is 'ai', 'human', 'tool', or 'system':

import { describe, it, expect } from 'vitest';
import { TestBed } from '@angular/core/testing';
import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph';
 
describe('basic transport', () => {
  it('projects an emitted assistant chunk into messages()', async () => {
    const transport = new MockAgentTransport();
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'test_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Hi' });
    await transport.emit([
      { type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Hello!' }] },
    ]);
 
    expect(chat.messages().at(-1)?.content).toBe('Hello!');
 
    await transport.close();
    await submitted;
    expect(chat.status()).toBe('idle');
  });
});

Scripted event sequences

Pass event batches to the constructor for sequential playback. Each call to nextBatch() returns the next batch; emit it to advance what the component sees. Re-emitting the same message id replaces its content, which is how a streamed answer grows:

import { describe, it, expect } from 'vitest';
import { TestBed } from '@angular/core/testing';
import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph';
 
describe('streaming lifecycle', () => {
  it('runs from submit through two batches back to idle', async () => {
    const transport = new MockAgentTransport([
      [{ type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Analyzing' }] }],
      [{ type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Analyzing... done.' }] }],
    ]);
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'test_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Explain signals' });
    expect(chat.status()).toBe('running');
    expect(chat.isLoading()).toBe(true);
 
    await transport.emit(transport.nextBatch());
    expect(chat.messages().at(-1)?.content).toBe('Analyzing');
 
    await transport.emit(transport.nextBatch());
    expect(chat.messages().at(-1)?.content).toBe('Analyzing... done.');
 
    await transport.close();
    await submitted;
    expect(chat.status()).toBe('idle');
    expect(chat.isLoading()).toBe(false);
  });
});

Testing interrupts

An interrupt arrives as an interrupt event whose payload sits under an interrupt key — { type: 'interrupt', interrupt: { id, value } }. The adapter normalizes it into the neutral interrupt() signal, where value is unknown, so cast it at the read site (or type it with the BagTemplate generic described in the Interrupts guide):

import { describe, it, expect } from 'vitest';
import { TestBed } from '@angular/core/testing';
import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph';
 
describe('interrupt handling', () => {
  it('surfaces the interrupt payload', async () => {
    const transport = new MockAgentTransport();
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'approval_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Delete my account' });
    await transport.emit([
      {
        type: 'interrupt',
        interrupt: { id: 'approval', value: { action: 'delete_account', risk: 'high' } },
      },
    ]);
    await transport.close();
    await submitted;
 
    const pending = chat.interrupt();
    expect(pending).toBeDefined();
    const value = pending?.value as { action: string; risk: string };
    expect(value.action).toBe('delete_account');
    expect(value.risk).toBe('high');
  });
});

Note: `interrupt` is required here

interrupt is optional on the runtime-neutral Agent contract, because a runtime without human-in-the-loop support omits it — code written against that contract needs interrupt?.() under strictNullChecks. LangGraphAgent narrows it to required, because the LangGraph adapter always provides it, so injectAgent().interrupt() type-checks as a plain call. langGraphInterrupts() carries the raw LangGraph array alongside it.

Resume the paused run with chat.submit({ resume: { approved: true } }), exactly as a component would.

Testing errors and retry

emitError() throws from inside the stream. The adapter classifies the failure through toAgentError(), so error() holds an AgentError with kind, retryable, and status rather than the raw thrown object. Note that the promise submit() returned rejects, so catch it:

import { describe, it, expect } from 'vitest';
import { TestBed } from '@angular/core/testing';
import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph';
 
describe('error handling', () => {
  it('classifies a transport failure', async () => {
    const transport = new MockAgentTransport();
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'test_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Hello' });
    await transport.emitError(new Error('HTTP 500: connection lost'));
    await submitted.catch(() => undefined);
 
    const err = chat.error();
    expect(err?.kind).toBe('server');
    expect(err?.retryable).toBe(true);
    expect(chat.status()).toBe('error');
    expect(chat.isLoading()).toBe(false);
  });
 
  it('clears the error and starts a new run on retry()', async () => {
    const transport = new MockAgentTransport();
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'test_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Hello' });
    await transport.emitError(new Error('HTTP 503: service unavailable'));
    await submitted.catch(() => undefined);
    expect(chat.status()).toBe('error');
 
    const retried = chat.retry();
    expect(chat.error()).toBeUndefined();
    expect(chat.isLoading()).toBe(true);
 
    await transport.close();
    await retried;
  });
});

retry() re-sends the last payload, so an error banner in a component wires straight to agent.retry() — no need to stash the last message yourself.

Testing thread switching

switchThread(null) starts a fresh conversation: the transcript, the value, and any pending interrupt are cleared. Passing a thread id instead loads that thread through the transport getHistory() hook.

import { describe, it, expect } from 'vitest';
import { TestBed } from '@angular/core/testing';
import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph';
 
describe('thread switching', () => {
  it('clears the transcript when switching to a new thread', async () => {
    const transport = new MockAgentTransport();
    TestBed.configureTestingModule({
      providers: [
        provideAgent({ apiUrl: '', assistantId: 'test_agent', transport, throttle: false }),
      ],
    });
    const chat = TestBed.runInInjectionContext(() => injectAgent());
 
    const submitted = chat.submit({ message: 'Hi' });
    await transport.emit([
      { type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Thread A response' }] },
    ]);
    await transport.close();
    await submitted;
    expect(chat.messages().at(-1)?.content).toBe('Thread A response');
 
    chat.switchThread(null);
    await transport.flush();
    expect(chat.messages()).toEqual([]);
  });
});

Test setup workflow

Install dependencies

Make sure @threadplane/langgraph is available in your test environment. MockAgentTransport ships with the main package — no extra install is needed.

Create the transport

Instantiate MockAgentTransport with optional pre-scripted batches for sequential playback, or leave it empty for imperative emit() calls.

Configure the provider

Pass the transport into provideAgent({ ..., transport }) in the TestBed providers array. All other options (assistantId, threadId, onThreadId) work identically to production code.

Resolve the agent in an injection context

Call TestBed.runInInjectionContext(() => injectAgent()) so the adapter can reach Angular's injector for signal creation and cleanup.

Script events, awaiting each one

Use emit() for ad-hoc events, nextBatch() for pre-scripted sequences, and emitError() for failures — awaiting the promise each one returns before asserting.

Assert signal values

Read chat.messages(), chat.status(), chat.interrupt(), and chat.error() to verify the agent reacted, then close() the stream and await the submit promise.

Integration testing

For end-to-end confidence, run tests against a real LangGraph dev server. The LangGraph CLI starts a local server your tests can hit directly.

# Start the dev server
langgraph dev --config langgraph.json
 
# Run Angular tests against it (no MockAgentTransport needed)
ng test --watch=false

Warning: Integration tests are slow

Integration tests hit a real server and, potentially, a real LLM. Reserve them for CI pipelines or pre-release smoke tests. Use provideFakeAgent() or MockAgentTransport for the vast majority of your suite — both run in milliseconds with zero external dependencies.