unit-testing - 测试 RXJS 自定义管道

Question

我有一个自定义管道，它只是多个管道的集合。但是，我想测试代码。

export type AnnotatedResponse<T> = T & {
    awaitingNewValues: boolean;
    failed: boolean;
};

function merge<T>(
    original: T,
    awaitingNewValues: boolean,
    failed: boolean,
): AnnotatedResponse<T> {
    return Object.assign({},
        original,
        { awaitingNewValues, failed },
    );
}

export function annotateAwaiting<I, T extends {}>(
    initialValue: T,
    toAnnotate: (input: I) => Observable<T>,
): (source: Observable<I>) => Observable<AnnotatedResponse<T>> {
    return (source: Observable<I>) => {
        let lastValue: T = initialValue;
        return source.pipe(
            switchMap(input => concat(
                of(merge(lastValue, true, false)),
                toAnnotate(input).pipe(
                    take(1),
                    map(result => {
                        lastValue = result;
                        return merge(result, false, false);
                    }),
                    catchError(() => {
                        return of(merge(lastValue, false, true));
                    }),
                ),
            )),
            finalize(() => {
                //  Free reference to prevent memory leaks
                lastValue = undefined as any;
            }),
        );
    };
}

测试

fdescribe('annotateAwaiting$', () => {
    let subject: Subject<string>;
    let annotated: Observable<AnnotatedResponse<{ letter: string }>>;
    let response: Subject<Error | { letter: string }>;

    beforeEach(() => {
        subject = new Subject<string>();
        response = new Subject();
        annotated = subject.pipe(
            annotateAwaiting(
                { letter: '' },
                letter => response.pipe(
                    take(1),
                    map(val => {
                        if (val instanceof Error) {
                            throw val;
                        } else {
                            return val;
                        }
                    }),
                ),
            ),
            shareReplay(1),
        );
    });

    afterEach(() => {
        subject.complete();
        response.complete();
    });

    it('should emit immediately', () => {
        let i = 0;
        subject.next('a');
        annotated.pipe(
            take(1),
        ).subscribe(obj => {    //  Runs
            expect(obj.awaitingNewValues).toBe(true);
            expect(obj.failed).toBe(false);
            expect(obj.letter).toBe('');
            i++;
        });

        expect(i).toBe(1);

        response.next({ letter: 'a'});
        annotated.pipe(
            take(1),
        ).subscribe(obj => {    //  Doesn't run (neither async)
            expect(obj.awaitingNewValues).toBe(false);
            expect(obj.failed).toBe(false);
            expect(obj.letter).toBe('a');
            i++;
        });

        expect(i).toBe(2);
    });
});

我错过了什么？为什么它似乎根本没有运行？

Answer 1

这次没那么难，只是逻辑有问题。

it('should emit immediately', () => {
    let i = 0;

    annotated.pipe(
        // Adding this skip makes sure that it does not subscribes early
        skipWhile(r => !r.awaitingNewValues),
        take(1),
    ).subscribe(obj => {  // Runs
        expect(obj.awaitingNewValues).toBe(true);
        expect(obj.failed).toBe(false);
        expect(obj.letter).toBe('');

        i++;
    });
    subject.next('a');
    expect(i).toBe(1);

    annotated.pipe(
        // Same here...
        skipWhile(r => r.awaitingNewValues),
        take(1),
    ).subscribe(obj => {
        expect(obj.awaitingNewValues).toBe(false);
        expect(obj.failed).toBe(false);
        expect(obj.letter).toBe('a');

        i++;
    });
    response.next({ letter: 'a'});
    expect(i).toBe(2);
});