test: add tests for some stream.Readable uses

* test: check invalid chunk error for readable.push

  Test that passing invalid chunks to readable.push() in
  non-object mode throw errors.

* test: add simple object mode + decoder stream test

* test: add test for readable stream lacking _read

  Check that using a readable stream without a _read method will throw
  an error.

* test: add basic test for piping to multiple dests

  Add a simple test for piping and unpiping from a readable stream
  to multiple writable streams.

PR-URL: #7260
Reviewed-By: Matteo Collina <[email protected]>

Author: addaleax

test/parallel/test-stream-decoder-objectmode.js

@@ -0,0 +1,19 @@
+'use strict';
+require('../common');
+const stream = require('stream');
+const assert = require('assert');
+
+const readable = new stream.Readable({
+  read: () => {},
+  encoding: 'utf16le',
+  objectMode: true
+});
+
+readable.push(Buffer.from('abc', 'utf16le'));
+readable.push(Buffer.from('def', 'utf16le'));
+readable.push(null);
+
+// Without object mode, these would be concatenated into a single chunk.
+assert.strictEqual(readable.read(), 'abc');
+assert.strictEqual(readable.read(), 'def');
+assert.strictEqual(readable.read(), null);

test/parallel/test-stream-pipe-multiple-pipes.js

@@ -0,0 +1,51 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+const assert = require('assert');
+
+const readable = new stream.Readable({
+  read: () => {}
+});
+
+const writables = [];
+
+for (let i = 0; i < 5; i++) {
+  const target = new stream.Writable({
+    write: common.mustCall((chunk, encoding, callback) => {
+      target.output.push(chunk);
+      callback();
+    }, 1)
+  });
+  target.output = [];
+
+  target.on('pipe', common.mustCall(() => {}));
+  readable.pipe(target);
+
+
+  writables.push(target);
+}
+
+const input = Buffer.from([1, 2, 3, 4, 5]);
+
+readable.push(input);
+
+// The pipe() calls will postpone emission of the 'resume' event using nextTick,
+// so no data will be available to the writable streams until then.
+process.nextTick(common.mustCall(() => {
+  for (const target of writables) {
+    assert.deepStrictEqual(target.output, [input]);
+
+    target.on('unpipe', common.mustCall(() => {}));
+    readable.unpipe(target);
+  }
+
+  readable.push('something else'); // This does not get through.
+  readable.push(null);
+  readable.resume(); // Make sure the 'end' event gets emitted.
+}));
+
+readable.on('end', common.mustCall(() => {
+  for (const target of writables) {
+    assert.deepStrictEqual(target.output, [input]);
+  }
+}));

test/parallel/test-stream-readable-invalid-chunk.js

@@ -0,0 +1,12 @@
+'use strict';
+require('../common');
+const stream = require('stream');
+const assert = require('assert');
+
+const readable = new stream.Readable({
+  read: () => {}
+});
+
+assert.throws(() => readable.push([]), /Invalid non-string\/buffer chunk/);
+assert.throws(() => readable.push({}), /Invalid non-string\/buffer chunk/);
+assert.throws(() => readable.push(0), /Invalid non-string\/buffer chunk/);

test/parallel/test-stream-readable-with-unimplemented-_read.js

@@ -0,0 +1,8 @@
+'use strict';
+require('../common');
+const stream = require('stream');
+const assert = require('assert');
+
+const readable = new stream.Readable();
+
+assert.throws(() => readable.read(), /not implemented/);