import { describe, it, expect, beforeEach, vi } from 'vitest'; import { EventEmitter } from 'events'; import { RpcClient } from '../src/transport/RpcClient'; import { RpcError } from '../src/transport/RpcError'; /** * A FramedDuplex stand-in just rich enough for RpcClient: it exposes * the same events (`message`, `close`, `error`) and captures anything * written via `writeMessage` so tests can assert the wire shape. */ class FakeFramed extends EventEmitter { public sent: Buffer[] = []; public closed = false; writeMessage(body: Buffer): boolean { if (this.closed) throw new Error('closed'); this.sent.push(body); return true; } close(): void { if (this.closed) return; this.closed = true; this.emit('close'); } /** Drive a response back to the client; for tests only. */ pushMessage(envelope: unknown): void { this.emit('message', Buffer.from(JSON.stringify(envelope), 'utf8')); } } function setup() { const framed = new FakeFramed(); const client = new RpcClient(framed as unknown as import('../src/transport/framing').FramedDuplex); return { framed, client }; } describe('RpcClient', () => { it('correlates a call with its response by id', async () => { const { framed, client } = setup(); const pending = client.call('server.info', {}); expect(framed.sent.length).toBe(1); const req = JSON.parse(framed.sent[0].toString('utf8')) as { id: number }; framed.pushMessage({ jsonrpc: '2.0', id: req.id, result: { version: '1.0.0', protocolVersion: 1, capabilities: [], vaultRoot: '/v' } }); const result = await pending; expect(result.version).toBe('1.0.0'); }); it('rejects with RpcError when the response is an error envelope', async () => { const { framed, client } = setup(); const pending = client.call('fs.stat', { path: 'missing.md' }); const req = JSON.parse(framed.sent[0].toString('utf8')) as { id: number }; framed.pushMessage({ jsonrpc: '2.0', id: req.id, error: { code: -32010, message: 'no such file' } }); await expect(pending).rejects.toBeInstanceOf(RpcError); try { await pending; } catch (e) { expect((e as RpcError).code).toBe(-32010); expect((e as RpcError).is(-32010 as never)).toBe(true); } }); it('demultiplexes concurrent calls', async () => { const { framed, client } = setup(); const p1 = client.call('fs.stat', { path: 'a.md' }); const p2 = client.call('fs.stat', { path: 'b.md' }); expect(framed.sent.length).toBe(2); const [id1, id2] = framed.sent.map(b => (JSON.parse(b.toString('utf8')) as { id: number }).id); // Deliver out of order. framed.pushMessage({ jsonrpc: '2.0', id: id2, result: null }); framed.pushMessage({ jsonrpc: '2.0', id: id1, result: { type: 'file', mtime: 1, size: 0, mode: 0 } }); const [r1, r2] = await Promise.all([p1, p2]); expect(r2).toBeNull(); expect(r1).not.toBeNull(); }); it('delivers notifications to registered handlers', () => { const { framed, client } = setup(); const handler = vi.fn(); client.onNotification('fs.changed', handler); framed.pushMessage({ jsonrpc: '2.0', method: 'fs.changed', params: { subscriptionId: 's1', path: 'a.md', event: 'modified', mtime: 99 }, }); expect(handler).toHaveBeenCalledTimes(1); expect(handler.mock.calls[0][0]).toMatchObject({ path: 'a.md', event: 'modified' }); }); it('unregisters a notification handler via the returned disposer', () => { const { framed, client } = setup(); const handler = vi.fn(); const off = client.onNotification('fs.changed', handler); off(); framed.pushMessage({ jsonrpc: '2.0', method: 'fs.changed', params: { subscriptionId: 's1', path: 'a.md', event: 'modified' }, }); expect(handler).not.toHaveBeenCalled(); }); it('rejects pending calls when the stream closes', async () => { const { framed, client } = setup(); const pending = client.call('fs.stat', { path: 'a.md' }); framed.close(); await expect(pending).rejects.toBeInstanceOf(RpcError); }); it('fires onClose handlers with no error on a clean close', () => { const { framed, client } = setup(); const cb = vi.fn(); client.onClose(cb); framed.close(); expect(cb).toHaveBeenCalledTimes(1); expect(cb.mock.calls[0][0]).toBeUndefined(); }); it('fires onClose handlers with the error on an abort', () => { const { framed, client } = setup(); const cb = vi.fn(); client.onClose(cb); framed.emit('error', new Error('boom')); expect(cb).toHaveBeenCalledTimes(1); expect((cb.mock.calls[0][0] as Error).message).toBe('boom'); }); it('rejects call() after close()', async () => { const { client } = setup(); client.close(); await expect(client.call('server.info', {})).rejects.toBeInstanceOf(RpcError); }); it('ignores malformed server responses without killing the session', async () => { const { framed, client } = setup(); const pending = client.call('server.info', {}); const req = JSON.parse(framed.sent[0].toString('utf8')) as { id: number }; // Garbage message first — should be silently dropped. framed.emit('message', Buffer.from('not-json-at-all', 'utf8')); // Valid response afterwards — promise still resolves. framed.pushMessage({ jsonrpc: '2.0', id: req.id, result: { version: '1.0.0', protocolVersion: 1, capabilities: [], vaultRoot: '' } }); await pending; }); it('isClosed() returns false before close and true after', () => { const { client } = setup(); expect(client.isClosed()).toBe(false); client.close(); expect(client.isClosed()).toBe(true); }); it('rejects the call when writeMessage throws', async () => { const { framed, client } = setup(); vi.spyOn(framed, 'writeMessage').mockImplementationOnce(() => { throw new Error('write failed'); }); await expect(client.call('server.info', {})).rejects.toThrow('write failed'); expect(client.isClosed()).toBe(false); }); it('onClose disposer removes the handler so it is not called on close', () => { const { framed, client } = setup(); const cb = vi.fn(); const off = client.onClose(cb); off(); framed.close(); expect(cb).not.toHaveBeenCalled(); }); });