add tests

test/parallel/test-stream-duplex-from.js

@@ -401,3 +401,75 @@ function makeATestWritableStream(writeFunc) {
     assert.strictEqual(d.writable, false);
   }));
 }
+
+{
+  const r = Readable.from(['foo', 'bar', 'baz']);
+  pipeline(
+    r,
+    Duplex.from(async function (asyncGenerator) {
+      const res = await asyncGenerator.next();
+      assert.strictEqual(res.done, false);
+      assert.strictEqual(res.value.toString(), 'foo');
+      await asyncGenerator.return();
+    }),
+    common.mustCall((err) => {
+      assert.ifError(err);
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}
+
+{
+  const r = Readable.from(['foo', 'bar', 'baz']);
+  pipeline(
+    r,
+    Duplex.from(async function (asyncGenerator) {
+      // Note: the generator is not even started at this point
+      await asyncGenerator.return();
+    }),
+    common.mustCall((err) => {
+      assert.ifError(err);
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}
+
+{
+  const r = Readable.from(['foo', 'bar', 'baz']);
+  pipeline(
+    r,
+    Duplex.from(async function (asyncGenerator) {}),
+    common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}
+
+{
+  const r = Readable.from(['foo']);
+  pipeline(
+    r,
+    Duplex.from(async function (asyncGenerator) {
+      await asyncGenerator.throw(new Error('my error'));
+    }),
+    common.mustCall((err) => {
+      assert.strictEqual(err.message, 'my error');
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}
+
+{
+  const r = Readable.from(['foo', 'bar']);
+  pipeline(
+    r,
+    Duplex.from(async function (asyncGenerator) {
+      for await (const _ of asyncGenerator) break;
+    }),
+    common.mustCall((err) => {
+      assert.ifError(err);
+      assert.strictEqual(r.destroyed, true);
+    })
+  );
+}