fix(MockHttpSocket): flush write buffer on "requestStream" read

package.json

@@ -86,9 +86,9 @@
   "scripts": {
     "start": "tsc --build -w",
     "test": "pnpm test:unit && pnpm test:integration",
-    "test:unit": "vitest run",
+    "test:unit": "vitest",
     "test:integration": "pnpm test:node && pnpm test:browser",
-    "test:node": "vitest run -c test/vitest.config.js",
+    "test:node": "vitest -c test/vitest.config.js",
     "test:browser": "pnpm playwright test -c test/playwright.config.ts",
     "clean": "rimraf lib",
     "build": "pnpm clean && cross-env NODE_ENV=production tsup --splitting",
@@ -186,4 +186,4 @@
       "path": "./node_modules/cz-conventional-changelog"
     }
   }
-}
+}

src/interceptors/ClientRequest/MockHttpSocket.ts

@@ -210,15 +210,9 @@ export class MockHttpSocket extends MockSocket {
     this.mockConnect()
     this.responseType = 'mock'
 
-    // Exhaust the "socket.write()" callbacks to
-    // transition the Writable into the right state
-    // (i.e. "request body written" state). There's nowhere
-    // to actually write the data so we disregard it.
-    // It has been used by the request parser to construct
-    // a Fetch API Request instance representing this request.
-    for (const [_, __, writeCallback] of this.writeBuffer) {
-      writeCallback?.()
-    }
+    // Flush the write buffer to trigger write callbacks
+    // if it hasn't been flushed already (e.g. someone started reading request stream).
+    this.flushWriteBuffer()
 
     const httpHeaders: Array<Buffer> = []
 
@@ -317,6 +311,13 @@ export class MockHttpSocket extends MockSocket {
     }
   }
 
+  private flushWriteBuffer(): void {
+    let args: NormalizedWriteArgs | undefined
+    while ((args = this.writeBuffer.shift())) {
+      args?.[2]?.()
+    }
+  }
+
   private onRequestStart: RequestHeadersCompleteCallback = (
     versionMajor,
     versionMinor,
@@ -356,7 +357,13 @@ export class MockHttpSocket extends MockSocket {
          * used as the actual request body (the stream calls "read()").
          * We control the queue in the onRequestBody/End functions.
          */
-        read: () => {},
+        read: () => {
+          // If the user attempts to read the request body,
+          // flush the write buffer to trigger the callbacks.
+          // This way, if the request stream ends in the write callback,
+          // it will indeed end correctly.
+          this.flushWriteBuffer()
+        },
       })
     }
 

test/modules/http/compliance/http-req-write.test.ts

@@ -95,7 +95,7 @@ it('writes Buffer request body', async () => {
 })
 
 it('supports Readable as the request body', async () => {
-  const req = http.request(httpServer.http.url('/resource'), {
+  const request = http.request(httpServer.http.url('/resource'), {
     method: 'POST',
     headers: {
       'Content-Type': 'application/json',
@@ -110,13 +110,10 @@ it('supports Readable as the request body', async () => {
     },
   })
 
-  readable.pipe(req)
+  readable.pipe(request)
 
-  const { text } = await waitForClientRequest(req)
-  const expectedBody = 'hello world'
-
-  expect(interceptedRequestBody).toHaveBeenCalledWith(expectedBody)
-  expect(await text()).toEqual(expectedBody)
+  await waitForClientRequest(request)
+  expect(interceptedRequestBody).toHaveBeenCalledWith('hello world')
 })
 
 it('calls the write callback when writing an empty string', async () => {
@@ -177,3 +174,24 @@ it('emits "finish" for a mocked request', async () => {
   expect(prefinishListener).toHaveBeenCalledTimes(1)
   expect(finishListener).toHaveBeenCalledTimes(1)
 })
+
+it('calls all write callbacks before the mocked response', async () => {
+  const requestBodyCallback = vi.fn()
+  interceptor.once('request', async ({ request }) => {
+    requestBodyCallback(await request.text())
+    request.respondWith(new Response('hello world'))
+  })
+
+  const request = http.request(httpServer.http.url('/resource'), {
+    method: 'POST',
+  })
+  request.write('one', () => {
+    console.log('write callback!')
+    request.end()
+  })
+
+  const { text } = await waitForClientRequest(request)
+
+  expect(requestBodyCallback).toHaveBeenCalledWith('one')
+  expect(await text()).toBe('hello world')
+})