如何用Jest编写测试用例处理RxJS delay的链式API调用

针对你这种包含delay操作符的RxJS链式API调用，不用真等5秒，直接用Jest的fake timers就能高效完成测试，核心思路是模拟时间流逝，而非等待真实时间。

先看示例业务代码 #

假设你的链式调用函数长这样：

import { of } from 'rxjs';
import { delay, switchMap } from 'rxjs/operators';

// 链式API调用函数
export function chainedApiWorkflow() {
  return firstApi().pipe(
    // 第一个API返回后，延迟5秒再执行下一步
    switchMap(firstRes => of(firstRes).pipe(delay(5000))),
    // 延迟后调用第二个API
    switchMap(delayedFirstRes => secondApi(delayedFirstRes)),
    // 处理第二个API返回的后续逻辑
    switchMap(secondRes => {
      return of(secondRes.code === 200 ? 'success' : 'error');
    })
  );
}

// 模拟实际API请求（项目中可能是axios/fetch封装）
export function firstApi() {
  return of({ id: 123, data: 'initial' });
}

export function secondApi(payload: { id: number; data: string }) {
  return of({ code: 200, result: `processed-${payload.data}` });
}

测试用例编写步骤 #

1. 配置Fake Timers

在测试前后启用/恢复Jest的fake timers，避免影响其他测试：

import { of, lastValueFrom } from 'rxjs';
import { chainedApiWorkflow, firstApi, secondApi } from './your-file-path';

describe('chainedApiWorkflow with RxJS delay', () => {
  beforeEach(() => {
    jest.useFakeTimers(); // 启用假计时器
    jest.clearAllMocks(); // 重置mock状态
  });

  afterEach(() => {
    jest.useRealTimers(); // 恢复真实计时器
  });

2. 验证执行顺序与延迟逻辑

测试核心是：先确认第一个API立即执行，快进时间后再验证第二个API和后续逻辑。

方式一：用subscribe + done回调

适合需要监听流的next/error事件的场景：

it('should run first API immediately, second API after 5s delay', (done) => {
    chainedApiWorkflow().subscribe({
      next: (finalResult) => {
        // 验证最终逻辑结果
        expect(finalResult).toBe('success');
        // 验证第二个API的调用参数正确
        expect(secondApi).toHaveBeenCalledWith({ id: 123, data: 'initial' });
        done(); // 通知Jest测试完成
      },
      error: (err) => done(err) // 捕获错误，避免测试超时
    });

    // 初始状态：第一个API已调用，第二个还没触发
    expect(firstApi).toHaveBeenCalledTimes(1);
    expect(secondApi).not.toHaveBeenCalled();

    // 快进5000毫秒，触发delay后的逻辑
    jest.advanceTimersByTime(5000);
  });

方式二：用lastValueFrom + async/await

更简洁的异步写法，适合只关心最终结果的场景：

it('should handle delay correctly with async/await', async () => {
    // 提前订阅流，获取Promise
    const finalResultPromise = lastValueFrom(chainedApiWorkflow());

    // 验证第一个API立即执行
    expect(firstApi).toHaveBeenCalled();
    expect(secondApi).not.toHaveBeenCalled();

    // 快进时间到delay结束
    jest.advanceTimersByTime(5000);

    // 等待流完成，获取结果并验证
    const result = await finalResultPromise;
    expect(result).toBe('success');
    expect(secondApi).toHaveBeenCalledTimes(1);
  });

3. 测试异常分支

比如第二个API返回错误的情况：

it('should handle second API failure', async () => {
    // 临时mock第二个API返回错误
    jest.spyOn(globalThis, 'secondApi').mockReturnValueOnce(of({ code: 500 }));

    const resultPromise = lastValueFrom(chainedApiWorkflow());
    jest.advanceTimersByTime(5000);

    const result = await resultPromise;
    expect(result).toBe('error');
  });
});

关键注意事项 #

• 必须用异步测试方式（done/async/await），因为Observable是异步流，Jest需要知道测试何时结束。
• 不要忘记在afterEach中恢复真实计时器，否则其他测试可能出现时间相关的bug。
• 如果delay的时长是动态变量，可以把它抽成可配置项，方便测试时直接修改为0，进一步加快测试速度。

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