如何在测试中Mock @golevelup/nestjs-rabbitmq的AmqpConnection发布方法

我已基于@golevelup/nestjs-rabbitmq模块实现了可用的发布/订阅功能，现需编写服务测试用例，测试MessagingService中调用AmqpConnection.publish方法的publish方法，请问如何在测试中Mock该publish方法？

现有代码实现 #

MessagingModule #

import { RabbitModule } from '../rabbit/rabbit.module';
import { Module } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { MessagingService } from './messaging.service';

@Module({
  imports: [
    RabbitModule,
    MessagingModule,
  ],
  providers: [ConfigService, MessagingService],
  controllers: [],
  exports: [MessagingService],
})
export class MessagingModule {}

MessagingService #

import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { QueueMessage } from 'src/models/queue-message.model';

@Injectable()
export class MessagingService {
  private readonly exchange: string = this.configService.get<string>('RABBITMQ_EXCHANGE') || 'drs-encrypt.exchange';
  private readonly routingKey: string = this.configService.get<string>('RABBITMQ_ROUTING_KEY') || 'drs-encrypt';
  private readonly logger = new Logger(MessagingService.name);

  constructor(
    private readonly configService: ConfigService,
    private readonly amqpConnection: AmqpConnection
  ) {}

  // 发布队列消息到交换机
  public async publish<T>(queueMessage: QueueMessage): Promise<void> {
    this.logger.log(`publishing queueMessage to exchange: ${this.exchange} routing key: ${this.routingKey}`);
    await this.amqpConnection.publish(this.exchange, this.routingKey, queueMessage);
  }
}

当前测试代码（待修正） #

import { Test, TestingModule } from '@nestjs/testing';
import { MessagingService } from '../messaging/messaging.service';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq'
import { ConfigService } from '@nestjs/config';

describe('MessagingService', () => {
  let service: MessagingService;

  type MockType<T> = {
    [P in keyof T]?: jest.Mock<{}>;
  };

  const mockFactoryAmqpConnection: () => MockType<AmqpConnection> = jest.fn(() => ({
    publish: jest.fn(() => AmqpConnection),
  }));

  beforeEach(async () => {
    const module: TestingModule = await Test.createTestingModule({
      providers: [
        ConfigService,
        MessagingService,
        {
          provide: AmqpConnection,
          useFactory: mockFactoryAmqpConnection,
        },
      ],
      controllers: [],
    }).compile();

    service = module.get<MessagingService>(MessagingService);

  });

  it('should be defined', () => {
    expect(service).toBeDefined();
  });  

});

修正后的测试方案 #

要正确Mock AmqpConnection.publish 并验证其调用情况，需调整测试代码如下：

import { Test, TestingModule } from '@nestjs/testing';
import { MessagingService } from '../messaging/messaging.service';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { ConfigService } from '@nestjs/config';
import { QueueMessage } from 'src/models/queue-message.model';

describe('MessagingService', () => {
  let service: MessagingService;
  // 保存AmqpConnection的mock实例，用于断言调用
  let amqpConnectionMock: jest.Mocked<AmqpConnection>;

  beforeEach(async () => {
    // 定义mock对象，确保publish返回Promise<void>（和原方法签名一致）
    amqpConnectionMock = {
      publish: jest.fn().mockResolvedValue(void 0),
    } as jest.Mocked<AmqpConnection>;

    const module: TestingModule = await Test.createTestingModule({
      providers: [
        ConfigService,
        MessagingService,
        // 直接使用useValue提供mock实例，更简洁
        {
          provide: AmqpConnection,
          useValue: amqpConnectionMock,
        },
      ],
    }).compile();

    service = module.get<MessagingService>(MessagingService);
  });

  it('should be defined', () => {
    expect(service).toBeDefined();
  });

  it('should call amqpConnection.publish with correct parameters', async () => {
    // 准备测试数据
    const testMessage: QueueMessage = { /* 填入你的QueueMessage测试结构 */ };
    // 获取服务中配置的exchange和routingKey（也可以直接用预期值）
    const exchange = service['exchange'];
    const routingKey = service['routingKey'];

    // 调用待测试的publish方法
    await service.publish(testMessage);

    // 断言AmqpConnection的publish方法被调用了一次，且参数正确
    expect(amqpConnectionMock.publish).toHaveBeenCalledTimes(1);
    expect(amqpConnectionMock.publish).toHaveBeenCalledWith(
      exchange,
      routingKey,
      testMessage
    );
  });
});

关键说明 #

• Mock实例管理：直接创建jest.Mocked<AmqpConnection>类型的mock对象，替代原有的工厂函数，更直观且方便后续断言。
• 方法签名匹配：确保mock的publish方法返回Promise<void>（用mockResolvedValue(void 0)），和原方法的返回值类型一致，避免测试时出现异步错误。
• 参数断言：通过获取服务内部的exchange和routingKey（或直接使用预期配置值），验证publish方法调用时传入的参数是否符合业务逻辑。
• 访问私有属性：测试中通过service['exchange']访问私有属性是测试场景下的合理操作，用于验证配置读取是否正确。

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