feat(net-stubbing): throttle/delay for StaticResponses (#8109)

packages/driver/cypress/integration/commands/net_stubbing_spec.ts

@@ -670,6 +670,31 @@ describe('network stubbing', function () {
       cy.contains('#result', '{"foo":1,"bar":{"baz":"cypress"}}').should('be.visible')
     })
 
+    it('can delay and throttle a StaticResponse', function (done) {
+      const payload = 'A'.repeat(10 * 1024)
+      const throttleKbps = 10
+      const delayMs = 250
+      const expectedSeconds = payload.length / (1024 * throttleKbps) + delayMs / 1000
+
+      cy.route2('/timeout', (req) => {
+        this.start = Date.now()
+
+        req.reply({
+          statusCode: 200,
+          body: payload,
+          throttleKbps,
+          delayMs,
+        })
+      }).then(() => {
+        return $.get('/timeout').then((responseText) => {
+          expect(Date.now() - this.start).to.be.closeTo(expectedSeconds * 1000 + 100, 100)
+          expect(responseText).to.eq(payload)
+
+          done()
+        })
+      })
+    })
+
     context('matches requests as expected', function () {
       it('handles querystrings as expected', function () {
         cy.route2({
@@ -1117,7 +1142,7 @@ describe('network stubbing', function () {
       cy.contains('#result', '{"foo":1,"bar":{"baz":"cypress"}}').should('be.visible')
     })
 
-    context('with StaticResponse shorthand', function () {
+    context('with StaticResponse', function () {
       it('res.send(body)', function () {
         cy.route2('/custom-headers', function (req) {
           req.reply((res) => {
@@ -1252,6 +1277,34 @@ describe('network stubbing', function () {
           })
         })
       })
+
+      it('can delay and throttle', function (done) {
+        const payload = 'A'.repeat(10 * 1024)
+        const throttleKbps = 50
+        const delayMs = 50
+        const expectedSeconds = payload.length / (1024 * throttleKbps) + delayMs / 1000
+
+        cy.route2('/timeout', (req) => {
+          req.reply((res) => {
+            this.start = Date.now()
+
+            // ensure .throttle and .delay are overridden
+            res.throttle(1e6).delay(1).send({
+              statusCode: 200,
+              body: payload,
+              throttleKbps,
+              delayMs,
+            })
+          })
+        }).then(() => {
+          return $.get('/timeout').then((responseText) => {
+            expect(responseText).to.eq(payload)
+            expect(Date.now() - this.start).to.be.closeTo(expectedSeconds * 1000 + 50, 50)
+
+            done()
+          })
+        })
+      })
     })
 
     context('errors', function () {

packages/driver/src/cy/net-stubbing/events/index.ts

@@ -38,6 +38,10 @@ export function registerEvents (Cypress: Cypress.Cypress) {
   function emitNetEvent (eventName: string, frame: any): Promise<void> {
     // all messages from driver to server are wrapped in backend:request
     return Cypress.backend('net', eventName, frame)
+    .catch((err) => {
+      err.message = `An error was thrown while processing a network event: ${err.message}`
+      failCurrentTest(err)
+    })
   }
 
   function failCurrentTest (err: Error) {

packages/driver/src/cy/net-stubbing/static-response-utils.ts

@@ -4,18 +4,18 @@ import {
   StaticResponse,
   BackendStaticResponse,
   FixtureOpts,
-  GenericStaticResponse,
 } from '@packages/net-stubbing/lib/types'
 import $errUtils from '../../cypress/error_utils'
 
-export const STATIC_RESPONSE_KEYS: (keyof GenericStaticResponse<void, void>)[] = ['body', 'fixture', 'statusCode', 'headers', 'forceNetworkError']
+// user-facing StaticResponse only
+export const STATIC_RESPONSE_KEYS: (keyof StaticResponse)[] = ['body', 'fixture', 'statusCode', 'headers', 'forceNetworkError', 'throttleKbps', 'delayMs']
 
 export function validateStaticResponse (cmd: string, staticResponse: StaticResponse): void {
   const err = (message) => {
     $errUtils.throwErrByPath('net_stubbing.invalid_static_response', { args: { cmd, message, staticResponse } })
   }
 
-  const { body, fixture, statusCode, headers, forceNetworkError } = staticResponse
+  const { body, fixture, statusCode, headers, forceNetworkError, throttleKbps, delayMs } = staticResponse
 
   if (forceNetworkError && (body || statusCode || headers)) {
     err('`forceNetworkError`, if passed, must be the only option in the StaticResponse.')
@@ -38,6 +38,14 @@ export function validateStaticResponse (cmd: string, staticResponse: StaticRespo
   if (headers && _.keys(_.omitBy(headers, _.isString)).length) {
     err('`headers` must be a map of strings to strings.')
   }
+
+  if (!_.isUndefined(throttleKbps) && (!_.isNumber(throttleKbps) || (throttleKbps < 0 || !_.isFinite(throttleKbps)))) {
+    err('`throttleKbps` must be a finite, positive number.')
+  }
+
+  if (delayMs && (!_.isFinite(delayMs) || delayMs < 0)) {
+    err('`delayMs` must be a finite, positive number.')
+  }
 }
 
 export function parseStaticResponseShorthand (statusCodeOrBody: number | string | any, bodyOrHeaders: string | { [key: string]: string }, maybeHeaders?: { [key: string]: string }) {
@@ -80,7 +88,7 @@ function getFixtureOpts (fixture: string): FixtureOpts {
 }
 
 export function getBackendStaticResponse (staticResponse: Readonly<StaticResponse>): BackendStaticResponse {
-  const backendStaticResponse: BackendStaticResponse = _.omit(staticResponse, 'body', 'fixture')
+  const backendStaticResponse: BackendStaticResponse = _.omit(staticResponse, 'body', 'fixture', 'delayMs')
 
   if (staticResponse.fixture) {
     backendStaticResponse.fixture = getFixtureOpts(staticResponse.fixture)
@@ -95,6 +103,10 @@ export function getBackendStaticResponse (staticResponse: Readonly<StaticRespons
     }
   }
 
+  if (staticResponse.delayMs) {
+    backendStaticResponse.continueResponseAt = Date.now() + staticResponse.delayMs
+  }
+
   return backendStaticResponse
 }
 

packages/net-stubbing/lib/external-types.ts

@@ -178,7 +178,12 @@ export type RouteHandler = string | StaticResponse | RouteHandlerController | ob
 /**
  * Describes a response that will be sent back to the browser to fulfill the request.
  */
-export type StaticResponse = GenericStaticResponse<string, string | object>
+export type StaticResponse = GenericStaticResponse<string, string | object> & {
+  /**
+  * If set, `delayMs` will pass before the response is sent.
+  */
+ delayMs?: number
+}
 
 export interface GenericStaticResponse<Fixture, Body> {
   /**
@@ -201,6 +206,10 @@ export interface GenericStaticResponse<Fixture, Body> {
    * If `forceNetworkError` is truthy, Cypress will destroy the connection to the browser and send no response. Useful for simulating a server that is not reachable. Must not be set in combination with other options.
    */
   forceNetworkError?: boolean
+  /**
+   * If set, the `body` will be sent at `throttleKbps` kbps.
+   */
+  throttleKbps?: number
 }
 
 /**

packages/net-stubbing/lib/internal-types.ts

@@ -10,7 +10,10 @@ export type FixtureOpts = {
   filePath: string
 }
 
-export type BackendStaticResponse = GenericStaticResponse<FixtureOpts, string>
+export type BackendStaticResponse = GenericStaticResponse<FixtureOpts, string> & {
+  // Millisecond timestamp for when the response should continue
+  continueResponseAt?: number
+}
 
 export const SERIALIZABLE_REQ_PROPS = [
   'headers',

packages/net-stubbing/lib/server/intercept-response.ts

@@ -1,8 +1,7 @@
 import _ from 'lodash'
 import concatStream from 'concat-stream'
 import Debug from 'debug'
-import { PassThrough, Readable } from 'stream'
-import ThrottleStream from 'throttle'
+import { Readable } from 'stream'
 
 import {
   ResponseMiddleware,
@@ -19,6 +18,7 @@ import {
   emit,
   sendStaticResponse,
   setBodyFromFixture,
+  getBodyStream,
 } from './util'
 
 const debug = Debug('cypress:net-stubbing:server:intercept-response')
@@ -74,7 +74,7 @@ export const InterceptResponse: ResponseMiddleware = function () {
   }))
 }
 
-export function onResponseContinue (state: NetStubbingState, frame: NetEventFrames.HttpResponseContinue) {
+export async function onResponseContinue (state: NetStubbingState, frame: NetEventFrames.HttpResponseContinue) {
   const backendRequest = state.requests[frame.requestId]
 
   if (typeof backendRequest === 'undefined') {
@@ -85,54 +85,22 @@ export function onResponseContinue (state: NetStubbingState, frame: NetEventFram
 
   debug('_onResponseContinue %o', { backendRequest: _.omit(backendRequest, 'res.body'), frame: _.omit(frame, 'res.body') })
 
-  async function continueResponse () {
-    let newResStream: Readable
+  const throttleKbps = _.get(frame, 'staticResponse.throttleKbps') || frame.throttleKbps
+  const continueResponseAt = _.get(frame, 'staticResponse.continueResponseAt') || frame.continueResponseAt
 
-    function throttleify (body) {
-      const throttleStr = new ThrottleStream(frame.throttleKbps! * 1024)
+  if (frame.staticResponse) {
+    // replacing response with a staticResponse
+    await setBodyFromFixture(backendRequest.route.getFixture, frame.staticResponse)
 
-      throttleStr.write(body)
-      throttleStr.end()
+    const staticResponse = _.chain(frame.staticResponse).clone().assign({ continueResponseAt, throttleKbps }).value()
 
-      return throttleStr
-    }
-
-    if (frame.staticResponse) {
-      await setBodyFromFixture(backendRequest.route.getFixture, frame.staticResponse)
-      const bodyStream = frame.throttleKbps ? throttleify(frame.staticResponse.body) : undefined
-
-      return sendStaticResponse(res, frame.staticResponse, backendRequest.onResponse!, bodyStream)
-    }
-
-    // merge the changed response attributes with our response and continue
-    _.assign(res, _.pick(frame.res, SERIALIZABLE_RES_PROPS))
-
-    function sendBody (bodyBuffer) {
-      // transform the body string into stream format
-      if (frame.throttleKbps) {
-        newResStream = throttleify(bodyBuffer)
-      } else {
-        const pt = new PassThrough()
-
-        pt.write(bodyBuffer)
-        pt.end()
-
-        newResStream = pt
-      }
-
-      backendRequest.continueResponse!(newResStream)
-    }
-
-    return sendBody(res.body)
+    return sendStaticResponse(res, staticResponse, backendRequest.onResponse!)
   }
 
-  if (typeof frame.continueResponseAt === 'number') {
-    const delayMs = frame.continueResponseAt - Date.now()
+  // merge the changed response attributes with our response and continue
+  _.assign(res, _.pick(frame.res, SERIALIZABLE_RES_PROPS))
 
-    if (delayMs > 0) {
-      return setTimeout(continueResponse, delayMs)
-    }
-  }
+  const bodyStream = getBodyStream(res.body, { throttleKbps, continueResponseAt })
 
-  return continueResponse()
+  return backendRequest.continueResponse!(bodyStream)
 }

packages/net-stubbing/lib/server/util.ts

@@ -12,6 +12,7 @@ import { Readable, PassThrough } from 'stream'
 import CyServer from '@packages/server'
 import { Socket } from 'net'
 import { GetFixtureFn } from './types'
+import ThrottleStream from 'throttle'
 
 // TODO: move this into net-stubbing once cy.route is removed
 import { parseContentType } from '@packages/server/lib/controllers/xhrs'
@@ -19,7 +20,10 @@ import { parseContentType } from '@packages/server/lib/controllers/xhrs'
 const debug = Debug('cypress:net-stubbing:server:util')
 
 export function emit (socket: CyServer.Socket, eventName: string, data: object) {
-  debug('sending event to driver %o', { eventName, data })
+  if (debug.enabled) {
+    debug('sending event to driver %o', { eventName, data: _.chain(data).cloneDeep().omit('res.body').value() })
+  }
+
   socket.toDriver('net:event', eventName, data)
 }
 
@@ -117,7 +121,7 @@ export async function setBodyFromFixture (getFixtureFn: GetFixtureFn, staticResp
  * @param onResponse Will be called with the response metadata + body stream
  * @param resStream Optionally, provide a Readable stream to be used as the response body (overrides staticResponse.body)
  */
-export function sendStaticResponse (res: ServerResponse, staticResponse: BackendStaticResponse, onResponse: (incomingRes: IncomingMessage, stream: Readable) => void, resStream?: Readable) {
+export function sendStaticResponse (res: ServerResponse, staticResponse: BackendStaticResponse, onResponse: (incomingRes: IncomingMessage, stream: Readable) => void) {
   if (staticResponse.forceNetworkError) {
     res.connection.destroy()
     res.destroy()
@@ -127,7 +131,7 @@ export function sendStaticResponse (res: ServerResponse, staticResponse: Backend
 
   const statusCode = staticResponse.statusCode || 200
   const headers = staticResponse.headers || {}
-  const body = resStream ? '' : staticResponse.body || ''
+  const body = staticResponse.body || ''
 
   const incomingRes = _getFakeClientResponse({
     statusCode,
@@ -135,17 +139,38 @@ export function sendStaticResponse (res: ServerResponse, staticResponse: Backend
     body,
   })
 
-  if (!resStream) {
-    const pt = new PassThrough()
+  const bodyStream = getBodyStream(body, _.pick(staticResponse, 'throttleKbps', 'continueResponseAt'))
 
-    if (staticResponse.body) {
-      pt.write(staticResponse.body)
+  onResponse(incomingRes, bodyStream)
+}
+
+export function getBodyStream (body: Buffer | string | Readable | undefined, options: { continueResponseAt?: number, throttleKbps?: number }): Readable {
+  const { continueResponseAt, throttleKbps } = options
+  const delayMs = continueResponseAt ? _.max([continueResponseAt - Date.now(), 0]) : 0
+  const pt = new PassThrough()
+
+  const sendBody = () => {
+    let writable = pt
+
+    if (throttleKbps) {
+      // ThrottleStream must be instantiated after any other delays because it uses a `Date.now()`
+      // called at construction-time to decide if it's behind on throttling bytes
+      writable = new ThrottleStream({ bps: throttleKbps * 1024 })
+      writable.pipe(pt)
     }
 
-    pt.end()
+    if (body) {
+      if ((body as Readable).pipe) {
+        return (body as Readable).pipe(writable)
+      }
 
-    resStream = pt
+      writable.write(body)
+    }
+
+    return writable.end()
   }
 
-  onResponse(incomingRes, resStream)
+  delayMs ? setTimeout(sendBody, delayMs) : sendBody()
+
+  return pt
 }