如何测试带pipe和mergeMap的RxJS Observable代码 解决Jest无法测subscribe内部逻辑问题

一、原有代码问题说明 #

• 用Array.map做遍历仅触发副作用，未使用返回值，多个分散订阅无法统一管控
• 外部可变变量invite_counts存在并发竞态风险，多个并行请求返回时计数可能不准
• 没有控制并发请求量，people数量大时会一次性发起大量apollo请求
• 订阅逻辑散落在循环中，没有统一的完成/错误收口，可测试性差

二、优化后代码 #

import { from, takeUntil, mergeMap, scan, filter } from 'rxjs';

// 只有people存在时才发起流
if (this.people?.length) {
  from(this.people).pipe(
    takeUntil(this.unsubscribe$),
    // 第二个参数控制并发数，可根据实际场景调整，比如最多同时发5个请求
    mergeMap(person => this.checkFriends(person).pipe(
      mergeMap(() => this.updateFriendCircle())
    ), 5),
    // 用scan内部维护计数，避免外部可变变量
    scan((count, result) => result.invites ? count : count + 1, 0),
    // 计数到100时直接触发，不用在subscribe里判断
    filter(count => count === 100)
  ).subscribe({
    next: () => popular(),
    error: () => logError()
  })
}

三、Jest测试实现方法 #

测试不生效的核心原因 #

原有代码的subscribe回调是异步执行逻辑，Jest默认不会等待异步流执行完毕，且你只mock了checkFriends，未mock依赖的updateFriendCircle方法，导致流中断无法走到subscribe回调。

测试用例代码 #

import { of, Subject, throwError } from 'rxjs';
import { fakeAsync, tick } from '@angular/core/testing'; // 非Angular项目可替换为RxJS的TestScheduler

describe('朋友圈更新逻辑', () => {
  let component: any;
  let unsubscribe$: Subject<void>;
  // 先mock所有依赖方法
  beforeEach(() => {
    unsubscribe$ = new Subject<void>();
    // 初始化模拟组件
    component = {
      unsubscribe$,
      people: [],
      checkFriends: jest.fn(),
      updateFriendCircle: jest.fn(),
    };
    // 全局mock用到的副作用函数
    (window as any).popular = jest.fn();
    (window as any).logError = jest.fn();
  });

  // 测试场景1：正常流程计数到100触发popular
  it('计数到100时触发popular方法', fakeAsync(() => {
    // 构造100个测试用户
    component.people = new Array(100).fill({ id: 1 });
    // mock checkFriends返回正常流
    jest.spyOn(component, 'checkFriends').mockReturnValue(of({}));
    // mock updateFriendCircle返回没有invites的结果，让计数+1
    jest.spyOn(component, 'updateFriendCircle').mockReturnValue(of({}));

    // 执行你的业务方法
    component.yourBusinessMethod();
    // 同步执行所有微任务
    tick();

    expect(popular).toHaveBeenCalled();
  }));

  // 测试场景2：请求出错时触发logError
  it('请求出错时调用logError', fakeAsync(() => {
    component.people = [{ id: 1 }];
    // mock checkFriends抛出错误
    jest.spyOn(component, 'checkFriends').mockReturnValue(throwError(() => new Error('请求失败')));

    component.yourBusinessMethod();
    tick();

    expect(logError).toHaveBeenCalled();
  }));

  // 测试场景3：unsubscribe$触发时终止所有请求
  it('unsubscribe触发后不再执行后续逻辑', fakeAsync(() => {
    component.people = new Array(10).fill({ id: 1 });
    jest.spyOn(component, 'checkFriends').mockReturnValue(of({}));
    jest.spyOn(component, 'updateFriendCircle').mockReturnValue(of({}));

    component.yourBusinessMethod();
    // 先发射终止信号
    unsubscribe$.next();
    unsubscribe$.complete();
    tick();

    expect(popular).not.toHaveBeenCalled();
    expect(component.updateFriendCircle).not.toHaveBeenCalled();
  }));
});

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