NestJS中使用Jest模拟Nats pub、sub调用的最佳方案是什么

核心思路 #

不需要启动真实NATS服务，通过自定义Provider替换NestJS中注入的NATS客户端实例，直接用Jest创建的Mock对象模拟pub/sub相关方法，仅验证业务代码的调用行为是否符合预期即可。

具体实现 #

根据测试场景分为发布端测试和订阅端测试两类：

场景1：测试调用NATS pub的业务服务 #

这类场景下我们需要验证业务代码是否正确调用了NATS的发布方法、传入的主题和 payload 是否符合预期。
示例业务代码：

// app.service.ts
import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';

@Injectable()
export class AppService {
  constructor(@Inject('NATS_CLIENT') private natsClient: ClientProxy) {}

  publishUserCreateEvent(userId: string) {
    // 发布user.created事件
    return this.natsClient.emit('user.created', { userId }).subscribe();
  }
}

对应的测试代码：

// app.service.spec.ts
import { Test, TestingModule } from '@nestjs/testing';
import { ClientProxy } from '@nestjs/microservices';
import { AppService } from './app.service';

describe('AppService', () => {
  let service: AppService;
  let mockNatsClient: Partial<jest.Mocked<ClientProxy>>;

  beforeEach(async () => {
    // 模拟NATS客户端的emit方法，返回带subscribe的Observable模拟对象
    mockNatsClient = {
      emit: jest.fn().mockReturnValue({ subscribe: jest.fn() }),
    };

    const module: TestingModule = await Test.createTestingModule({
      providers: [
        AppService,
        // 替换真实的NATS客户端Provider为Mock实例
        { provide: 'NATS_CLIENT', useValue: mockNatsClient },
      ],
    }).compile();

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

  it('调用publishUserCreateEvent时应该正确发布user.created事件', () => {
    const testUserId = '123456';
    service.publishUserCreateEvent(testUserId);
    // 验证emit方法的调用参数符合预期
    expect(mockNatsClient.emit).toHaveBeenCalledWith('user.created', { userId: testUserId });
  });
});

场景2：测试监听NATS sub的业务服务 #

这类场景下不需要依赖NATS服务接收消息，直接调用监听方法验证业务逻辑是否正确执行即可。
示例业务代码：

// user.listener.ts
import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';
import { UserService } from './user.service';

@Controller()
export class UserListener {
  constructor(private userService: UserService) {}

  // 订阅user.created事件
  @EventPattern('user.created')
  handleUserCreated(@Payload() data: { userId: string }) {
    return this.userService.processNewUser(data.userId);
  }
}

对应的测试代码：

// user.listener.spec.ts
import { Test, TestingModule } from '@nestjs/testing';
import { UserListener } from './user.listener';
import { UserService } from './user.service';

describe('UserListener', () => {
  let listener: UserListener;
  let mockUserService: Partial<jest.Mocked<UserService>>;

  beforeEach(async () => {
    mockUserService = {
      processNewUser: jest.fn().mockResolvedValue(true),
    };

    const module: TestingModule = await Test.createTestingModule({
      controllers: [UserListener],
      providers: [
        { provide: UserService, useValue: mockUserService },
      ],
    }).compile();

    listener = module.get<UserListener>(UserListener);
  });

  it('收到user.created事件时应该调用processNewUser处理数据', () => {
    const testData = { userId: '123456' };
    // 直接调用监听方法，模拟NATS推送消息
    listener.handleUserCreated(testData);
    expect(mockUserService.processNewUser).toHaveBeenCalledWith(testData.userId);
  });
});

关键注意事项 #

• 如果使用的是send方法（请求响应模式），Mock时需要返回符合要求的Observable，可通过rxjs的of方法封装返回值：jest.fn().mockReturnValue(of(mockResponse))
• 单元测试阶段不要连接真实NATS服务，会降低测试速度，同时引入外部依赖的不稳定因素
• 仅需要验证业务逻辑的调用行为即可，不需要测试NATS本身的消息传输能力，该部分属于集成测试的覆盖范围

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