Observable pipelines and operators benefit greatly from TestScheduler. The tool enables declarative, straightforward, fast, and accurate tests. This post examines three considerations for making the most of RxJS marble testing:

  • Is it acceptable to work with mutable data structures?
  • What is the approach for testing streams built on Promises?
  • How do we go about testing meta-observables and inner observables?

This content assumes you already understand the foundations of RxJS and the TestScheduler API. If you need a primer, consult the guide How to test Observables.

You can explore the accompanying source code on Stackblitz.

We begin with the first topic.

Mutable Data

A strong convention in RxJS pipelines is to rely on immutable data and pure functions. This approach aligns with the library’s functional programming philosophy and prevents aliasing bugs. The framework, however, does not mandate this; mutable objects remain an option.

Consider this example:

const toggle = 
  () => pipe(
    scan((acc, value) => {
        const found = acc.indexOf(value) > -1;
        if (found) {
          acc.splice(acc.indexOf(value), 1);    
        } else {
          acc.push(value);
        }

        return acc;
    }, []) 
  );

mutable toggle operator

The toggle operator examines the accumulator array for a given number. If present, the number is removed; if absent, it is appended. Testing the operator using the snippet below:

of(1, 2, 3, 2, 3, 1).pipe(
  toggle()
)
.subscribe(x => console.log(x));

yields this output:

[1]
[1, 2]
[1, 2, 3]
[1, 3]
[1]
[]

The logic works as intended, and the printed result matches expectations.

Yet, the following marble test will not pass:

testScheduler.run(({ expectObservable }) => {
    const source =  of(1, 2, 3, 2, 3, 1);

    const result = source.pipe(toggle());

    expectObservable(result).toBe(
    	'(abcdef|)',
        {a:[1], b:[1,2], c:[1,2,3], d:[1,3], e:[1], f:[]}
    );
});

The failure stems from how TestScheduler behaves. It does not record a snapshot of the object’s state; instead, it holds a reference to the object it receives, as seen in the source code. Consequently, any mutation after the observable processes the object can lead to test outcomes that are misleading, even if they appear to succeed.

Refactoring the scan predicate into a pure function resolves the problem:

const toggle = 
  () => pipe(
    scan((acc, value) => {
        const found = acc.indexOf(value) > -1;
        if (found) {
          return acc.filter(v => v !== value);    
        } else {
          return [...acc, value];
        }
    }, []) 
  );

With that issue resolved, let’s move on to a more fundamental constraint.

Promises

RxJS integrates smoothly with Promises. Converting an observable to a Promise—or the reverse—is straightforward. However, leaning on this interoperability means giving up the precision that marble diagrams offer.

To illustrate, imagine a third-party client called RedisCache that handles remote storage via a Promise-based interface:

interface RedisCache {
  get(...keys: string[]): Promise<Record<string, string>>;
}

Our goal is to interact exclusively with Observables in our application code, so we avoid calling this API directly.

class RemoteCache {

  constructor(private redis: RedisCache) {    
  }

  get(key: string): Observable<{key: string; value: string;}> {
    return from(this.redis.get(key))
      .pipe(
        map(value => ({key: key, value: value[key]}))
      );
  }
}

For testing the remote cache abstraction, one might create a stub like this:

class RedisCacheStub implements RedisCache {
  get(...keys: string[]): Promise<Record<string, string>> {
    return Promise.resolve({key: 'value'});
  }
}

The downside is that this approach prevents us from employing marble syntax in our tests. For a deeper discussion, see this issue.

testScheduler.run(({ expectObservable }) => {
    const cache = new RemoteCache(new RedisCacheStub());

    const result = cache.get('key');

    expectObservable(result).toBe('(a|)', {a:[1]});
});

A workaround involves substituting Promises with observables.

class RedisCacheObservableStub implements RedisCache {
  get(...keys: string[]): Promise<Record<string, string>> {
    return of({key: 'value'}) as unknown as Promise<Record<string, string>>;
  }
}

TypeScript will not raise an issue when an observable is stated as a Promise. Additionally, at runtime, RxJS identifies the object as an Observable based on its shape, not its declared type.

The from factory method accepts various inputs. According to the official docs:

A subscription object, a Promise, an Observable-like, an Array, an iterable, or an array-like object to be converted.

This flexibility lets us use an Array and simply cast it to a Promise type.

class RedisCacheArrayStub implements RedisCache {
  get(...keys: string[]): Promise<Record<string, string>> {
    return [{key: 'value'}] as unknown as Promise<Record<string, string>>;
  }
}

Now, we address the final installment of this article.

Meta and Inner Observables

RxJS supports observables that emit other observables, known as meta-observables. A common operator in this category is groupBy. Marble diagrams make testing such nested values quite straightforward. The following example draws from the documentation.

testScheduler.run(({ expectObservable, hot, cold }) => {
    const sourceValues = {
      a: {type: 'language', name: 'JavaScript'},
      b: {type: 'bundler', name: 'Parcel'},
      c: {type: 'bundler', name: 'webpack'},
      d: {type: 'language', name: 'TypeScript'},
      e: {type: 'linter', name: 'TSLint'}
    };
    const source = hot('--a---b---c---d---e---|', sourceValues);
    
    const result = source.pipe(groupBy(({type}) => type));

    const x = cold('      a-----------d-------|', languages()),
          y = cold('          b---c-----------|', bundlers()),
          z = cold('                      e---|', linters());
    expectObservable(result)
                 .toBe('--x---y-----------z---|', { x, y, z });
});

function languages() { 
  return {
    a: {type: 'language', name: 'JavaScript'},
    d: {type: 'language', name: 'TypeScript'}
  };
}

function bundlers() { 
  return {
    b: {type: 'bundler', name: 'Parcel'},
    c: {type: 'bundler', name: 'webpack'}
  };
}

function linters() { 
  return {
    e: {type: 'linter', name: 'TSLint'}
  };
}

Beyond the main output, x, y, and z are also expressed as observables using marble syntax.

Inner observables pose a different challenge. Testing them requires moving beyond the elegant marble notation.

testScheduler.run(({ expectObservable: expect, hot, cold }) => {
    const values = {
      a: {
         'languages': cold('--a--b', {a: 'JavaScript', b: 'TypeScript'}),
         'bundlers': cold('--a--b', {a: 'Parcel', b: 'Webpack'}),
         'linters': cold('--a', {a: 'TSLint'})
      }
    };
    
    const source = hot('--a--', values);

    source.subscribe(emited => {
      expect(emited.languages).toBe('----a--b', {a: 'JavaScript', b: 'TypeScript'});
      expect(emited.bundlers).toBe('----a--b', {a: 'Parcel', b: 'Webpack'});
      expect(emited.linters).toBe('----a---', {a: 'TSLint'});
    });
});

The approach above is valid but not ideal. We need to confirm that the source observable emits exactly one value.

Closing Thoughts

Marble-based testing is an essential tool for anyone working with RxJS. Understanding its capabilities, constraints, and nuances allows us to wield it with greater precision and impact.