使用Jasmine的fakeAsync、delay和tick测试Observable状态管道

我有一个用于返回Observable状态的Angular管道，代码如下：

import {Pipe, PipeTransform} from '@angular/core';
import {Observable, of} from 'rxjs';
import {catchError, map, startWith} from 'rxjs/operators';

/** Specifies the status of an Observable. */
export interface ObservableStatus<T> {
  loading?: boolean;
  value?: T;
  error?: boolean;
}

/** Returns the status {@code ObservableStatus} of a given Observable. */
@Pipe({name: 'getObserverStatus'})
export class ObservableStatusPipe implements PipeTransform {
  transform<T = Item>(observer: Observable<T>):
      Observable<ObservableStatus<T>> {
    return observer.pipe(
        map((value: T) => {
          return {
            loading: false,
            error: false,
            value,
          };
        }),
        startWith({loading: true}),
        catchError(error => of({loading: false, error: true})));
  }
}

我想用Jasmine为该管道编写单元测试，尝试了fakeAsync、delay、tick、flush、discardPeriodicTasks等方法，但均未成功。以下是我尝试的几种方式及对应的报错：

尝试方式1 #

describe('loading test', () => {
  const loadingPipe = new ObservableStatusPipe();

  it('returns state of an observable', fakeAsync(() => {
    const input: Observable<Item> = of({name: 'Item'}).pipe(delay(1000));

    const result: Observable<ObservableStatus<Item>> = loadingPipe.transform(input);

    result.subscribe(val => {
      expect(val.loading).toEqual(true);
      expect(val.value).toBeUndefined();
    });
    tick(2000);
    result.subscribe(val => {
      expect(val.loading).toEqual(false);
      expect(val.value!.name).toEqual('Item');
    });
  }));
});

报错信息：

Error: Expected true to equal false. (at expect(val.loading).toEqual(false))
Error: 1 periodic timer(s) still in the queue.

尝试方式2：添加flush清理任务 #

describe('loading test', () => {
  const loadingPipe = new ObservableStatusPipe();

  it('returns state of an observable', fakeAsync(() => {
    const input: Observable<Item> = of({name: 'Item'}).pipe(delay(1000));

    const result: Observable<ObservableStatus<Item>> = loadingPipe.transform(input);

    result.subscribe(val => {
      expect(val.loading).toEqual(true);
      expect(val.value).toBeUndefined();
    });
    tick(2000);
    result.subscribe(val => {
      expect(val.loading).toEqual(false);
      expect(val.value!.name).toEqual('Item');
    });
    flush();     // <----- 此处添加
  }));
});

报错信息：

Error: Expected true to equal false.
TypeError: Cannot read properties of undefined (reading 'name')

其他尝试 #

• 直接测试无延迟的输入Observable时测试通过，但使用flush会报错，改用discardPeriodicTasks则正常
• 将discardPeriodicTasks用于原测试问题，仍失败

另外，我不想使用debounceTime或setTimeout，因为它们会真实等待时间而非模拟时间。尝试在订阅Observable时使用delay操作符，测试可以通过，但疑惑在fakeAsync中该delay是否会真实等待时间？

问题原因分析 #

1. 多次订阅的核心问题：每次调用result.subscribe()都会创建新订阅，第二个订阅会重新触发管道逻辑，此时startWith({loading: true})会再次发送loading状态，导致你拿到的始终是loading=true的结果，而非完成后的value状态。
2. 定时器处理逻辑：delay(1000)创建的宏任务定时器，在fakeAsync中需要用tick或flush推进时间，但错误的订阅时机导致无法捕获正确状态。

解决方案 #

只订阅一次，在回调中收集所有状态后做断言，或者用操作符配合订阅捕获多阶段状态。以下是正确的测试代码：

describe('ObservableStatusPipe', () => {
  let pipe: ObservableStatusPipe;

  beforeEach(() => {
    pipe = new ObservableStatusPipe();
  });

  it('should emit loading first, then value when observable completes', fakeAsync(() => {
    const testItem = {name: 'Item'};
    const input$ = of(testItem).pipe(delay(1000));
    const result$ = pipe.transform(input$);

    let receivedStates: ObservableStatus<typeof testItem>[] = [];

    result$.subscribe(state => {
      receivedStates.push(state);
    });

    // 验证初始loading状态
    expect(receivedStates.length).toBe(1);
    expect(receivedStates[0].loading).toBeTrue();
    expect(receivedStates[0].value).toBeUndefined();

    // 推进时间到delay完成
    tick(1000);

    // 验证完成后的value状态
    expect(receivedStates.length).toBe(2);
    expect(receivedStates[1].loading).toBeFalse();
    expect(receivedStates[1].error).toBeFalse();
    expect(receivedStates[1].value?.name).toBe('Item');

    discardPeriodicTasks();
  }));

  it('should emit loading first, then error when observable throws', fakeAsync(() => {
    const input$ = throwError(() => new Error('test error')).pipe(delay(1000));
    const result$ = pipe.transform(input$);

    let receivedStates: ObservableStatus<unknown>[] = [];

    result$.subscribe(state => {
      receivedStates.push(state);
    });

    // 验证初始loading状态
    expect(receivedStates[0].loading).toBeTrue();

    tick(1000);

    // 验证错误状态
    expect(receivedStates[1].loading).toBeFalse();
    expect(receivedStates[1].error).toBeTrue();
    expect(receivedStates[1].value).toBeUndefined();

    discardPeriodicTasks();
  }));
});

关于fakeAsync中delay的疑问 #

在fakeAsync包裹的测试中，rxjs的delay操作符会被Angular测试框架拦截，不会真实等待时间，而是通过tick或flush模拟时间推进，完全不用担心真实耗时。

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