stream: allow pass stream class to stream.compose

PR-URL: nodejs#50187
Fixes: nodejs#50176
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

lib/internal/streams/duplexify.js

@@ -85,6 +85,11 @@ module.exports = function duplexify(body, name) {
   if (typeof body === 'function') {
     const { value, write, final, destroy } = fromAsyncGen(body);
 
+    // Body might be a constructor function instead of an async generator function.
+    if (isDuplexNodeStream(value)) {
+      return value;
+    }
+
     if (isIterable(value)) {
       return from(Duplexify, value, {
         // TODO (ronag): highWaterMark?

test/parallel/test-runner-run.mjs

@@ -96,6 +96,16 @@ describe('require(\'node:test\').run', { concurrency: true }, () => {
       assert.match(stringResults[1], /tests 1/);
       assert.match(stringResults[1], /pass 1/);
     });
+
+    it('spec', async () => {
+      const result = await run({
+        files: [join(testFixtures, 'default-behavior/test/random.cjs')]
+      }).compose(spec).toArray();
+      const stringResults = result.map((bfr) => bfr.toString());
+      assert.match(stringResults[0], /this should pass/);
+      assert.match(stringResults[1], /tests 1/);
+      assert.match(stringResults[1], /pass 1/);
+    });
   });
 
   it('should be piped with tap', async () => {