如何使用JEST测试基于amqplib的RabbitMQ队列消费NodeJS代码

这段代码完全可以进行单元测试，不需要依赖真实启动的RabbitMQ服务，核心思路是Mock所有外部依赖（amqplib库、setTimeout定时器、传入的socket.io实例io），即可覆盖所有逻辑分支的验证。

1. 核心测试覆盖场景 #

• 服务启动后延迟10秒正确调用RabbitMQ连接方法
• RabbitMQ连接失败时正确抛出错误
• 连接成功后正确创建Channel
• 正确断言队列，参数符合配置预期
• 接收到队列消息时，正确打印日志且调用io.emit推送消息
• Channel创建失败时正确抛出错误

2. 测试代码实现 #

const amqp = require('amqplib/callback_api');
// 替换为你实际存放rabbitConsumer代码的文件路径
const rabbitConsumer = require('./rabbit-consumer');

// Mock整个amqplib/callback_api库
jest.mock('amqplib/callback_api');
// Mock定时器，不需要真实等待10秒
jest.useFakeTimers();

describe('rabbitConsumer 逻辑验证', () => {
  // Mock socket.io的io实例
  const mockIo = {
    emit: jest.fn()
  };

  beforeEach(() => {
    // Mock console.log避免测试过程产生冗余输出
    jest.spyOn(console, 'log').mockImplementation(() => {});
    // 每次用例执行前清空所有Mock调用记录
    amqp.connect.mockClear();
    mockIo.emit.mockClear();
  });

  it('启动后延迟10秒触发RabbitMQ连接', () => {
    rabbitConsumer(mockIo);
    // 定时器未触发时连接方法不会被调用
    expect(amqp.connect).not.toHaveBeenCalled();
    // 快进10秒触发定时器回调
    jest.advanceTimersByTime(10000);
    expect(amqp.connect).toHaveBeenCalledWith('amqp://rabbitmq', expect.any(Function));
  });

  it('RabbitMQ连接失败时正确抛出错误', () => {
    const mockConnectError = new Error('RabbitMQ连接超时');
    rabbitConsumer(mockIo);
    jest.advanceTimersByTime(10000);
    // 拿到connect的回调函数传入错误参数
    const connectCallback = amqp.connect.mock.calls[0][1];
    expect(() => connectCallback(mockConnectError, null)).toThrow(mockConnectError);
  });

  it('Channel创建失败时正确抛出错误', () => {
    const mockChannelError = new Error('Channel创建失败');
    const mockConnection = {
      createChannel: jest.fn((cb) => cb(mockChannelError, null))
    };
    rabbitConsumer(mockIo);
    jest.advanceTimersByTime(10000);
    const connectCallback = amqp.connect.mock.calls[0][1];
    expect(() => connectCallback(null, mockConnection)).toThrow(mockChannelError);
  });

  it('连接成功后正确初始化队列和消费逻辑', () => {
    const mockChannel = {
      assertQueue: jest.fn(),
      consume: jest.fn()
    };
    const mockConnection = {
      createChannel: jest.fn((cb) => cb(null, mockChannel))
    };
    rabbitConsumer(mockIo);
    jest.advanceTimersByTime(10000);
    const connectCallback = amqp.connect.mock.calls[0][1];
    connectCallback(null, mockConnection);

    // 验证Channel创建、队列断言、消费注册逻辑的参数符合预期
    expect(mockConnection.createChannel).toHaveBeenCalled();
    expect(mockChannel.assertQueue).toHaveBeenCalledWith('message', { durable: false });
    expect(mockChannel.consume).toHaveBeenCalledWith('message', expect.any(Function), { noAck: true });
    expect(console.log).toHaveBeenCalledWith(' [*] Waiting for message', 'message');
  });

  it('收到队列消息时正确打印日志并推送消息到socket.io', () => {
    const testMsg = '测试消费消息内容';
    const mockMsg = {
      content: {
        toString: jest.fn(() => testMsg)
      }
    };
    const mockChannel = {
      assertQueue: jest.fn(),
      consume: jest.fn((queue, consumeCb) => consumeCb(mockMsg))
    };
    const mockConnection = {
      createChannel: jest.fn((cb) => cb(null, mockChannel))
    };

    rabbitConsumer(mockIo);
    jest.advanceTimersByTime(10000);
    const connectCallback = amqp.connect.mock.calls[0][1];
    connectCallback(null, mockConnection);

    // 验证消息处理逻辑正确性
    expect(mockMsg.content.toString).toHaveBeenCalledWith('utf-8');
    expect(console.log).toHaveBeenCalledWith(' [x] Received data:', testMsg);
    expect(mockIo.emit).toHaveBeenCalledWith('sendMessage', testMsg);
  });
});

3. 可选集成测试方案 #

如果需要做端到端的集成验证，可以使用testcontainers库临时启动RabbitMQ测试容器，走真实的连接、发消息、消费全流程验证功能正确性。

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