如何在测试环境中Stub/Mock ChatGPT的流式响应？

测试环境中需要拦截/api/chat-stream请求并替换为Mock的流式响应，避免直接调用ChatGPT API导致E2E测试不稳定。现有前端代码通过ReadableStream.getReader()读取流式数据，解析后渲染UI，但不知道如何正确构造Mock的流式响应适配现有逻辑。

现有代码片段 #

请求拦截代码（待完善） #

// 需在此实现Mock/Stub逻辑
await page.route(/.*\/api\/chat-stream/, async (route) => {
  await route.fulfill({
    status: 200,
    headers: {
      "some-headers"
    },
    body: someBody,
  });
});

后端/api/chat-stream实现 #

// localhost:3000/api/chat-stream

export async function POST(request: Request) {
  const { messages } = await request.json();

  const completion = await fetch("https://api.openai.com/v1/chat/completions", {
    method: "POST",
    body: JSON.stringify({
      model: "gpt-3.5-turbo",
      messages: messages,
      stream: true,
    }),
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${process.env.NEXT_PUBLIC_OPENAI_API_KEY}`,
    },
  });

  return new Response(completion.body, {
    status: 200,
    headers: {
      "Content-Type": "application/json; charset=utf-8",
    },
  });
}

前端流式响应处理代码 #

export const handleStreamMessage = async (messages: any) => {
  const response = await fetch("/api/chat-stream", {
    method: "POST",
    body: JSON.stringify({
      messages: messages,
    }),
    headers: {
      "Content-Type": "application/json",
    },
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  const onParse: EventSourceParseCallback = (event) => {
    if (event.type === "event") {
      try {
        const data: { choices: { delta: { content: string } }[] } = JSON.parse(
          event.data
        );

        // 过滤带内容的ChatGPT增量响应
        data.choices
          .filter(({ delta }) => !!delta.content)
          .forEach(({ delta }) => {
            // 更新React状态渲染内容
            setCurrentMessage((prev) => {
              return `${prev || ""}${delta.content}`;
            });
          });
      } catch (error) {
        console.log("error", error);
      }
    }
  };

  const parser = createParser(onParse);

  if (reader) {
    while (true) {
      const readOperation = await reader.read();
      console.log("readOperation", readOperation);

      const dataString = decoder.decode(readOperation.value);

      if (readOperation.done || dataString.includes("[DONE]")) {
        break;
      }

      parser.feed(dataString);
    }
  }
};

要Mock符合ChatGPT流式格式的响应，需构造**Server-Sent Events(SSE)**格式的分块数据，完全匹配前端解析逻辑，具体实现如下：

1. 构造Mock流式响应内容 #

ChatGPT流式响应遵循固定格式：

• 每个数据块以data: 开头，后跟JSON字符串，结尾是\n\n
• 最后一个块为data: [DONE]\n\n，用于通知前端流结束

示例Mock数据（模拟打字式输出）：

const mockStreamChunks = [
  'data: {"choices":[{"delta":{"content":"你"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"好"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"，"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"我"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"是"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"Mock"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"响"}}]}\n\n',
  'data: {"choices":[{"delta":{"content":"应"}}]}\n\n',
  'data: [DONE]\n\n'
];

2. 在Playwright拦截中返回流式响应 #

通过创建ReadableStream模拟分块输出，同时设置正确的响应头：

await page.route(/.*\/api\/chat-stream/, async (route) => {
  // 构造Mock流式数据
  const mockStreamChunks = [
    'data: {"choices":[{"delta":{"content":"你"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"好"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"，"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"我"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"是"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"Mock"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"响"}}]}\n\n',
    'data: {"choices":[{"delta":{"content":"应"}}]}\n\n',
    'data: [DONE]\n\n'
  ];

  // 创建ReadableStream模拟流式输出
  const stream = new ReadableStream({
    async start(controller) {
      for (const chunk of mockStreamChunks) {
        // 每个块之间延迟100ms，模拟真实打字速度
        await new Promise(resolve => setTimeout(resolve, 100));
        controller.enqueue(new TextEncoder().encode(chunk));
      }
      controller.close();
    }
  });

  await route.fulfill({
    status: 200,
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive"
    },
    body: stream
  });
});

3. 关键注意事项 #

• 响应头必须正确：指定Content-Type: text/event-stream，否则前端解析器无法识别SSE格式
• 模拟真实节奏：添加setTimeout避免一次性返回所有数据，还原真实流式输出的打字效果
• 格式严格匹配：每个数据块必须遵循data: {JSON}\n\n格式，最后以data: [DONE]\n\n结束，确保前端逻辑能正确终止读取

内容的提问来源于stack exchange，提问作者kevin