feat(MockInterceptor): allow async reply callbacks

lib/mock/mock-utils.js

@@ -10,6 +10,11 @@ const {
 } = require('./mock-symbols')
 const { buildURL, nop } = require('../core/util')
 const { STATUS_CODES } = require('http')
+const {
+  types: {
+    isPromise
+  }
+} = require('util')
 
 function matchValue (match, value) {
   if (typeof match === 'string') {
@@ -241,14 +246,27 @@ function mockDispatch (opts, handler) {
     handleReply(this[kDispatches])
   }
 
-  function handleReply (mockDispatches) {
+  function handleReply (mockDispatches, _data = data) {
     // fetch's HeadersList is a 1D string array
     const optsHeaders = Array.isArray(opts.headers)
       ? buildHeadersFromArray(opts.headers)
       : opts.headers
-    const responseData = getResponseData(
-      typeof data === 'function' ? data({ ...opts, headers: optsHeaders }) : data
-    )
+    const body = typeof _data === 'function'
+      ? _data({ ...opts, headers: optsHeaders })
+      : _data
+
+    // util.types.isPromise is likely needed for jest.
+    if (isPromise(body)) {
+      // If handleReply is asynchronous, throwing an error
+      // in the callback will reject the promise, rather than
+      // synchronously throw the error, which breaks some tests.
+      // Rather, we wait for the callback to resolve if it is a
+      // promise, and then re-run handleReply with the new body.
+      body.then((newData) => handleReply(mockDispatches, newData))
+      return
+    }
+
+    const responseData = getResponseData(body)
     const responseHeaders = generateKeyValues(headers)
     const responseTrailers = generateKeyValues(trailers)
 

test/jest/issue-1757.test.js

@@ -0,0 +1,61 @@
+'use strict'
+
+const { Dispatcher, setGlobalDispatcher, MockAgent } = require('../..')
+
+/* global expect, it */
+
+class MiniflareDispatcher extends Dispatcher {
+  constructor (inner, options) {
+    super(options)
+    this.inner = inner
+  }
+
+  dispatch (options, handler) {
+    return this.inner.dispatch(options, handler)
+  }
+
+  close (...args) {
+    return this.inner.close(...args)
+  }
+
+  destroy (...args) {
+    return this.inner.destroy(...args)
+  }
+}
+
+const runIf = (condition) => condition ? it : it.skip
+const nodeMajor = Number(process.versions.node.split('.', 1)[0])
+
+runIf(nodeMajor >= 16)('https://github.com/nodejs/undici/issues/1757', async () => {
+  // fetch isn't exported in <16.8
+  const { fetch } = require('../..')
+
+  const mockAgent = new MockAgent()
+  const mockClient = mockAgent.get('http://localhost:3000')
+  mockAgent.disableNetConnect()
+  setGlobalDispatcher(new MiniflareDispatcher(mockAgent))
+
+  mockClient.intercept({
+    path: () => true,
+    method: () => true
+  }).reply(200, async (opts) => {
+    if (opts.body?.[Symbol.asyncIterator]) {
+      const chunks = []
+      for await (const chunk of opts.body) {
+        chunks.push(chunk)
+      }
+
+      return Buffer.concat(chunks)
+    }
+
+    return opts.body
+  })
+
+  const response = await fetch('http://localhost:3000', {
+    method: 'POST',
+    body: JSON.stringify({ foo: 'bar' })
+  })
+
+  expect(response.json()).resolves.toMatchObject({ foo: 'bar' })
+  expect(response.status).toBe(200)
+})

test/mock-agent.js

@@ -2495,6 +2495,82 @@ test('MockAgent - headers in mock dispatcher intercept should be case-insensitiv
   t.end()
 })
 
+// https://github.com/nodejs/undici/issues/1757
+test('MockAgent - reply callback can be asynchronous', { skip: nodeMajor < 16 }, async (t) => {
+  const { fetch } = require('..')
+  const ReadableStream = globalThis.ReadableStream || require('stream/web').ReadableStream
+
+  class MiniflareDispatcher extends Dispatcher {
+    constructor (inner, options) {
+      super(options)
+      this.inner = inner
+    }
+
+    dispatch (options, handler) {
+      return this.inner.dispatch(options, handler)
+    }
+
+    close (...args) {
+      return this.inner.close(...args)
+    }
+
+    destroy (...args) {
+      return this.inner.destroy(...args)
+    }
+  }
+
+  const mockAgent = new MockAgent()
+  const mockClient = mockAgent.get('http://localhost:3000')
+  mockAgent.disableNetConnect()
+  setGlobalDispatcher(new MiniflareDispatcher(mockAgent))
+
+  t.teardown(mockAgent.close.bind(mockAgent))
+
+  mockClient.intercept({
+    path: () => true,
+    method: () => true
+  }).reply(200, async (opts) => {
+    if (opts.body && opts.body[Symbol.asyncIterator]) {
+      const chunks = []
+      for await (const chunk of opts.body) {
+        chunks.push(chunk)
+      }
+
+      return Buffer.concat(chunks)
+    }
+
+    return opts.body
+  }).persist()
+
+  {
+    const response = await fetch('http://localhost:3000', {
+      method: 'POST',
+      body: JSON.stringify({ foo: 'bar' })
+    })
+
+    t.same(await response.json(), { foo: 'bar' })
+  }
+
+  {
+    const response = await fetch('http://localhost:3000', {
+      method: 'POST',
+      body: new ReadableStream({
+        start (controller) {
+          controller.enqueue(new TextEncoder().encode('{"foo":'))
+
+          setTimeout(() => {
+            controller.enqueue(new TextEncoder().encode('"bar"}'))
+            controller.close()
+          }, 100)
+        }
+      }),
+      duplex: 'half'
+    })
+
+    t.same(await response.json(), { foo: 'bar' })
+  }
+})
+
 test('MockAgent - headers should be array of strings', async (t) => {
   const mockAgent = new MockAgent()
   mockAgent.disableNetConnect()