tcp-ingestion/test/frame.test.ts

import { describe, it, expect, beforeEach, vi } from 'vitest';
import { EventEmitter } from 'node:events';
import type * as net from 'node:net';
import { BufferedReader, readNextFrame, FrameDropError } from '../src/adapters/teltonika/frame.js';
import { crc16Ibm } from '../src/adapters/teltonika/crc.js';

// ---------------------------------------------------------------------------
// Minimal mock socket
// ---------------------------------------------------------------------------

function makeMockSocket(): net.Socket & { push(buf: Buffer): void; simulateClose(): void } {
  const emitter = new EventEmitter() as net.Socket & {
    push(buf: Buffer): void;
    simulateClose(): void;
    destroyed: boolean;
    remoteAddress?: string;
    remotePort?: number;
    setNoDelay: ReturnType<typeof vi.fn>;
    setKeepAlive: ReturnType<typeof vi.fn>;
    write: ReturnType<typeof vi.fn>;
    destroy: ReturnType<typeof vi.fn>;
  };

  emitter.destroyed = false;
  emitter.remoteAddress = '127.0.0.1';
  emitter.remotePort = 12345;
  emitter.setNoDelay = vi.fn();
  emitter.setKeepAlive = vi.fn();
  emitter.write = vi.fn().mockReturnValue(true);
  emitter.destroy = vi.fn(() => {
    emitter.destroyed = true;
    emitter.emit('close', false);
  });

  emitter.push = (buf: Buffer) => emitter.emit('data', buf);
  emitter.simulateClose = () => {
    emitter.destroyed = true;
    emitter.emit('close', false);
  };

  return emitter;
}

// ---------------------------------------------------------------------------
// BufferedReader tests
// ---------------------------------------------------------------------------

describe('BufferedReader.readExact', () => {
  it('resolves immediately when enough bytes are already buffered', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);

    socket.push(Buffer.from([1, 2, 3, 4, 5]));

    const result = await reader.readExact(3);
    expect(result).toEqual(Buffer.from([1, 2, 3]));
  });

  it('correctly accumulates bytes across multiple data events', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);

    const readPromise = reader.readExact(6);

    // Send data in three separate chunks
    socket.push(Buffer.from([0xaa, 0xbb]));
    socket.push(Buffer.from([0xcc, 0xdd]));
    socket.push(Buffer.from([0xee, 0xff]));

    const result = await readPromise;
    expect(result).toEqual(Buffer.from([0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]));
  });

  it('handles byte-by-byte arrival', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);

    const readPromise = reader.readExact(4);

    for (const byte of [0x01, 0x02, 0x03, 0x04]) {
      socket.push(Buffer.from([byte]));
    }

    const result = await readPromise;
    expect(result).toEqual(Buffer.from([0x01, 0x02, 0x03, 0x04]));
  });

  it('leaves unconsumed bytes available for subsequent reads', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);

    socket.push(Buffer.from([1, 2, 3, 4, 5, 6]));

    const first = await reader.readExact(3);
    const second = await reader.readExact(3);

    expect(first).toEqual(Buffer.from([1, 2, 3]));
    expect(second).toEqual(Buffer.from([4, 5, 6]));
  });

  it('rejects with FrameDropError when socket closes before read completes', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);

    const readPromise = reader.readExact(10);
    socket.push(Buffer.from([1, 2, 3])); // only 3 of 10 bytes
    socket.simulateClose();

    await expect(readPromise).rejects.toBeInstanceOf(FrameDropError);
  });
});

// ---------------------------------------------------------------------------
// Frame builder helper (ESM-safe, no require())
// ---------------------------------------------------------------------------

function buildFrame(options: {
  codecId?: number;
  n1?: number;
  n2?: number;
  records?: Buffer;
  crcOverride?: number;
}): Buffer {
  const codecId = options.codecId ?? 0x08;
  const n1 = options.n1 ?? 1;
  const n2 = options.n2 ?? n1; // default N2 = N1 for valid frames
  const records = options.records ?? Buffer.alloc(0);

  // Body = CodecID (1B) + N1 (1B) + records + N2 (1B)
  const body = Buffer.concat([Buffer.from([codecId, n1]), records, Buffer.from([n2])]);

  const realCrc = crc16Ibm(body);
  const crc = options.crcOverride !== undefined ? options.crcOverride : realCrc;

  const preamble = Buffer.alloc(4, 0);
  const lengthBuf = Buffer.alloc(4);
  lengthBuf.writeUInt32BE(body.length, 0);
  const crcBuf = Buffer.alloc(4);
  crcBuf.writeUInt16BE(0, 0); // upper 2 bytes always 0
  crcBuf.writeUInt16BE(crc, 2);

  return Buffer.concat([preamble, lengthBuf, body, crcBuf]);
}

async function feedFrameToReader(frameBytes: Buffer): Promise<ReturnType<typeof readNextFrame>> {
  const socket = makeMockSocket();
  const reader = new BufferedReader(socket);
  const readPromise = readNextFrame(reader);
  socket.push(frameBytes);
  return readPromise;
}

// ---------------------------------------------------------------------------
// readNextFrame tests
// ---------------------------------------------------------------------------

describe('readNextFrame', () => {
  beforeEach(() => {
    vi.clearAllMocks();
  });

  it('parses a minimal valid frame with correct CRC', async () => {
    const frameBytes = buildFrame({ codecId: 0x08, n1: 1, records: Buffer.alloc(0) });
    const frame = await feedFrameToReader(frameBytes);

    expect(frame.codecId).toBe(0x08);
    expect(frame.crcValid).toBe(true);
  });

  it('reports crcValid=false on CRC mismatch without throwing', async () => {
    const frameBytes = buildFrame({ codecId: 0x08, n1: 1, crcOverride: 0xdead });
    const frame = await feedFrameToReader(frameBytes);

    expect(frame.crcValid).toBe(false);
    expect(frame.expectedCrc).toBe(0xdead);
  });

  it('throws FrameDropError(invalid_preamble) when preamble is non-zero', async () => {
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);
    const readPromise = readNextFrame(reader);

    const badFrame = Buffer.concat([
      Buffer.from([0x00, 0x00, 0x00, 0x01]), // bad preamble
      Buffer.from([0x00, 0x00, 0x00, 0x03]), // length
      Buffer.from([0x08, 0x01, 0x01]), // body
      Buffer.from([0x00, 0x00, 0x00, 0x00]), // CRC
    ]);
    socket.push(badFrame);

    await expect(readPromise).rejects.toMatchObject({
      reason: 'invalid_preamble',
    });
  });

  it('throws FrameDropError(n1_n2_mismatch) when N1 ≠ N2', async () => {
    const frameBytes = buildFrame({ codecId: 0x08, n1: 2, n2: 1 });
    await expect(feedFrameToReader(frameBytes)).rejects.toMatchObject({
      reason: 'n1_n2_mismatch',
    });
  });

  it('handles frame bytes arriving across multiple data events', async () => {
    const frameBytes = buildFrame({ codecId: 0x08, n1: 1 });
    const socket = makeMockSocket();
    const reader = new BufferedReader(socket);
    const readPromise = readNextFrame(reader);

    // Split frame into 3-byte chunks
    for (let offset = 0; offset < frameBytes.length; offset += 3) {
      socket.push(frameBytes.subarray(offset, offset + 3));
    }

    const frame = await readPromise;
    expect(frame.crcValid).toBe(true);
    expect(frame.codecId).toBe(0x08);
  });
});