如何对正在进行的 RXJS 请求的取消进行单元测试

Posted 2023-03-23

【中文标题】如何对正在进行的 RXJS 请求的取消进行单元测试

【英文标题】：How to unit test cancellation of in-flight RXJS requests

【发布时间】：2019-05-04 08:15:46

【问题描述】：

我正在使用 NGRX，并使用 Effects 发送 HTTP 请求。如果用户发送另一个请求，则应取消任何先前的请求。当我手动测试时它工作正常，但我希望能够对此进行单元测试。为了测试这一点，我模拟了发送 HTTP 请求并在一定延迟后发送响应的服务。然后，我有一个触发 4 个请求的 Marble hot observable。我希望我的效果只触发一次。但是，它根本没有被触发。

单元测试：

it('should only do one request at a time', fakeAsync(() => 
    // Arrange
    const data = createTestData();
    const dataServiceSpy = TestBed.get(DataService);
    dataServiceSpy.getData = jest.fn(
        (query: DataQuery) => 
            const waitTime = 1000 * + query.index;
            return of(assets).pipe(delay(waitTime));
        
    );         
    // Act
    actions = hot('-abcd-|', 
        a: new SearchData( index: '6' ),
        b: new SearchData( index: '5' ),
        c: new SearchData( index: '4' ),
        d: new SearchData( index: '1' )
    );

    tick(10000);

    // Assert
    expect(effects.loadData$).toBeObservable(
        hot('-a-|',  a: new SearchDataComplete(assets) )
    );
));

所以，我发送了 4 个搜索请求；第一个应该在 6 秒后返回数据，第二个应该在 5 秒后返回数据，依此类推。 但是，我的单元测试失败了 loadData$ 是一个空的可观察对象，而它期望有一个项目。

如果我删除了 spy 中的延迟，它会按预期工作，并且 loadData$ 包含 4 个结果。

我的效果正在使用 NX DataPersistence，如果您提供 id 函数，它会负责取消；如果新的操作带有相同的 id，它将取消初始请求。类似使用 this.actions$.pipe(switchMap(...))

@Effect()
loadData$ = this.dataPersistence.fetch(ActionTypes.SearchData, 
    id:  (action, state) => 
        return action.type
    ,

    run: (action, state) => 
        return this.dataService
            .searchData(action.payload)
            .pipe(                   
                map(data => new SearchDataComplete(data))
            );
    ,

【问题讨论】：

不确定这是否是一个错字，但你嘲笑了getData，你的效果是searchData

感谢您的回复！我已经匿名化了函数名称，确实打错了。我的测试中的实际函数名称是正确的。

【参考方案1】：

所以我对此进行了一些研究。我有两个想法：

在单元测试中，我们真的只想测试我们编写的代码。如果我使用的是第三方库，我会假设它经过了适当的单元测试（例如，通过查看该库的源代码）。 DataPersistence 的单元测试现在不测试取消（因为我们正在使用 switchMap 并假设它的功能有效）。 在示例中尝试使用 delay 进行测试时存在实际问题

在您的测试中，tick 在订阅效果之前触发（当您在下方调用 expect 时）。

一种解决方法如下：

describe('loadTest$', () => 
    it('should only do one request at a time', () => 

      // create a synchronous scheduler (VirtualTime)
      const scheduler = new VirtualTimeScheduler();

      // Arrange
      const data = createTestData();
      const dataServiceSpy = TestBed.get(DataService);

      TestBed.configureTestingModule(
        imports: [NxModule.forRoot(), StoreModule.forRoot()],
        providers: [
          TestEffects,
          provideMockActions(() => actions),
          
            provide: DataService,
            useValue: 
              searchData: jest.fn(
                  (query: DataQuery) => 
                      const waitTime = 1000 * + query.index;
                      return of(assets).pipe(delay(waitTime, scheduler));
                  
              )
            
          
        ]
      );

      actions = of(
          new SearchData( index: '6' ),
          new SearchData( index: '5' ),
          new SearchData( index: '4' ),
          new SearchData( index: '1' )
      );

      const res = [];
      effects.loadData$.subscribe(val => res.push(val));

      scheduler.flush(); // we flush the observable here

      expect(res).toEqual([ a: new SearchDataComplete(assets) ]);
    );
);

我们可以使用同步调度器并手动刷新它。通常我们不需要这样做；这只是因为我们需要delay（它也会发生在其他需要调度程序的运营商身上，比如debounceTime）。

希望这会有所帮助。我认为您的测试不需要测试底层库的功能（那时它可能不是严格的单元测试，而是更多的集成测试）。