rxjs - 当您订阅 rxjs 时，如果测试失败，您如何向测试发出信号？

Question

我是一个完整的初学者。我遇到的问题是，一旦我在 rxjs observable 中抛出错误，我的测试就不知道了。当我订阅测试时，它在 rxjs 中失败了，它只会抛出一个错误，我需要通知我的测试发生了错误。这是一个更简单的示例，它表明从未打印过“测试失败”。

import { sample } from "rxjs/operators";
const source = interval(1000);
// sample last emitted value from source every 2s
// output: 2..4..6..8..
const example = source.pipe(sample(interval(2000)));
async function test_runner() {
    setup();
    try {
        await test();
        console.log("test succeeded");
    } catch (e) {
        console.log("test failed");
    }
}
async function setup() {
    console.log("setup");
    const subscribe = example.subscribe((val) => {
        console.log(val);
        if (val === 4) {  throw Error("error!"); }
    });
}
async function test() {
    console.log("test");
    await waitMs(10000);
}

test_runner();

async function waitMs(waitTime: number): Promise<void> {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            resolve();
        }, waitTime);
    });
}

有没有办法处理这个？我很感激任何帮助。

Answer 1

如果你想测试 rx 流，最好的方法之一是使用弹珠图。这就是 ngrx 用于效果测试的内容。

https://www.npmjs.com/package/jasmine-marbles https://github.com/ngrx/platform/blob/master/docs/effects/testing.md

使用弹珠图，您可以在期望发出/错误的地方编写样式并对其进行断言。

例如，语法hot('---#')意味着 30 毫秒后流中出现错误。

Answer 2

订阅时，您可以将函数传递给：

• 处理流发出的项目
• 处理错误
• 处理完成信号

您也可以在测试中使用它：

describe('when a stream emits an error', () => {
  it('should call your error handler', () => {
    const stream$ = throwError('wat?!');
    stream$.subscribe({ error: (err) => {
      chai.expect(err === 'wat?!').to.be.true;
    }});
  });
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js"></script>
<script src="https://unpkg.com/chai/chai.js"></script>
<script src="https://unpkg.com/mocha/mocha.js"></script>
<script>const {throwError} = rxjs;</script>

<div id="mocha"></div>

<script class="mocha-init">mocha.setup('bdd');</script>
<script class="mocha-exec">mocha.run();</script>