NestJS CQRS模式下Saga单元测试事件未触发问题求助

问题场景 #

基于NestJS CQRS模式开发的应用，用Jest编写单元测试覆盖Saga逻辑时遇到异常：在inventory.saga.spec.ts中通过EventBus发布InventoryEvent后，inventory.saga.ts中的inventory()方法未被触发，但实际应用中发布相同事件时，Saga能正常处理并触发InventoryCacheCommand。

应用正常流程：

• Inventory Handler发布InventoryEvent；
• Saga监听InventoryEvent，触发InventoryCacheCommand。

核心原因 #

1. 测试模块未初始化CQRS订阅机制：直接在测试模块中声明InventorySaga、EventBus和CommandBus，但NestJS的Saga需要通过CQRS模块内部逻辑与EventBus建立订阅关系，手动实例化的EventBus不会自动订阅Saga的事件流。
2. 错误的断言目标：直接spyOn Saga的inventory方法无效，因为@Saga装饰器会对该方法进行包装，实际运行时调用的是CQRS框架封装后的版本，而非原始方法。

解决方案 #

步骤1：导入CQRS模块到测试模块 #

通过CqrsModule.forRoot()确保测试环境中CQRS的订阅机制正常初始化，让Saga自动与EventBus关联。

步骤2：调整测试断言逻辑 #

不再断言Saga方法是否被调用，改为验证CommandBus是否收到预期的InventoryCacheCommand，这更贴合Saga的业务逻辑目标。

步骤3：处理异步事件流 #

由于Saga基于RxJS Observable处理事件，需确保事件流完成后再执行断言，可使用done回调或结合RxJS操作符等待异步操作完成。

修正后的测试代码 #

import { Test, TestingModule } from '@nestjs/testing';
import { InventorySaga } from './inventory.saga';
import { InventoryEvent } from '../events/impl/inventory.event';
import { CommandBus, EventBus, CqrsModule } from '@nestjs/cqrs';
import { InventoryCacheCommand } from '../commands/impl/inventory-cache.command';

// 模拟CommandBus的execute方法
jest.mock('@nestjs/cqrs', () => ({
  ...jest.requireActual('@nestjs/cqrs'),
  CommandBus: jest.fn().mockImplementation(() => ({
    execute: jest.fn().mockResolvedValue(null),
  })),
}));

describe('InventorySaga', () => {
  let saga: InventorySaga;
  let eventBus: EventBus;
  let commandBus: CommandBus;

  beforeEach(async () => {
    const module: TestingModule = await Test.createTestingModule({
      imports: [CqrsModule.forRoot()], // 导入CQRS模块初始化订阅机制
      providers: [InventorySaga],
    }).compile();

    saga = module.get<InventorySaga>(InventorySaga);
    eventBus = module.get<EventBus>(EventBus);
    commandBus = module.get<CommandBus>(CommandBus);
  });

  describe('inventory saga', () => {
    it('should trigger InventoryCacheCommand when InventoryEvent is published', (done) => {
      const inventoryData = [
        { sku: 'TH4344-43-L', qty: 3 },
        { sku: 'TH4344-43-S', qty: 55 },
        { sku: 'TH4344-43-XL', qty: 55 },
      ];

      // 监听CommandBus的execute方法，验证是否收到预期命令
      const executeSpy = jest.spyOn(commandBus, 'execute');

      // 发布事件
      eventBus.publish(new InventoryEvent(inventoryData));

      // 等待RxJS异步流处理完成
      setTimeout(() => {
        expect(executeSpy).toHaveBeenCalledWith(
          expect.any(InventoryCacheCommand)
        );
        // 验证命令中的数据正确性
        const command = executeSpy.mock.calls[0][0];
        expect(command.inventoryData).toEqual(inventoryData);
        done();
      }, 100);
    });
  });
});

额外说明 #

• 若不想导入完整CqrsModule，也可手动订阅Saga的事件流：在测试中调用saga.inventory(eventBus.getSubject())，但需手动管理订阅，不如导入模块简洁。
• 避免直接测试Saga内部方法，应聚焦业务行为（事件触发后是否生成预期命令），这样测试更稳定且贴合业务需求。

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