Angular/Jest中无法模拟派生Observable流，求NgRx Effect测试方案

问题背景 #

在Angular项目中使用Jest测试NgRx Effect时，无法正确模拟基于ReplaySubject的派生Observable流wsService.cancelPendingTrade$，导致测试无法通过。其余依赖（Store、TradingControllerService）均已正确模拟，核心问题集中在WebSocketService的派生流模拟上。

核心代码 #

NgRx Effect代码 #

export const cancel$ = createEffect(
    (
      actions$ = inject(Actions),
      tradingControllerService = inject(TradingControllerService),
      wsService = inject(WebSocketService),
      store = inject(Store),
    ) => {
      return actions$.pipe(
        ofType(pendingOrdersActions.cancel),
        switchWith(() => store.select(WebTraderSelectors.selectTradingAccountId)),
        switchMap(([{ id }, tradingAccountId]) => {
          const referenceId = TradingMathFunctions.generateReferenceId();
          return tradingControllerService.cancelOrder(id, <number>tradingAccountId, referenceId);
        }),
        switchMap(() => wsService.cancelPendingTrade$),
        map(({ tradeId }) => pendingOrdersActions.cancelSuccess({ id: tradeId })),
      );
    },
    { functional: true },
  );

WebSocketService关键代码 #

cancelPendingTrade$ = this.message$.pipe(
    filter(WebSocketFunctions.isCancelPendingTradeMessage),
    map(WebSocketFunctions.getCancelPendingTradeData),
    filter(Boolean),
  );

webSocket$: WebSocketSubject<Cli2Ser | Ser2Cli | MockCli2Ser> | undefined;
message$ = new ReplaySubject<Ser2Cli>(1);

现有测试代码 #

describe('PendingOrdersEffects', () => {
  let tradingControllerServiceMock: MockProxy<TradingControllerService>;
  let storeMock: MockProxy<Store>;
  let wsServiceMock: MockProxy<WebSocketService>;

  beforeEach(() => {
    storeMock = mock<Store>({
      select: jest.fn().mockReturnValueOnce(of(1234)),
    });
    tradingControllerServiceMock = mock<TradingControllerService>({
      cancelOrder: jest.fn(),
    });
    wsServiceMock = mock<WebSocketService>();
  });

describe('cancel$', () => {
    it('should return `cancelSuccess` action', () => {
      const spy = jest.fn();
      const actions$ = of(pendingOrdersActions.cancel({ id: 1 }));
      tradingControllerServiceMock.cancelOrder.mockReturnValueOnce(
        of({ data: { orderId: 1 } } as unknown as HttpEvent<StandardResponseOrderResult>),
      );

// Mock this stream here

      PendingOrdersEffects.cancel$(actions$, tradingControllerServiceMock, wsServiceMock, storeMock)
        .pipe(take(1))
        .subscribe(spy);

      expect(spy).toHaveBeenCalledWith(pendingOrdersActions.cancelSuccess({ id: 1 }));
    });
  });

解决方案 #

针对cancelPendingTrade$的模拟，有两种可靠实现方式：

方式1：直接模拟派生Observable流 #

由于cancelPendingTrade$是WebSocketService的公共属性，可直接为其赋值一个预定义的Observable，跳过底层ReplaySubject的依赖：

// 在测试代码的Mock区域添加：
wsServiceMock.cancelPendingTrade$ = of({ tradeId: 1 });

方式2：触发底层ReplaySubject生成派生流数据 #

如果需要更贴近真实逻辑的测试（验证过滤/映射函数的正确性），可以直接操作message$ ReplaySubject，发送符合条件的消息，让cancelPendingTrade$自然派生数据：

// 1. 初始化mock的message$
wsServiceMock.message$ = new ReplaySubject<Ser2Cli>(1);
// 2. 构造符合校验规则的消息对象，需匹配WebSocketFunctions的过滤条件
const mockCancelMessage = { 
  // 填写Ser2Cli类型中满足isCancelPendingTradeMessage校验的字段，例如：
  type: 'CANCEL_PENDING_TRADE',
  data: { tradeId: 1 }
} as unknown as Ser2Cli;
// 3. 发送消息触发派生流
wsServiceMock.message$.next(mockCancelMessage);

验证测试 #

将上述任一方式添加到测试代码的// Mock this stream here位置后，重新运行测试：

• Effect会正确订阅到cancelPendingTrade$的模拟数据
• 最终会触发cancelSuccess action，测试断言通过

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