RxJS Testing — Write Unit Tests for Observables | by Denis Loh | Jun, 2022

Advance your unit testing skills by writing tests with the RxJS testing features.

Photo by Davyn Ben on Unsplash

RxJS is a very powerful and a cool set of tools to create reactive apps. When developing angular applications you will be very likely faced with RxJS, so I assume, you are familiar with how to use it.

However, when I started with Angular about 3 years ago, I got a couple of puzzled looks, when I asked them how to do testing with observables. Here is how I started and how I ended up writing unit tests for observables.

Let’s start with a short example: A function, which takes a sentence and returns it in reverse order.

export function myFunction1(x: string): Observable<string> {
return of(x).pipe(
map((x) => x.split(' ')),
map((x) => x.reverse()),
map((x) => x.join(' '))
);
}

I want to show you how the typical unit tests looked like, when started with unit testing with observables:

describe('RxJS', () => {
it('returns simple value', () => {
const y = myFunction1('humans eat tomatoes');
y.subscribe((value) => expect(value).toBe('tomatoes eat humans'));
});
});

Well, that was an easy one and to no surprise it worked.

Successful test

If you don’t know the movie: it’s the one of the worst horror-movie in human history. Definitely worth to watch it, if you want to waste time…

Okay, let’s make our call asynchronous and add some delay:

export function myFunction1(x: string): Observable<string> {
return of(x).pipe(
map((x) => x.split(' ')),
map((x) => x.reverse()),
map((x) => x.join(' ')),
delay(10)
);
}

The test is obviously successful at a first glance, but when you look closer you can read, that the test has no expectations. This means that our test did not check anything.

Successful tests but without expectations.

You may also get this warning instead:

Failed test due to timeout.

The test simply finished before the emitted value has been evaluated. A simple approach would be to make an asynchronous test out of it:

describe('RxJS', () => {
it('returns simple value', (done: DoneFn) => {
const y = myFunction1('humans eat tomatoes');
y.subscribe((value) => {
expect(value).toBe('tomatoes eat humans')
done()
});
});
});

Well done! It works. By the way: putting async in front of the test case callback will not work unless you run a promisable test case. On the other hand you could use fakeAsync() from the @angular/testing package instead and use tick() to simulate a computation cycle.

But that’s not we want to use for complex RxJS testing. We want to do it the right way: RxJS Testing

So let’s begin with some basics about marble diagrams. If you work with observables, you can consider the emission of the values ​​like a stream or flow. When you apply operators the stream, the values ​​will change and the subsequent stream emit other values. You have some kind of input value, the according operator, which modifies it and the resulting output value (see the example of the map()operator below)

A simple marble diagram

Now you can assign each emitted value a letter like a, b and c. The end of the stream is symbolized with a pipe |. Errors are symbolized with a # . Now it may be the case, that there is some time passed between the emission the values. Time in observables can be represented with virtual frames, which are usually 1ms long and symbolised with a - or with a time progression value like 5s or 100ms . If multiple values ​​are emitted at the same time frame, you can group them together using (). Some examples:

a--b| // emits 'a' then after ~40ms 'b' and completes
(a|) // emits 'a' and completes immediately
a--b-a--c-a--| // emits a series of values
a---# // emits 'a' and errors out after some time.

RxJS comes with a handy tool set to test observables using marble diagrams as explained above.

First of all, we need a TestScheduler which we put at the top of our unit test file:

import { TestScheduler } from 'rxjs/testing'describe('RxJS', () => {
let testScheduler: TestScheduler
beforeEach(() => {
testScheduler = new TestScheduler((actual, expected) => {
return expect(actual).toEqual(expected);
});
});
...
})

Now, we can adjust our unit test again:

describe('RxJS', () => {
it('returns simple value', () => {
const y = myFunction1('humans eat tomatoes');
const expectedMarbles = '----------(a|)'; // (1)
const expectedValues = {
a: 'tomatoes eat humans',
}; // (2)
testScheduler.run(({ expectObservable }) => {
expectObservable(y).toBe(
expectedMarbles,
expectedValues
); // (3)
});
});
});

The test is green again. Hurray! But let’s check what we added here. First of all, we define our expected marble diagram (1).

----------(a|) // after 10 'ticks' the observable emits 'a' and completes immediately after that.

Instead of writing ---------- to represent 10ms of time, you can also directly write 10ms (a|) .

Then, we define our expected values ​​(2), where we create an object structure with the keys from our marble diagram and the according value.

const expectedValues = {
a: 'tomatoes eat humans',
};

And finally, we let the test scheduler run (3).

testScheduler.run(({ expectObservable }) => {
/* expectObservable is a helper, which subscribes to an obserable
* and performs the callback which we passed to the TestScheduler
* constructor.
*/
expectObservable(y).toBe(
expectedMarbles,
expectedValues
);
});

Now, when everything is fine, the unit test is green and we are all good. However, it is usually not the case, that things work out immediately. So, what do we get, if something goes wrong?

Let’s have a look what our testing scheduler is actually comparing. So for recap, this is our comparator:

(actual, expected) => { return expect(actual).toEqual(expected) }

It is a deep equality check of actual and expected, which are of type any duh!
Anyway, this check is actually verifying a couple of things, which may happen during the subscription.

There are 4 types of issues:

  • Length of frames (ie number of emitted unexpected items)
  • Number of frame (ie the expected time frame is unexpected)
  • Kind of frame (ie item, completion, error, subscription, unsubscription)
  • Frame value (ie emitted item value or error message)

