Testing RxJS streams with rxjs-for-await

Modified August 15, 2022

Tim Deschryver

timdeschryver.dev

While most of the examples that I've seen are using marbles diagrams to assert that a RxJS stream is behaving as intended, I'm of the opinion that most of these tests don't benefit from the marble diagram syntax.

The reason why I think that, is because marbles are testing implementation details and this isn't useful for us as consumers. We can expect that each operator is tested (within the producer's codebase, e.g. the RxJS repository) to verify that it operates in an expected way. When we, as consumers. write tests we should only be interested, for the most part, in the output of the stream. I do think that marble diagrams are useful when you're writing your custom operators because then it's important to test these implementation details. Besides, I can say from my experience that it can take a while to understand your first marble diagram and take up some more time to write your first test.

Because of this, I used rxjs-for-await to write the tests for rx-query. In this post, we'll write a test using rxjs-for-await and then compare it with other RxJS testing strategies. I hope this simple example illustrates why I do prefer these tests of marble tests.

As an example, let's test a simple map to map a value to its corresponding letter in the alphabet, over a certain amount of time. The example code looks like this:

 content_paste
const ALPHABET = 'abcdefghijklmnopqrstuvwxyz';
 
const toAlphabet = (): OperatorFunction<any, string> => (source) =>
  source.pipe(map((v) => ALPHABET[v]));
 
const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));

rxjs-for-await link

rxjs-for-await written by Ben Lesh doesn't have its primary focus on making testing easy, but it's created to support the async/await syntax. A happy coincidence, because this does come in handy while writing tests because we can just await the stream to complete and verify the outcome.

When you're writing tests with rxjs-for-await it's important to complete the stream. Otherwise, it will just keep waiting until the stream completes, resulting in a timeout error.

 content_paste
import { eachValueFrom } from 'rxjs-for-await';
 
test('rxjs-for-await', async () => {
  const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));
 
  const result = [];
  for await (const value of eachValueFrom(source)) {
    result.push(value);
  }
 
  expect(result).toStrictEqual(ALPHABET.split(''));
});

When we compare this test to other RxJS testing strategies, using the marble syntax we can clearly see the differences between the two and and each other's focus.

RxJS Testing link

The TestScheduler is provided by RxJS to write tests using the marble diagram syntax. In the test case below, you can see that every frame of the stream is asserted, each frame that is just waiting, and each frame that emits a value. This is important when you're writing operators, and that's probably why this approach is used in the RxJS codebase to test the RxJS operators. But it also makes things more complex, especially if you're not used to reading or writing marble diagrams.

format_quote

For more information about RxJS (and marble) testing, I would like to refer you to Jay Phelps's talk Don’t Lose Your Marbles, We Can Test RxJS Code at RxJS live.

 content_paste
import { TestScheduler } from 'rxjs/testing';
 
test('TestScheduler', () => {
  const scheduler = new TestScheduler((actual, expected) => expect(actual).toEqual(expected));
 
  scheduler.run(({ expectObservable }) => {
    const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));
    const expected =
      '100ms a 9ms b 9ms c 9ms d 9ms e 9ms f 9ms g 9ms h 9ms i 9ms j 9ms k 9ms l 9ms m 9ms n 9ms o 9ms p 9ms q 9ms r 9ms s 9ms t 9ms u 9ms v 9ms w 9ms x 9ms y 9ms z 9ms |';
    expectObservable(source).toBe(expected);
  });
});

jasmine-marbles link

Now, let's take a look at how the same test is written with the popular jasmine-marbles package. It uses the same syntax to write the marble diagrams, but it doesn't require the setup to create the TestScheduler.

By using jasmine-marbles, we end up with the following test.

 content_paste
import { cold } from 'jasmine-marbles';
 
test('jasmine-marbles', () => {
  const sourceValues = {
      a: 0,
      b: 1,
      c: 2,
      ...
  };
  const source = cold(
    '100ms a 9ms b 9ms c 9ms d 9ms e 9ms f 9ms g 9ms h 9ms i 9ms j 9ms k 9ms l 9ms m 9ms n 9ms o 9ms p 9ms q 9ms r 9ms s 9ms t 9ms u 9ms v 9ms w 9ms x 9ms y 9ms z 9ms |',
    sourceValues,
  )
  const result = source.pipe(toAlphabet())
  const expected = cold(
    '100ms a 9ms b 9ms c 9ms d 9ms e 9ms f 9ms g 9ms h 9ms i 9ms j 9ms k 9ms l 9ms m 9ms n 9ms o 9ms p 9ms q 9ms r 9ms s 9ms t 9ms u 9ms v 9ms w 9ms x 9ms y 9ms z 9ms |',
  )
  expect(result).toBeObservable(expected)
})

