端到端（E2E）集成测试：如何用NodeJS验证Azure Event Grid消息？

问题背景 #

我有一个.NET HttpTrigger Azure Function，接收请求体后会将其转发到Azure Event Grid主题。现在需要用Node.js/TypeScript编写端到端（E2E）或集成测试，核心流程是：模拟HTTP POST请求调用该Function，然后验证消息是否成功发送到Event Grid主题。目前遇到的问题：

• 找到的.NET相关测试方案无法复用，因为测试项目基于Playwright+NodeJS
• 了解到Azure Relay Connection的实现方式，但不确定是否适合当前场景，也不知道如何整合到测试流程中

当前尝试的代码片段：

import { test as base, APIResponse } from '@playwright/test';

let _respStatus: number;
let _response: APIResponse;

// 调用Azure Function
base.step('Send POST to "admin/functions/FunctionName"', async ({ request }) => {
  _response = await request.post('admin/functions/FunctionName', { data: { /* 测试请求体 */ } });
  _respStatus = _response.status();
});

// 尝试通过Azure Relay监听Event Grid消息
base.step("Listen for Event Grid message", async () => {
  const https = require("hyco-https");
  const ns = "{RelayNamespace}";
  const path = "{HybridConnectionName}";
  const keyrule = "{SASKeyName}";
  const key = "{SASKeyValue}";

  var uri = https.createRelayListenUri(ns, path);
  var server = https.createRelayedServer(
    {
      server: uri,
      token: () => https.createRelayToken(uri, keyrule, key),
    },
    (req, res) => {
      res.setHeader("Content-Type", "text/html");
      res.end("<html><head><title>Hey!</title></head><body>Relayed Node.js Server!</body></html>");
    }).listen(() => {});
});

可行解决方案 #

方案1：利用Azure Storage队列中转验证（推荐） #

通过给Event Grid主题配置存储队列订阅，将主题消息转发到Azure Storage队列，测试时直接查询队列验证消息，实现简单且稳定。

步骤：

1. 配置Event Grid订阅：
   • 给目标Event Grid主题创建新订阅，端点类型选择Azure 存储队列，指定一个Azure Storage队列作为接收端
2. 测试流程：
   • 调用Azure Function的HttpTrigger接口
   • 等待1-2秒（处理异步转发延迟）
   • 使用Azure Storage SDK连接到队列，拉取消息并验证内容

代码示例：

import { test as base, APIResponse } from '@playwright/test';
import { QueueClient } from "@azure/storage-queue";

// 初始化存储队列客户端
const queueClient = new QueueClient(
  "<你的Storage连接字符串>",
  "<队列名称>"
);

let _respStatus: number;
let _response: APIResponse;
const testRequestBody = { test: "data" }; // 测试请求体

base.step('调用Azure Function并发送测试数据', async ({ request }) => {
  _response = await request.post('admin/functions/FunctionName', { data: testRequestBody });
  _respStatus = _response.status();
  // 先验证Function返回状态
  expect(_respStatus).toBe(200);
});

base.step('验证消息到达Event Grid（通过存储队列）', async () => {
  // 等待消息转发完成
  await new Promise(resolve => setTimeout(resolve, 2000));

  // 拉取队列消息
  const response = await queueClient.receiveMessages({ maxMessages: 1 });
  const receivedMessage = response.receivedMessages[0];

  // 验证消息内容
  expect(receivedMessage).not.toBeUndefined();
  const messageBody = JSON.parse(receivedMessage.messageText);
  // 根据Event Grid消息结构，实际业务数据在data字段中
  expect(messageBody.data).toEqual(testRequestBody);

  // 验证完成后删除消息，避免影响下一次测试
  if (receivedMessage) {
    await queueClient.deleteMessage(receivedMessage.messageId, receivedMessage.popReceipt);
  }
});

方案2：用Azure Relay作为消息接收端点 #

如果必须使用Azure Relay，可以将Event Grid主题的订阅指向Relay混合连接，测试时启动Relay监听服务捕获消息。

步骤：

1. 配置Event Grid订阅：
   • 创建新订阅，端点类型选择Web Hook，填写Relay混合连接的地址：https://<你的Relay命名空间>.servicebus.windows.net/<混合连接名称>
2. 修改测试代码：
   • 将Relay监听逻辑改为捕获请求体，用Promise等待消息到达，完成验证

修正后的代码示例：

import { test as base, APIResponse } from '@playwright/test';

let _respStatus: number;
let _response: APIResponse;
const testRequestBody = { test: "data" };

base.step('调用Azure Function', async ({ request }) => {
  _response = await request.post('admin/functions/FunctionName', { data: testRequestBody });
  _respStatus = _response.status();
  expect(_respStatus).toBe(200);
});

base.step('通过Azure Relay监听Event Grid消息并验证', async () => {
  const https = require("hyco-https");
  const ns = "{RelayNamespace}";
  const path = "{HybridConnectionName}";
  const keyrule = "{SASKeyName}";
  const key = "{SASKeyValue}";

  // 用Promise包装监听逻辑，等待消息到达
  const messageReceived = new Promise((resolve, reject) => {
    const uri = https.createRelayListenUri(ns, path);
    const server = https.createRelayedServer(
      {
        server: uri,
        token: () => https.createRelayToken(uri, keyrule, key),
      },
      (req, res) => {
        // 读取请求体（Event Grid消息）
        let body = '';
        req.on('data', chunk => body += chunk.toString());
        req.on('end', () => {
          try {
            const eventGridMessages = JSON.parse(body);
            // 找到对应测试消息并验证
            const targetMessage = eventGridMessages.find((msg: any) => JSON.stringify(msg.data) === JSON.stringify(testRequestBody));
            expect(targetMessage).not.toBeUndefined();
            resolve(true);
          } catch (err) {
            reject(err);
          } finally {
            // 发送响应给Event Grid，避免重复投递
            res.writeHead(200);
            res.end();
            // 关闭服务器
            server.close();
          }
        });
      }
    ).listen(() => console.log('Relay监听已启动'));

    // 超时处理，避免无限等待
    setTimeout(() => {
      server.close();
      reject(new Error('超时未收到Event Grid消息'));
    }, 10000);
  });

  await messageReceived;
});

方案3：本地测试用Event Grid模拟器 #

如果是本地开发阶段，可以使用Azure Event Grid模拟器，直接通过模拟器API查询已发送的消息，无需依赖云服务：

1. 安装并启动Event Grid模拟器
2. 将Azure Function的Event Grid输出绑定指向模拟器地址
3. 测试时调用模拟器的/api/events接口查询消息

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