fix(MockHttpSocket): handle response stream errors

src/interceptors/ClientRequest/MockHttpSocket.ts

@@ -239,25 +239,34 @@ export class MockHttpSocket extends MockSocket {
     }
 
     if (response.body) {
-      const reader = response.body.getReader()
-
-      while (true) {
-        const { done, value } = await reader.read()
-
-        if (done) {
-          break
+      try {
+        const reader = response.body.getReader()
+
+        while (true) {
+          console.log('reading chunk...')
+          const { done, value } = await reader.read()
+
+          if (done) {
+            break
+          }
+
+          // Flush the headers upon the first chunk in the stream.
+          // This ensures the consumer will start receiving the response
+          // as it streams in (subsequent chunks are pushed).
+          if (httpHeaders.length > 0) {
+            flushHeaders(value)
+            continue
+          }
+
+          // Subsequent body chukns are push to the stream.
+          this.push(value)
         }
-
-        // Flush the headers upon the first chunk in the stream.
-        // This ensures the consumer will start receiving the response
-        // as it streams in (subsequent chunks are pushed).
-        if (httpHeaders.length > 0) {
-          flushHeaders(value)
-          continue
+      } catch (error) {
+        if (error instanceof Error) {
+          // Forward response streaming errors as response errors.
+          this.errorWith(error)
         }
-
-        // Subsequent body chukns are push to the stream.
-        this.push(value)
+        return
       }
     }
 

test/modules/http/compliance/http-res-readable.test.ts

@@ -0,0 +1,57 @@
+/**
+ * @vitest-environment node
+ */
+import { vi, it, expect, beforeAll, afterAll } from 'vitest'
+import http from 'node:http'
+import { ClientRequestInterceptor } from '../../../../src/interceptors/ClientRequest'
+import { waitForClientRequest } from '../../../helpers'
+
+const interceptor = new ClientRequestInterceptor()
+
+beforeAll(() => {
+  interceptor.apply()
+})
+
+afterAll(() => {
+  interceptor.dispose()
+})
+
+it('supports ReadableStream as a mocked response', async () => {
+  const encoder = new TextEncoder()
+  interceptor.once('request', ({ request }) => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(encoder.encode('hello'))
+        controller.enqueue(encoder.encode(' '))
+        controller.enqueue(encoder.encode('world'))
+        controller.close()
+      },
+    })
+    request.respondWith(new Response(stream))
+  })
+
+  const request = http.get('http://example.com/resource')
+  const { text } = await waitForClientRequest(request)
+  expect(await text()).toBe('hello world')
+})
+
+it('forwards ReadableStream errors to the request', async () => {
+  interceptor.once('request', ({ request }) => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('hello'))
+        controller.error(new Error('stream error'))
+      },
+    })
+    request.respondWith(new Response(stream))
+  })
+
+  const request = http.get('http://example.com/resource')
+  const requestError = await vi.waitFor(() => {
+    return new Promise<Error>((resolve) => {
+      request.on('error', resolve)
+    })
+  })
+
+  expect(requestError).toEqual(new Error('stream error'))
+})