Angular中含RxJS race操作符的NgRx Effects单元测试咨询

针对你描述的使用switchMap嵌套race（监听3秒timer或指定action）的Effects，以下是可行的单元测试写法，分fakeAsync同步测试和Marble事件流测试两种方案：

先明确Effects示例代码（对应你的场景） #

假设你的Effects逻辑如下：

import { Injectable } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { of, race, timer } from 'rxjs';
import { map, switchMap } from 'rxjs/operators';

// 假设的Action类型
export enum MyActionTypes {
  Init = '[My] Init',
  TriggerAction = '[My] Trigger Action',
  FinalAction = '[My] Final Action',
}

export const init = () => ({ type: MyActionTypes.Init });
export const triggerAction = () => ({ type: MyActionTypes.TriggerAction });
export const finalAction = (trigger: 'timer' | 'action') => ({
  type: MyActionTypes.FinalAction,
  payload: { trigger },
});

@Injectable()
export class MyEffects {
  myEffect$ = createEffect(() =>
    this.actions$.pipe(
      ofType(MyActionTypes.Init),
      switchMap(() =>
        race(
          timer(3000).pipe(mapTo('timer')),
          this.actions$.pipe(ofType(MyActionTypes.TriggerAction), mapTo('action'))
        ).pipe(
          switchMap((trigger) => of(finalAction(trigger)))
        )
      )
    )
  );

  constructor(private actions$: Actions) {}
}

方案1：fakeAsync同步测试 #

这种方式适合直观测试时间推进和action触发场景：

import { fakeAsync, flush, tick } from '@angular/core/testing';
import { TestBed } from '@angular/core/testing';
import { provideMockActions } from '@ngrx/effects/testing';
import { Observable, of, concat } from 'rxjs';
import { Action } from '@ngrx/store';
import { MyEffects, init, triggerAction, finalAction } from './my.effects';

describe('MyEffects', () => {
  let actions$: Observable<Action>;
  let effects: MyEffects;

  beforeEach(() => {
    TestBed.configureTestingModule({
      providers: [
        MyEffects,
        provideMockActions(() => actions$), // 模拟Action流
      ],
    });

    effects = TestBed.inject(MyEffects);
  });

  // 测试：3秒timer先触发，派发FinalAction
  it('dispatches FinalAction when timer completes first', fakeAsync(() => {
    // 触发初始Init Action
    actions$ = of(init());

    let emittedAction: Action | undefined;
    // 必须订阅Effect流，否则内部逻辑不会执行
    effects.myEffect$.subscribe((action) => {
      emittedAction = action;
    });

    // 快进3秒，触发timer
    tick(3000);

    // 断言结果
    expect(emittedAction).toEqual(finalAction('timer'));
  }));

  // 测试：TriggerAction先派发，立即触发FinalAction
  it('dispatches FinalAction when TriggerAction is emitted first', fakeAsync(() => {
    // 先发出Init，紧接着发出TriggerAction
    actions$ = concat(of(init()), of(triggerAction()));

    let emittedAction: Action | undefined;
    effects.myEffect$.subscribe((action) => {
      emittedAction = action;
    });

    // 刷新所有异步任务（处理action流的订阅）
    flush();

    expect(emittedAction).toEqual(finalAction('action'));
  }));
});

方案2：Marble事件流测试（NgRx推荐） #

使用TestScheduler可以更精准地模拟时间和事件流，适合复杂的RxJS逻辑测试：

import { TestBed } from '@angular/core/testing';
import { provideMockActions } from '@ngrx/effects/testing';
import { Observable } from 'rxjs';
import { Action } from '@ngrx/store';
import { TestScheduler } from 'rxjs/testing';
import { MyEffects, init, triggerAction, finalAction } from './my.effects';

describe('MyEffects (Marble Tests)', () => {
  let actions$: Observable<Action>;
  let effects: MyEffects;
  let scheduler: TestScheduler;

  beforeEach(() => {
    TestBed.configureTestingModule({
      providers: [
        MyEffects,
        provideMockActions(() => actions$),
      ],
    });

    effects = TestBed.inject(MyEffects);
    // 初始化测试调度器，用于断言事件流
    scheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it('emits FinalAction after 3s when timer wins the race', () => {
    scheduler.run(({ hot, expectObservable }) => {
      const initAction = init();
      const expectedFinalAction = finalAction('timer');

      // 模拟：0时刻发出Init Action
      actions$ = hot('-a', { a: initAction });

      // 断言：3秒后发出FinalAction
      expectObservable(effects.myEffect$).toBe('3s b', { b: expectedFinalAction });
    });
  });

  it('emits FinalAction immediately when TriggerAction wins the race', () => {
    scheduler.run(({ hot, expectObservable }) => {
      const initAction = init();
      const triggerAct = triggerAction();
      const expectedFinalAction = finalAction('action');

      // 模拟：0时刻发出Init，10ms后发出TriggerAction
      actions$ = hot('-a-b', { a: initAction, b: triggerAct });

      // 断言：TriggerAction发出后立即发出FinalAction
      expectObservable(effects.myEffect$).toBe('--c', { c: expectedFinalAction });
    });
  });
});

关键注意事项 #

• 必须订阅Effect流：Effects是冷Observable，只有订阅后才会执行内部逻辑，这是很多测试失败的核心原因
• fakeAsync的正确使用：tick(3000)用于快进时间触发timer，flush()用于处理所有待执行的异步任务（比如action流的订阅）
• Marble测试优势：可以精准控制事件的时间顺序，避免fakeAsync中可能出现的异步时序问题，NgRx官方更推荐这种方式

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