RxJS Observable抛错时Jest测试等待即失败的解决方法

问题描述 #

在 RxJS 7.8.0 和 Jest 29.5.0 环境下，测试 Observable 时出现以下异常：

• 测试中加入等待逻辑（包括setTimeout(done, 0)），Observable 抛出错误时测试直接失败
• 未添加等待逻辑的测试可正常通过
• try/catch包裹订阅无法捕获错误，也无法阻止测试失败

复现代码 #

import { throwError } from 'rxjs';

describe('Unhandled Promise rejection', () => {
    it('fails when waiting for a bit', (done) => {
        const error = new Error('Error waiting for 1000');
        const obs$ = throwError(() => error);

        try {
            obs$.subscribe();
        } catch (e) {
            console.warn('The error was emitted: %s', e);
        }

        setTimeout(done, 1000);
    });

    it('fails when waiting for 0ms', (done) => {
        const error = new Error('Error waiting for 0');
        const obs$ = throwError(() => error);

        try {
            obs$.subscribe();
        } catch (e) {
            console.warn('The error was emitted: %s', e);
        }

        setTimeout(done, 0);
    });

    it('passes when not waiting', () => {
        const error = new Error('Error not waiting');
        const obs$ = throwError(() => error);

        try {
            obs$.subscribe();
        } catch (e) {
            console.warn('The error was emitted: %s', e);
        }
    });
});

测试结果 #

前两个带等待逻辑的测试失败，控制台未打印警告：

Error: Error waiting for 1000
    at test.spec.ts:5:17

Error: Error waiting for 0
    at test.spec.ts:18:17

第三个无等待的测试正常通过。

解决方案 #

1. 为订阅添加错误回调 #

Observable 的错误是异步触发的，try/catch无法捕获，必须通过订阅的错误处理回调处理：

it('passes with error callback', (done) => {
    const error = new Error('Error waiting for 1000');
    const obs$ = throwError(() => error);

    obs$.subscribe({
        error: (e) => {
            console.warn('The error was emitted: %s', e);
            done(); // 标记测试完成
        }
    });
});

2. 使用 RxJS 的 lastValueFrom/firstValueFrom 结合 async/await #

RxJS 7 中toPromise()已废弃，推荐用lastValueFrom或firstValueFrom将 Observable 转换为 Promise，通过async/await捕获错误：

import { lastValueFrom, throwError } from 'rxjs';

it('passes with lastValueFrom', async () => {
    const error = new Error('Error waiting for 1000');
    const obs$ = throwError(() => error);

    try {
        await lastValueFrom(obs$);
    } catch (e) {
        console.warn('The error was emitted: %s', e);
    }
});

3. 结合 Jest 断言验证错误（可选） #

如果需要验证错误是否正确抛出，可配合 Jest 的expect.assertions()确保断言执行：

it('verifies error is thrown correctly', async () => {
    const error = new Error('Error waiting for 1000');
    const obs$ = throwError(() => error);

    expect.assertions(1);
    try {
        await lastValueFrom(obs$);
    } catch (e) {
        expect(e).toBe(error);
    }
});

4. 全局处理未捕获的 RxJS 错误（不推荐） #

若不想每个测试都单独处理错误，可全局设置 RxJS 错误处理器，但可能掩盖潜在问题：

import { setErrorHandler } from 'rxjs';

beforeAll(() => {
    setErrorHandler((err) => {
        console.warn('Global RxJS error caught:', err);
        // 不抛出错误，避免 Jest 测试失败
    });
});

内容的提问来源于stack exchange，提问作者Aisteru Firë