fix(MockHttpSocket): exhaust .write() callbacks for mocked requests (#…

…542)

package.json

@@ -186,4 +186,4 @@
       "path": "./node_modules/cz-conventional-changelog"
     }
   }
-}
+}

src/interceptors/ClientRequest/MockHttpSocket.ts

@@ -125,7 +125,8 @@ export class MockHttpSocket extends MockSocket {
   public passthrough(): void {
     const socket = this.createConnection()
 
-    // Write the buffered request body chunks.
+    // Flush the buffered "socket.write()" calls onto
+    // the original socket instance (i.e. write request body).
     // Exhaust the "requestBuffer" in case this Socket
     // gets reused for different requests.
     let writeArgs: NormalizedWriteArgs | undefined
@@ -190,6 +191,7 @@ export class MockHttpSocket extends MockSocket {
       .on('prefinish', () => this.emit('prefinish'))
       .on('finish', () => this.emit('finish'))
       .on('close', (hadError) => this.emit('close', hadError))
+      .on('end', () => this.emit('end'))
   }
 
   /**
@@ -208,6 +210,10 @@ export class MockHttpSocket extends MockSocket {
     this.mockConnect()
     this.responseType = 'mock'
 
+    // 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> = []
 
     httpHeaders.push(
@@ -305,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,
@@ -344,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()
+        },
       })
     }
 

src/interceptors/Socket/MockSocket.test.ts

@@ -1,6 +1,7 @@
 /**
  * @vitest-environment node
  */
+import { Socket } from 'node:net'
 import { vi, it, expect } from 'vitest'
 import { MockSocket } from './MockSocket'
 
@@ -117,6 +118,20 @@ it('calls the "write" on "socket.end()" without any arguments', () => {
   expect(writeCallback).toHaveBeenCalledWith(undefined, undefined, undefined)
 })
 
+it('emits "finished" on .end() without any arguments', async () => {
+  const finishListener = vi.fn()
+  const socket = new MockSocket({
+    write: vi.fn(),
+    read: vi.fn(),
+  })
+  socket.on('finish', finishListener)
+  socket.end()
+
+  await vi.waitFor(() => {
+    expect(finishListener).toHaveBeenCalledTimes(1)
+  })
+})
+
 it('calls the "read" on "socket.read(chunk)"', () => {
   const readCallback = vi.fn()
   const socket = new MockSocket({
@@ -150,43 +165,74 @@ it('calls the "read" on "socket.read(null)"', () => {
   expect(readCallback).toHaveBeenCalledWith(null, undefined)
 })
 
-it('updates the readable/writable state on "socket.end()"', async () => {
+it('updates the writable state on "socket.end()"', async () => {
+  const finishListener = vi.fn()
+  const endListener = vi.fn()
   const socket = new MockSocket({
     write: vi.fn(),
     read: vi.fn(),
   })
+  socket.on('finish', finishListener)
+  socket.on('end', endListener)
 
   expect(socket.writable).toBe(true)
   expect(socket.writableEnded).toBe(false)
   expect(socket.writableFinished).toBe(false)
-  expect(socket.readable).toBe(true)
-  expect(socket.readableEnded).toBe(false)
 
   socket.write('hello')
+  // Finish the writable stream.
   socket.end()
 
   expect(socket.writable).toBe(false)
   expect(socket.writableEnded).toBe(true)
-  expect(socket.readable).toBe(true)
 
+  // The "finish" event is emitted when writable is done.
+  // I.e. "socket.end()" is called.
   await vi.waitFor(() => {
-    socket.once('finish', () => {
-      expect(socket.writableFinished).toBe(true)
-    })
+    expect(finishListener).toHaveBeenCalledTimes(1)
+  })
+  expect(socket.writableFinished).toBe(true)
+})
+
+it('updates the readable state on "socket.push(null)"', async () => {
+  const endListener = vi.fn()
+  const socket = new MockSocket({
+    write: vi.fn(),
+    read: vi.fn(),
   })
+  socket.on('end', endListener)
+
+  expect(socket.readable).toBe(true)
+  expect(socket.readableEnded).toBe(false)
+
+  socket.push('hello')
+  socket.push(null)
+
+  expect(socket.readable).toBe(true)
+  expect(socket.readableEnded).toBe(false)
+
+  // Read the data to free the buffer and
+  // make Socket emit "end".
+  socket.read()
 
   await vi.waitFor(() => {
-    socket.once('end', () => {
-      expect(socket.readableEnded).toBe(true)
-    })
+    expect(endListener).toHaveBeenCalledTimes(1)
   })
+  expect(socket.readable).toBe(false)
+  expect(socket.readableEnded).toBe(true)
 })
 
 it('updates the readable/writable state on "socket.destroy()"', async () => {
+  const finishListener = vi.fn()
+  const endListener = vi.fn()
+  const closeListener = vi.fn()
   const socket = new MockSocket({
     write: vi.fn(),
     read: vi.fn(),
   })
+  socket.on('finish', finishListener)
+  socket.on('end', endListener)
+  socket.on('close', closeListener)
 
   expect(socket.writable).toBe(true)
   expect(socket.writableEnded).toBe(false)
@@ -198,17 +244,21 @@ it('updates the readable/writable state on "socket.destroy()"', async () => {
   expect(socket.writable).toBe(false)
   // The ".end()" wasn't called.
   expect(socket.writableEnded).toBe(false)
+  expect(socket.writableFinished).toBe(false)
   expect(socket.readable).toBe(false)
 
   await vi.waitFor(() => {
-    socket.once('finish', () => {
-      expect(socket.writableFinished).toBe(true)
-    })
+    expect(closeListener).toHaveBeenCalledTimes(1)
   })
 
-  await vi.waitFor(() => {
-    socket.once('end', () => {
-      expect(socket.readableEnded).toBe(true)
-    })
-  })
+  // Neither "finish" nor "end" events are emitted
+  // when you destroy the stream. If you want those,
+  // call ".end()", then destroy the stream.
+  expect(finishListener).not.toHaveBeenCalled()
+  expect(endListener).not.toHaveBeenCalled()
+  expect(socket.writableFinished).toBe(false)
+
+  // The "end" event was never emitted so "readableEnded"
+  // remains false.
+  expect(socket.readableEnded).toBe(false)
 })

src/interceptors/Socket/MockSocket.ts

@@ -22,6 +22,10 @@ export class MockSocket extends net.Socket {
     super()
     this.connecting = false
     this.connect()
+
+    this._final = (callback) => {
+      callback(null)
+    }
   }
 
   public connect() {
@@ -46,13 +50,6 @@ export class MockSocket extends net.Socket {
 
   public push(chunk: any, encoding?: BufferEncoding): boolean {
     this.options.read(chunk, encoding)
-
-    if (chunk !== null) {
-      this.emit('data', chunk)
-    } else {
-      this.emit('end')
-    }
-
-    return true
+    return super.push(chunk, encoding)
   }
 }

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',
@@ -104,22 +104,19 @@ it('supports Readable as the request body', async () => {
 
   const input = ['hello', ' ', 'world', null]
   const readable = new Readable({
-    read: async function() {
+    read: async function () {
       await sleep(10)
       this.push(input.shift())
     },
   })
 
-  readable.pipe(req)
-
-  const { text } = await waitForClientRequest(req)
-  const expectedBody = 'hello world'
+  readable.pipe(request)
 
-  expect(interceptedRequestBody).toHaveBeenCalledWith(expectedBody)
-  expect(await text()).toEqual(expectedBody)
+  await waitForClientRequest(request)
+  expect(interceptedRequestBody).toHaveBeenCalledWith('hello world')
 })
 
-it('calls the callback when writing an empty string', async () => {
+it('calls the write callback when writing an empty string', async () => {
   const request = http.request(httpServer.http.url('/resource'), {
     method: 'POST',
   })
@@ -132,7 +129,7 @@ it('calls the callback when writing an empty string', async () => {
   expect(writeCallback).toHaveBeenCalledTimes(1)
 })
 
-it('calls the callback when writing an empty Buffer', async () => {
+it('calls the write callback when writing an empty Buffer', async () => {
   const request = http.request(httpServer.http.url('/resource'), {
     method: 'POST',
   })
@@ -145,3 +142,56 @@ it('calls the callback when writing an empty Buffer', async () => {
 
   expect(writeCallback).toHaveBeenCalledTimes(1)
 })
+
+it('emits "finish" for a passthrough request', async () => {
+  const prefinishListener = vi.fn()
+  const finishListener = vi.fn()
+  const request = http.request(httpServer.http.url('/resource'))
+  request.on('prefinish', prefinishListener)
+  request.on('finish', finishListener)
+  request.end()
+
+  await waitForClientRequest(request)
+
+  expect(prefinishListener).toHaveBeenCalledTimes(1)
+  expect(finishListener).toHaveBeenCalledTimes(1)
+})
+
+it('emits "finish" for a mocked request', async () => {
+  interceptor.once('request', ({ request }) => {
+    request.respondWith(new Response())
+  })
+
+  const prefinishListener = vi.fn()
+  const finishListener = vi.fn()
+  const request = http.request(httpServer.http.url('/resource'))
+  request.on('prefinish', prefinishListener)
+  request.on('finish', finishListener)
+  request.end()
+
+  await waitForClientRequest(request)
+
+  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')
+})