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 <matteo.collina@gmail.com>

Author: addaleax

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

@@ -0,0 +1,19 @@
+'use strict';
+require('../common');
+conststream=require('stream');
+constassert=require('assert');
+
+constreadable=newstream.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';
+constcommon=require('../common');
+conststream=require('stream');
+constassert=require('assert');
+
+constreadable=newstream.Readable({
+read: ()=>{}
+});
+
+constwritables=[];
+
+for(leti=0;i<5;i++){
+consttarget=newstream.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);
+}
+
+constinput=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(consttargetofwritables){
+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(consttargetofwritables){
+assert.deepStrictEqual(target.output,[input]);
+}
+}));

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

@@ -0,0 +1,12 @@
+'use strict';
+require('../common');
+conststream=require('stream');
+constassert=require('assert');
+
+constreadable=newstream.Readable({
+read: ()=>{}
+});
+
+assert.throws(()=>readable.push([]),/Invalidnon-string\/bufferchunk/);
+assert.throws(()=>readable.push({}),/Invalidnon-string\/bufferchunk/);
+assert.throws(()=>readable.push(0),/Invalidnon-string\/bufferchunk/);

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

@@ -0,0 +1,8 @@
+'use strict';
+require('../common');
+conststream=require('stream');
+constassert=require('assert');
+
+constreadable=newstream.Readable();
+
+assert.throws(()=>readable.read(),/notimplemented/);