Assume, we forgot something in our function:

export function myFunction1(x: string): Observable<string> {
return of().pipe( // we forgot to call of(x)
map((x) => x.split(' ')),
map((x) => x.reverse()),
map((x) => x.join(' ')),
delay(10)
);
}

This is, what we get then:

All kind of errors

Frames length incorrect

Expected $.length = 1 to equal 2.

The above error tells, that we expected two frames, but got only one. So you get Expected $.length = X to equal Y when you either emit more or less items than expected.

Frame number incorrect

Expected $[0].frame = 0 to equal 10.

The above error tells, that the frame number is incorrect. So you get Expected $[i].frame = X to equal Y where i is the index of the frame, ie in our example the first frame.

Frame type incorrect

Expected $[0].notification.kind = 'C' to equal 'N'.

The above error tells, that the kind or type of frame is incorrect. So you get Expected $[i].notification.kind = '...' to equal '...' if the stream emitted another type of signal. Typical signals are:

  • N = Emitted item
  • C = Completion
  • E = Error

In our case, we expected an item on the first frame, but got a completion signal.

Frame value incorrect

Expected $[0].notification.value = undefined to equal 'tomatoes eat humans'.Expected $[1] = undefined to equal Object({ frame: 10, notification: Object({ kind: 'C', value: undefined, error: undefined }) }).

The above error tells, that the value of the frame / item is incorrect. So you get Expected $[i].notification.value = ... to equal ... if the actual value is incorrect. In the second error you can see, that this kind of error applies also to a complete frame. In our example, the value of the first frame is undefined, but should hold ‘tomatoes eat humans’ and the second frame does not exist at all.

Sometimes you have event-driven use-cases where you have triggers, which causes any kind of action and an event pipe, where you can listen for updates.

The following example is a BehaviorSubject with an initial value. Everytime we call the add or minus function, we use the scan operator to calculate the sum out of the current value and the new value. The two function might be bound to buttons and an input field, for example.

const someSubject = new BehaviorSubject<number>(0);export const eventObservable = someSubject.pipe(
scan((acc, val) => acc + val)
);
export function add(number: number) {
someSubject.next(number);
}
export function minus(number: number) {
someSubject.next(-number);
}

Let’s write the unit test for this:

it('returns event values', () => {
const expectedMarbles = 'a';
const expectedValues = {
a: 0
};
testScheduler.run(({ expectObservable }) => {
expectObservable(eventObservable).toBe(
expectedMarbles,
expectedValues
);
});
});

This is the result, which is so good:

Successful test for subjects

But how do we now trigger the function calls? A naive implementation would be something like this:

it('returns event values', () => {
const expectedMarbles = 'abc';
const expectedValues = {
a: 0,
b: 5,
c: 7
};
setTimeout(() => add(5), 1)
setTimeout(() => add(2), 2)
testScheduler.run(({ expectObservable }) => {
expectObservable(eventObservable).toBe(
expectedMarbles,
expectedValues
);
});
});

But this will not work. There are no items emitted at all. Why that? Essentially because the testScheduler is running in its own synchronous scope and therefore the two tasks of setTimeout are executed after the test.

Failed test when using triggers

To fix this, we have to run the triggers within the testScheduler scope. The testScheduler provides some additional helpers beside the expectObservable . You can find the documentation about them here: https://rxjs.dev/guide/testing/marble-testing#api

With triggerMarbles and triggerValues we can create function calls, which will happen at the exact time frame, we need it.

const triggerMarbles = '-abcabc';
const triggerValues = {
a: () => add(5),
b: () => add(2),
c: () => minus(7),
};

We will now need the cold helper to run these triggers:

testScheduler.run(({ expectObservable, cold }) => {
...
expectObservable(
cold(triggerMarbles, triggerValues).pipe(tap(fn => fn()))
);
});

expectObservable subscribes (and also unsubscribes) to our cold observable. With the .pipe(tap(fn => fn())) we run the corresponding trigger function.

Now, we can adjust the expected output:

const expectedMarbles = 'abcabca';
const expectedValues = {
a: 0,
b: 5,
c: 7,
};

Putting everything together, we have this unit test:

it('returns event values', () => {
const expectedMarbles = 'abcabca';
const expectedValues = {
a: 0,
b: 5,
c: 7,
};
const triggerMarbles = '-abcabc';
const triggerValues = {
a: () => add(5),
b: () => add(2),
c: () => minus(7),
};
testScheduler.run(({ expectObservable, cold }) => {
expectObservable(eventObservable).toBe(
expectedMarbles,
expectedValues
);
expectObservable(
cold(triggerMarbles, triggerValues).pipe(tap(fn => fn()))
);
});
});

You can read about the known limitations here: Known issues. However, I want to summarise some of them, because it may be not that obvious.

Promises / setTimeout()

Promises are not supported as an observable source, ie something like the code below cannot be asserted with the testScheduler :

from(Promise.resolve('something'))

This means, that also every API, which returns a promise like fetch() or async functions which are wrapped into observables. To make these work, you have to have to fall back to asynchronous testing as explained above, eg using done().

You can find the examples above in this stackblitz project here: https://stackblitz.com/edit/rxjs-testing-kit?file=main.ts

I hope this will help you to write unit tests with RxJS observables.

Photo by the blowup on Unsplash

Leave a Comment