smartstream/test/test.streamwrapper.ts

import { expect, tap } from '@push.rocks/tapbundle';
import * as fs from 'fs';
import { StreamWrapper, SmartDuplex } from '../ts/index.js';

tap.test('StreamWrapper: should pipe read to write', async () => {
  const wrapper = new StreamWrapper([
    fs.createReadStream('./test/assets/test.md'),
    fs.createWriteStream('./test/assets/testCopy.md'),
  ]);
  await wrapper.run();
});

tap.test('StreamWrapper: should propagate errors', async (tools) => {
  const failingStream = new SmartDuplex<Buffer, Buffer>({
    writeFunction: async () => {
      throw new Error('intentional error');
    },
  });

  const wrapper = new StreamWrapper([
    fs.createReadStream('./test/assets/test.md'),
    failingStream,
  ]);

  let errorCaught = false;
  try {
    await wrapper.run();
  } catch (err) {
    errorCaught = true;
    expect(err.message).toEqual('intentional error');
  }
  expect(errorCaught).toBeTrue();
});

tap.test('StreamWrapper: streamStarted should resolve', async () => {
  const wrapper = new StreamWrapper([
    fs.createReadStream('./test/assets/test.md'),
    fs.createWriteStream('./test/assets/testCopy.md'),
  ]);

  const runPromise = wrapper.run();
  await wrapper.streamStarted();
  await runPromise;
});

tap.test('StreamWrapper: onCustomEvent should fire', async (tools) => {
  const results: string[] = [];

  const emitter = new SmartDuplex<Buffer, Buffer>({
    writeFunction: async (chunk, streamTools) => {
      (emitter as any).emit('custom-progress', 'progress');
      return chunk;
    },
  });

  const wrapper = new StreamWrapper([
    fs.createReadStream('./test/assets/test.md'),
    emitter,
    fs.createWriteStream('./test/assets/testCopy.md'),
  ]);

  wrapper.onCustomEvent('custom-progress', () => {
    results.push('fired');
  });

  await wrapper.run();
  expect(results.length).toBeGreaterThan(0);
});

export default tap.start();