如何在 rxjs 中对重试当运算符进行单元测试?

Question

我正在尝试对自定义 RxJS 运算符进行单元测试。该运算符非常简单，它使用 RetryWhen 重试失败的 HTTP 请求，但有延迟，并且仅在 HTTP 错误在 500 范围内时才重试。使用茉莉花，这是在 Angular 应用程序中。

我看过这个：

使用重试的 rxjs 单元测试何时

不幸的是，更新 SpyOn 调用似乎不会更改连续重试时返回的可观察量。每次重试时，它都会使用原始间谍值重试。

我还看过一堆 rxjs 大理石示例，这些示例似乎都不起作用。我不确定是否可以在这里使用 rxjs 弹珠，因为 (AFAIK( 无法模拟您首先提交错误的可观察量，然后在后续尝试中提交成功的可观察量的情况。

代码基本上是这个的克隆： https://blog.angularindepth.com/retry-failed-http-requests-in-angular-f5959d486294

export function delayedRetry(delayMS: number, maxRetry) {
let retries = maxRetry;
return (src: Observable<any>) =>
src.pipe(
retryWhen((errors: Observable<any>) => errors.pipe(
delay(delayMS),
mergeMap(error =>
(retries-- > 0 && error.status >= 500) ? of(error) : throwError(error))
))
);
}

我希望能够证明它可以订阅一个在第一次尝试时返回错误但随后返回成功响应的可观察量。最终订阅应显示可观察发出的成功值。

提前感谢您的任何见解。

Answer 1

尝试将此可观察量用作源可观察量进行测试

const source = (called,successAt)=>{
return defer(()=>{
if(called<successAt){
called++
return throwError({status:500})
}
else return of(true)
})
}

测试

this.delayedRetry(1000,3)(source(0,3)).subscribe()

Answer 2

若要测试重试功能，需要一个每次调用时都会发出不同事件的可观察量。例如：

let alreadyCalled = false;
const spy = spyOn<any>(TestBed.inject(MyService), 'getObservable').and.returnValue(
new Observable((observer) => {
if (alreadyCalled) {
observer.next(message);
}
alreadyCalled = true;
observer.error('error message');
})
);

此可观察量将首先发出错误，然后发出下一个事件。

您可以检查您的可观察对象是否收到如下消息：

it('should retry on error', async(done) => {
let alreadyCalled = false;
const spy = spyOn<any>(TestBed.inject(MyDependencyService), 'getObservable').and.returnValue(
new Observable((observer) => {
if (alreadyCalled) {
observer.next(message);
}
alreadyCalled = true;
observer.error('error message');
})
);
const observer = {
next: (result) => {
expect(result.value).toBe(expectedResult);
done();
}
}
subscription = service.methodUnderTest(observer);
expect(spy).toHaveBeenCalledTimes(1);
}

Answer 3

在之前的答案的基础上，我一直在使用这个，它可以让您更好地控制返回的内容。

const source = (observables) => {
let count = 0;
return defer(() => {
return observables[count++];
});
};

然后可以这样使用

const obsA = source([
throwError({status: 500}),
of(1),
]);

或者它可以与 rxjs 弹珠一起使用，例如

const obsA = source([
cold('--#', null, { status: 500 }),
cold('--(a|)', { a: 1 }),
]);