使用Jest单元测试时RxJS Observable抛出的异常被忽略

问题原因 #

1. RxJS错误处理逻辑：在tap操作符里抛出的异常，会被RxJS自动转化为Observable的error通知，不会直接抛到当前同步执行栈中。
2. 订阅错误处理的影响：
   • 如果订阅未配置error回调，RxJS会把错误抛到全局上下文，而Jest的测试逻辑已经走到赋值语句之后，不会关联这个全局错误，导致测试继续执行。
   • 如果添加了error回调（比如你代码里的console.error），错误会被回调捕获处理，自然不会触发测试中断。
3. Jest错误捕获范围：Jest默认只捕获测试函数同步执行栈内的错误，以及通过done回调、async/await处理的异步错误。RxJS流内的错误如果没关联到测试上下文，就不会被识别为测试失败。

解决方法 #

方法1：让验证错误在同步执行栈抛出（推荐，符合业务逻辑直觉） #

把父类的验证逻辑从RxJS流中移出，放到子类的value setter中同步执行，设置值时直接抛出异常，Jest能立刻捕获：

修改Child.ts，新增验证回调注册机制：

class Child {
  public id: string;
  private _value: string;
  private _valueChangeNotifier = new Subject<string>();
  private _validators: ((value: string) => void)[] = [];

  get onValueChange$() {
    return this._valueChangeNotifier.asObservable();
  } 

  get value() {
    return this._value;
  }

  set value(newValue: string) {
    // 先执行所有验证，失败直接抛出异常
    this._validators.forEach(validator => validator(newValue));
    this._value = newValue;
    this.notifyChangeOfValue();
  }

  // 供父类注册验证逻辑
  addValidator(validator: (value: string) => void) {
    this._validators.push(validator);
  }

  private notifyChangeOfValue() {
    this._valueChangeNotifier.next(this._value);
  }
}

修改Parent.ts的构造器，注册同步验证：

class Parent {
  private _children: Child[] = [];

  constructor(parentSetup: any) {
    for (const childSetup of parentSetup.childrenSetup) {
      const child = ChildBuilder.create(childSetup);
      // 注册同步验证回调
      child.addValidator(newValue => this.throwErrorIfTheNewValueHasSomethingWrong(newValue));
      // 订阅值变更做后续转换
      child.onValueChange$
        .subscribe({
          next: newValue => this.transformValueForAFutureGenerationOfOutput(newValue)
        });
      this._children.push(child);
    }
  }

  private throwErrorIfTheNewValueHasSomethingWrong(newValue: string) {
    // 你的验证逻辑，失败时抛出异常
    if (newValue.includes('invalid')) {
      throw new Error('Invalid value');
    }
  }

  getChildById(id: string) {
    return this._children.find(child => child.id === id);
  }

  // 其他方法...
}

方法2：在父类订阅的error回调中重新抛出错误 #

如果不想修改业务逻辑结构，可以在订阅的error回调中把错误重新抛出，利用RxJS同步执行的特性，让错误进入Jest的捕获范围：

constructor(parentSetup: any) {
  this._children = [];
  for (const childSetup of parentSetup.childrenSetup) {
    const child = ChildBuilder.create(childSetup);
    child.onValueChange$
      .pipe(
        tap(newValue => this.throwErrorIfTheNewValueHasSomethingWrong(newValue))
      )
      .subscribe({
        next: newValue => this.transformValueForAFutureGenerationOfOutput(newValue),
        error: (error) => {
          console.error(error);
          throw error; // 重新抛出错误，让Jest捕获
        }
      });
    this._children.push(child);
  }
}

方法3：在测试中主动捕获RxJS错误 #

如果无法修改业务代码，可以在测试中通过done回调监听RxJS的错误：

it('should test parent output after child value change', (done) => {
  const parent = ...// 初始化父类

  // 监听全局错误，转发给Jest的done回调
  const originalOnError = global.onerror;
  global.onerror = (_, __, ___, ____, error) => {
    global.onerror = originalOnError;
    done(error); // 标记测试失败
    return true;
  };

  try {
    parent.getChildById('<CHILD_UUID>').value = 'invalid value';
    const output = parent.generateSomeOutputBasedOnChildrenAndTheirValues();
    expect(output).toMatchObject({ ...myExpectedOutput });
    done();
  } catch (error) {
    global.onerror = originalOnError;
    done(error);
  }
});

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