stream: add tests to validate possible stream arg scenarios

Author: efekrskl

test/parallel/test-stream-pipeline-validation.js

@@ -0,0 +1,220 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Readable, Writable, PassThrough, pipeline } = require('stream');
+const { ReadableStream, WritableStream, TransformStream } = require('stream/web');
+
+class NonReadableNodeStream extends Writable {
+  _write(chunk, encoding, callback) {
+    callback();
+  }
+}
+
+class NonWritableNodeStream extends Readable {
+  _read() {
+    this.push('x');
+    this.push(null);
+  }
+}
+
+function assertInvalidArg(fn, name) {
+  assert.throws(fn, (err) => {
+    assert.strictEqual(err.code, 'ERR_INVALID_ARG_TYPE');
+    assert.strictEqual(err.message.includes(`The "${name}"`), true);
+    return true;
+  });
+}
+
+// Non-readable Node stream
+{
+  assertInvalidArg(() => {
+    pipeline(
+      new NonReadableNodeStream(), // source
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'source');
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new PassThrough(),
+      new NonReadableNodeStream(), // transform
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'transform[1]');
+
+  pipeline(
+    Readable.from(['x']),
+    new NonReadableNodeStream(), // destination
+    common.mustSucceed(),
+  );
+}
+
+// Non-writable Node stream
+{
+  pipeline(
+    new NonWritableNodeStream(), // source
+    new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      },
+    }),
+    common.mustSucceed(),
+  );
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new NonWritableNodeStream(), // transform
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'transform[0]');
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new NonWritableNodeStream(), // destination
+      common.mustNotCall(),
+    );
+  }, 'destination');
+}
+
+// ReadableStream
+{
+  const chunks = [];
+  pipeline(
+    new ReadableStream({ // source
+      start(controller) {
+        controller.enqueue('x');
+        controller.close();
+      },
+    }),
+    new Writable({
+      write(chunk, encoding, callback) {
+        chunks.push(chunk.toString());
+        callback();
+      },
+    }),
+    common.mustSucceed(() => {
+      assert.deepStrictEqual(chunks, ['x']);
+    }),
+  );
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new ReadableStream({ // transform
+        start(controller) {
+          controller.enqueue('x');
+          controller.close();
+        },
+      }),
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'transform[0]');
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new ReadableStream({ // destination
+        start(controller) {
+          controller.enqueue('x');
+          controller.close();
+        },
+      }),
+      common.mustNotCall(),
+    );
+  }, 'destination');
+}
+
+// TransformStream
+{
+  const source = new TransformStream();
+  const writer = source.writable.getWriter();
+  const seen = [];
+  pipeline(
+    source, // source
+    new Writable({
+      write(chunk, encoding, callback) {
+        seen.push(chunk.toString());
+        callback();
+      },
+    }),
+    common.mustSucceed(() => {
+      assert.deepStrictEqual(seen, ['x']);
+    }),
+  );
+
+  writer.write('x').then(common.mustCall(() => writer.close()));
+
+  const transformed = [];
+  pipeline(
+    Readable.from(['x']),
+    new TransformStream({ // transform
+      transform(chunk, controller) {
+        controller.enqueue(chunk.toString().toUpperCase());
+      },
+    }),
+    new Writable({
+      write(chunk, encoding, callback) {
+        transformed.push(chunk.toString());
+        callback();
+      },
+    }),
+    common.mustSucceed(() => {
+      assert.deepStrictEqual(transformed, ['X']);
+    }),
+  );
+
+  assert.doesNotThrow(() => {
+    pipeline(
+      Readable.from(['x']),
+      new TransformStream({ // destination
+        transform() {},
+      }),
+      () => {},
+    );
+  });
+}
+
+// WritableStream
+{
+  assertInvalidArg(() => {
+    pipeline(
+      new WritableStream({ // source
+        write() {},
+      }),
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'source');
+
+  assertInvalidArg(() => {
+    pipeline(
+      Readable.from(['x']),
+      new WritableStream({ // transform
+        write() {},
+      }),
+      new PassThrough(),
+      common.mustNotCall(),
+    );
+  }, 'transform[0]');
+
+  const values = [];
+  pipeline(
+    Readable.from(['x']),
+    new WritableStream({ // destination
+      write(chunk) {
+        values.push(chunk.toString());
+      },
+    }),
+    common.mustSucceed(() => {
+      assert.deepStrictEqual(values, ['x']);
+    }),
+  );
+}