MockAgentTransport
MockAgentTransport is a deterministic AgentTransport implementation for tests. It records stream calls, lets tests emit LangGraph stream events manually, and avoids a running LangGraph server.
Complete test example
The pattern below covers the full lifecycle: provide a transport instance, create the component, emit stream events, and assert signal state.
import { Component } from '@angular/core';
import { TestBed } from '@angular/core/testing';
import { AgentError } from '@threadplane/chat';
import {
provideAgent,
MockAgentTransport,
injectAgent,
} from '@threadplane/langgraph';
@Component({ template: '' })
class ChatComponent {
readonly chat = injectAgent();
}
describe('ChatComponent', () => {
let transport: MockAgentTransport;
beforeEach(() => {
transport = new MockAgentTransport();
TestBed.configureTestingModule({
imports: [ChatComponent],
providers: [
provideAgent({
apiUrl: '',
assistantId: 'test-agent',
threadId: 'thread-1',
transport,
}),
],
});
});
it('reflects streamed messages', async () => {
const fixture = TestBed.createComponent(ChatComponent);
fixture.detectChanges();
const stream = fixture.componentInstance.chat.submit({ message: 'Hello' });
await transport.emit([
{
type: 'values',
messages: [{ type: 'ai', content: 'Hi there' }],
},
]);
await transport.close();
await stream;
fixture.detectChanges();
// submit() appends the user message optimistically, so the streamed
// assistant reply is the last entry, not the only one.
expect(fixture.componentInstance.chat.messages()).toEqual([
expect.objectContaining({ role: 'user', content: 'Hello' }),
expect.objectContaining({ role: 'assistant', content: 'Hi there' }),
]);
expect(fixture.componentInstance.chat.status()).toBe('idle');
});
it('surfaces errors through the error signal', async () => {
const fixture = TestBed.createComponent(ChatComponent);
fixture.detectChanges();
const stream = fixture.componentInstance.chat.submit({ message: 'Hello' });
await transport.emitError(new Error('not found'));
await stream;
fixture.detectChanges();
expect(fixture.componentInstance.chat.status()).toBe('error');
expect(fixture.componentInstance.chat.error()).toBeInstanceOf(AgentError);
});
});MockAgentTransport API
| Method | Description |
|---|---|
constructor(script?: StreamEvent[][]) | Optionally seeds scripted event batches for manual stepping (defaults to []). |
nextBatch() | Returns the next scripted event batch. |
emit(events) | Pushes one or more StreamEvent objects into the active stream. Returns a promise that resolves once the adapter has consumed the batch. |
emitError(err) | Makes the active stream reject with err. The runtime catches it, so submit() still resolves — assert on status() and error() instead of on a rejected promise. Returns a promise that resolves once the stream has thrown. |
close() | Closes the active stream after queued events drain. Returns a promise that resolves once the run has finished. |
flush() | Resolves once everything emitted so far has been consumed, without emitting anything new. |
isStreaming() | Returns whether a stream is currently active. |
Messages inside a values event are raw LangGraph messages, so their role comes from type ('human', 'ai', 'tool', 'system'), not from a role field. The projector reads _getType() or the raw type and falls back to 'ai'; a role key is never read.
The transport also records calls in streams, createdQueuedRuns, cancelledRuns, joinedRuns, and historyCalls, which is useful for asserting payloads and options.
MockAgentTransport emits only when your test calls emit(), emitError(),
or close(). This makes stream state and payload assertions deterministic.
emit(), emitError(), close() and flush() are asynchronous. stream()
is an async generator, so nothing has reached the signals at the moment
emit() returns. Await the promise before asserting — a spec that reads
signals synchronously after an emit reads stale state and passes or fails for
the wrong reason. flush() covers the same wait when the agent was driven by
something other than an emit.