单元测试 Observable.interval

Question

我有一个函数可以轮询 api，直到返回所需的结果或达到超时期限。

// sample poll, filter and timeout
   pollStatus = (): Observable<string> => {
            return Observable.interval(1000)
              .mergeMap(() => this.getStatus(); // this makes a http call to a status api
              .map((response) => response.json())
              .filter((responseData) => responseData.status !== 'PENDING')
              .take(1)
              .map((response) => response.status)
              .catch((error) => Observable.throw({message: 'FAILED'}))
              .timeout(5000);
          };

在我的测试中，我想执行 pollStatus() 函数并确保 getStatus() 函数已被正确调用次数等，但我无法让它调用一次。测试 Observable.interval 的最佳方法是什么？如何强制出现新的间隔？

Answer 1

为了测试 Rx 代码，您需要传递 VirtualTimeScheduler，以便您可以提前测试时间。您可以使用 compat 包 @kwonoj/rxjs-testscheduler-compat，这样您就可以在 RxJs5 中编写测试，就像在 RxJs4 中编写它们一样：

import { TestScheduler, next, complete } from '@kwonoj/rxjs-testscheduler-compat';

const scheduler = new TestScheduler();
const results = scheduler.startScheduler(
  () => pollStatus(scheduler),
  { created: 100, subscribed: 200, unsubscribed: 4000 }
);

您需要将调度程序传递给基于时间的操作员，以便他们不会在默认调度程序上而是在您提供的调度程序上安排工作：

pollStatus = (scheduler?:Rx.Scheduler): Observable<string> => {
  return Observable.interval(1000, scheduler)
    .mergeMap(() => this.getStatus(); // this makes a http call to a status api
    .map((response) => response.json())
    .filter((responseData) => responseData.status !== 'PENDING')
    .take(1)
    .map((response) => response.status)
    .catch((error) => Observable.throw({message: 'FAILED'}))
    .timeout(5000, scheduler);
};

然后您可以观察到您的代码在您的测试中以虚拟时间运行，然后您可以开始为 this.getStatus() 创建模拟，使其慢于 5000 毫秒，以便您的 timeout() 命中等。