rxjs-marbles link

Another great library from Nicholas Jamieson, called rxjs-marbles. It can be compared with jasmine-marbles as it also gets rid of the setup and it also uses the same marble diagram syntax. The main difference is that rxjs-marbles can be used across all test frameworks. That's the reason why its API is different compared to jasmine-marbles.

 content_paste
import { marbles } from 'rxjs-marbles';
 
test(
  'rxjs-marbles',
  marbles((m) => {
    const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));
    const expected =
      '100ms a 9ms b 9ms c 9ms d 9ms e 9ms f 9ms g 9ms h 9ms i 9ms j 9ms k 9ms l 9ms m 9ms n 9ms o 9ms p 9ms q 9ms r 9ms s 9ms t 9ms u 9ms v 9ms w 9ms x 9ms y 9ms z 9ms |';
 
    m.expect(source).toBeObservable(expected);
  }),
);

Main differences and my thoughts link

I think that marble tests are great for testing implementation details, every frame is important while writing your own RxJS operators. The only way to test these, in a descriptive way, is by using marble tests. This is clearly showcased in the RxJS repository.

But the advantages don't outweigh the disadvantages when it comes to testing most of the code that lives inside an application. Here, we should be interested in the output of a stream. Marble diagrams are also not helping to enlarge the pit of success, I haven't encountered a developer yet that immediately gets marble diagrams and feel comfortable to write their first test cases with it. Most of the tests I've seen in application code are also brittle to change, e.g. they fail when a detail, for example a timer duration, is modified.

Alex Okrushko did point out a drawback to this approach. Some tests will take longer to run (and might even time-out), or you will have to mock the timers. This is crucial when you have timers that wait multiple seconds. This is not the case with marble tests, and also with ObserverSpy.

For more info about fake times, see https://ncjamieson.com/testing-with-fake-time/ by Nicholas Jamieson, and see the ObserverSpy docs.

That's why I like rxjs-for-await. It helps to reduce the complexity of writing and reading RxJS tests. Simply put, it's simply input in, and output out. That's why I wrote all the tests for rx-query using the rxjs-for-await package. For more use-cases, check out the test cases inside the repository.

Alternatives link

ObserverSpy link

This is a new library, written by Shai Reznik not too long ago. This was my first look into ObserverSpy, and I think it's great. It solves the same problem that I had with marble tests, and I couldn't have said it any better than Shai:

format_quote

Marble tests are very powerful, but at the same time can be very complicated to learn and to reason about for some people. You need to learn and understand cold and hot observables, schedulers and to learn a new syntax just to test a simple observable chain.

The only, small, downside compared to rxjs-for-await is that you have to learn a new API, whereas rxjs-for-await is just using the JavaScript async/await feature. The plus side of usingObserverSpy is that it has useful helper methods to read the output values of the stream.

 content_paste
import { subscribeSpyTo } from '@hirez_io/observer-spy';
 
test('ObserverSpy', async () => {
  const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));
  const observerSpy = subscribeSpyTo(source);
 
  await observerSpy.onComplete();
 
  expect(observerSpy.getValues()).toStrictEqual(ALPHABET.split(''));
});

Manually link

You don't need an extra dependency to test your RxJS streams. The output of a stream can be tested via the callbacks provided by a subscription, for example, the complete callback.

But be careful when you go down this road, as this might lead to false positives. In the test below it's crucial to use the done callback (from your test framework). If you don't use the done callback, the test completes before the complete callback is invoked, and thus will the assertion inside of the complete callback never be tested. In other words, the test will always pass, even when it shouldn't.

This will probably lead to false-positives, as it was in my case. That's why I suggest you to take a look at rxjs-for-await or ObserverSpy. This way you can have confidence in the tests you write.

 content_paste
test('manual', (done) => {
  const source = timer(100, 10).pipe(toAlphabet(), takeWhile(Boolean));
 
  const result = [];
  source.subscribe({
    next: (value) => result.push(value),
    complete: () => {
      expect(result).toStrictEqual(ALPHABET.split(''));
      done();
    },
  });
});

Feel free to update this blog post on GitHub, thanks in advance!

Support me

I appreciate it if you would support me if have you enjoyed this post and found it useful, thank you in advance.

Share this post

Twitter LinkedIn