import { describe, it, expect, beforeAll, afterAll, beforeEach } from '@jest/globals'; import { ChannelCredentials, Server, ServerCredentials, type Metadata, type ServerWritableStream, } from '@grpc/grpc-js'; import { ChatServiceClient, ChatServiceService, type ChatEvent, type ChatRequest, } from '../gen/apf-ai/chat'; import { ChatClient } from './chat.client'; import { GrpcMetadataBuilder } from './grpc-metadata.builder'; import type { AiServiceConfig } from '../../config/check-ai-service-config'; /** * Integration test: drives `ChatClient` against an in-process fake * gRPC ChatService that emits a canned `ChatEvent` sequence. The * spec covers ADR-0024's confirmation criteria: * * - happy-path stream emits every event the server sent and ends * after `done`; * - metadata (`x-client-id` + `x-correlation-id`) reaches the * server unchanged; * - browser-close (modelled here as an `AbortController.abort()`) * propagates as `call.cancel()` and the server's stream callback * observes the cancellation. */ const CONFIG: AiServiceConfig = { endpoint: '', clientId: 'apf-portal-test', useTls: false, }; interface ServerObservations { request: ChatRequest | null; metadata: Metadata | null; cancelled: boolean; } let server: Server; let port: number; let serverObservations: ServerObservations; let chatHandler: (call: ServerWritableStream) => void; beforeAll(async () => { server = new Server(); server.addService(ChatServiceService, { chat: (call: ServerWritableStream) => { serverObservations.request = call.request; serverObservations.metadata = call.metadata; call.on('cancelled', () => { serverObservations.cancelled = true; }); chatHandler(call); }, }); port = await new Promise((resolve, reject) => { server.bindAsync('127.0.0.1:0', ServerCredentials.createInsecure(), (err, bound) => { if (err) { reject(err); return; } resolve(bound); }); }); }); afterAll(async () => { await new Promise((resolve, reject) => { server.tryShutdown((err) => { if (err) { reject(err); return; } resolve(); }); }); }); beforeEach(() => { serverObservations = { request: null, metadata: null, cancelled: false }; }); function buildClient(): ChatClient { const config = { ...CONFIG, endpoint: `127.0.0.1:${port}` }; const grpc = new ChatServiceClient(config.endpoint, ChannelCredentials.createInsecure()); return new ChatClient(grpc, new GrpcMetadataBuilder(config)); } async function collect(stream: AsyncIterable): Promise { const events: ChatEvent[] = []; for await (const event of stream) { events.push(event); } return events; } describe('ChatClient (in-process fake server)', () => { it('emits every server event then terminates on done', async () => { chatHandler = (call) => { call.write({ token: { token: 'hello', value: 'Hello' } }); call.write({ token: { token: ' world', value: ' world' } }); call.write({ done: { stats: { tokensIn: 1, tokensOut: 2, chunksRetrieved: 0 } } }); call.end(); }; const client = buildClient(); const stream = client.chat({ messages: [], conversationId: 'conv-1', model: '', provider: '', toolsAvailable: [], principal: { subject: 'subj-1', roles: ['admin'], attributes: { tenantId: 't' } }, }); const events = await collect(stream); expect(events).toHaveLength(3); expect(events[0]?.token?.value).toBe('Hello'); expect(events[2]?.done?.stats?.tokensOut).toBe(2); }); it('propagates x-client-id and x-correlation-id metadata to the server', async () => { chatHandler = (call) => { call.write({ done: { stats: { tokensIn: 0, tokensOut: 0, chunksRetrieved: 0 } } }); call.end(); }; const client = buildClient(); await collect( client.chat( { messages: [], conversationId: 'conv-2', model: '', provider: '', toolsAvailable: [], principal: { subject: 'subj-2', roles: [], attributes: {} }, }, { correlationId: 'corr-from-test' }, ), ); expect(serverObservations.metadata?.get('x-client-id')).toEqual(['apf-portal-test']); expect(serverObservations.metadata?.get('x-correlation-id')).toEqual(['corr-from-test']); }); it('cancels the call when the AbortSignal aborts mid-stream', async () => { const cancellationObserved = new Promise((resolve) => { chatHandler = (call) => { call.write({ token: { token: 't1', value: 't1' } }); call.on('cancelled', () => resolve()); // Deliberately do not `end()` — the test cancels. }; }); const controller = new AbortController(); const client = buildClient(); const stream = client.chat( { messages: [], conversationId: 'conv-3', model: '', provider: '', toolsAvailable: [], principal: { subject: 'subj-3', roles: [], attributes: {} }, }, { signal: controller.signal }, ); // Drain the first event then abort. const iter = stream[Symbol.asyncIterator](); await iter.next(); controller.abort(); await cancellationObserved; expect(serverObservations.cancelled).toBe(true); }); it('terminates the stream without data when the AbortSignal is already aborted', async () => { // The signal aborts before the call dial completes. gRPC-js // cancels locally — depending on timing the server may or may // not observe the call, so the spec only locks the client-side // outcome: the stream ends in error and yields no payload. chatHandler = (call) => { call.on('cancelled', () => { // best-effort end on cancellation so the suite never hangs try { call.end(); } catch { // already torn down } }); }; const controller = new AbortController(); controller.abort(); const client = buildClient(); const stream = client.chat( { messages: [], conversationId: 'conv-4', model: '', provider: '', toolsAvailable: [], principal: { subject: 'subj-4', roles: [], attributes: {} }, }, { signal: controller.signal }, ); const events: ChatEvent[] = []; let caught: unknown; try { for await (const ev of stream) { events.push(ev); } } catch (err) { caught = err; } expect(events).toHaveLength(0); // gRPC-js may surface the cancellation as an error or as a clean // end-of-stream depending on whether the call had time to dial. // Either way, no payload is emitted — that is the contract. if (caught !== undefined) { expect((caught as { code?: number }).code).toBeDefined(); } }); });