RxJS Marble tests in Angular

Description

There is an excellent feature called Marble tests from RxJS to test asynchronous code synchronously. We could easily use it in Angular unit tests. I have made a few basic examples about testing code by Marble tests in Angular.

Setup project

My project is using Angular, so I have created a new project (using Angular CLI) by typing in console:

ng new marble-tests

My demo project is quite simple, so I answered no on
routing and selected SCSS as my stylesheet format.

Component

Service

I have made a simple dummy service for getting data. All of its methods return observables using of operator, which returns the stream from given arguments. The complete code is below.

import { Injectable } from '@angular/core';
import { Observable, of } from 'rxjs';

@Injectable({
  providedIn: 'root',
})
export class DataService {
  constructor() {}

  getList$(): Observable<string[]> {
    return of(['value1', 'value2', 'value3']);
  }

  getNumbers1$(): Observable<number[]> {
    return of([1, 2, 3]);
  }

  getNumbers2$(): Observable<number[]> {
    return of([4, 5, 6]);
  }

  getNumbers3$(): Observable<number[]> {
    return of([7, 8, 9]);
  }

  getBooleans$(): Observable<boolean> {
    return of(false, false, true, false);
  }
}

TS part

My component has simple form, made by using FormBuilder class.

readonly form = this.formBuilder.group({
    name: [],
  });

In ngOnInit method I am listening to value changes made on form values.

this.form.valueChanges
      .pipe(takeUntil(this.destroy$))
      .subscribe((data) => console.log(data));

To avoid memory leak, I am using the takeUntil operator, which completes the source stream when the given stream completes. In my situation, I am using Subject observable and assigning it to the destroy$ variable. To close it, I am calling complete method inside ngOnDestroy life-cycle hook (Remember to add OnDestroy class to implements on AppComponent). Variable:

readonly destroy$ = new Subject<void>();

and complete:

ngOnDestroy(): void {
    this.destroy$.complete();
  }

To render list to values I am using method getList which returns observable from my DataService. When any error occurs on that observable, I am catching it by catchError operator which expects observable to be returned, so I am returning empty array when error occurs.

getList(): Observable<string[]> {
    return this.dataService.getList$().pipe(catchError(() => of([])));
  }

My component has method which is setting flag variable to true when given stream emits true. To complete stream when true is emmited, I am using takeWhile operator which keeps stream active when given functions returns true.

setFlagOnTrue(stream$: Observable<boolean>): void {
    stream$.pipe(takeWhile((value) => !value)).subscribe({
      complete: () => (this.flag = true),
    });
  }

The following component method accepts any number of observables that return an array of numbers. I am using combineLatest operator, which emits when all of the given streams emit at least one time. Then I am flattening those arrays to a single one by flat method.

combineStreams$(...streams: Observable<number[]>[]): Observable<number[]> {
    return combineLatest(streams).pipe(map((lists) => lists.flat()));
  }

To display example numbers array I am getting numbers method from DataService and passing them to combineStreams$ method.

getNumbers$(): Observable<number[]> {
    return this.combineStreams$(
      this.dataService.getNumbers1$(),
      this.dataService.getNumbers2$(),
      this.dataService.getNumbers3$()
    );
  }

You can find the complete component code here.

HTML

HTML part is simple. It is only about usage of async pipe to convert async stream to pure values and json pipe for displaying arrays. Full html code below.

<form [formGroup]="form">

  <input type="text" formControlName="name">

</form>

<pre>{{ getList() | async | json }}</pre>

<pre>{{ getNumbers$() | async | json }}</pre>

<pre>FLAG: {{ flag }}</pre>

Marble tests

At the beginning of my AppComponent unit tests, I am declaring variables:

let component: AppComponent;
  let dataService: DataService;
  let testScheduler: TestScheduler;

TestScheduler is a class which allows us to virtualize time. Instance of that scheduler is created before each test. It provieds actual and expected assertions and expects boolean value on return.

testScheduler = new TestScheduler((actual, expected) =>
      expect(actual).toEqual(expected)
    );

TestScheduler has method run which as paramters has object of helpers used to define marble tests. My first test is checking if destroy$ variable is completed when component called ngOnDestroy.

it('should complete destroy', () => {
    testScheduler.run((helpers) => {
      const { expectObservable } = helpers;
      const expected = '|';
      component.ngOnDestroy();
      expectObservable(component.destroy$).toBe(expected);
    });
  });

expectObservable is method, which gets observable as a parameter and performs assertion on it

| indicates that the method should set observable as completed.

Next test checks if streams is unsubscribed when emitted value is true.

it('should unsubscribe when flag is true', () => {
    testScheduler.run((helpers) => {
      const { expectSubscriptions, cold } = helpers;
      const stream = cold('aaaba', { a: false, b: true });
      component.setFlagOnTrue(stream);
      const expect = '^--!';
      expectSubscriptions(stream.subscriptions).toBe(expect);
    });
  });

cold is method which creates cold observable. The first parameter (aaaba) is marble syntax, an extraordinary string of combinations of how observable behavior should be. It can be:

is ignored and used only for vertically marbles align
- represents the frame of virtual time passing
[0-9]+[ms|s|m] to specify exact amount of passed time
| indicates that the method should set observable as completed
# indicates that observable finished with error [a-z0-9] is any alphanumeric character that tells which value (from the second parameter) should use.
second parameter can be an object of values, which assertion can use keys in the first parameter

^--! is a subscription marble syntax, which is an extraordinary string of combinations of how a subscription should behave. It can be:

- represents the frame of virtual time passing
[0-9]+[ms|s|m] to specify exact amount of passed time
^ indicates that subscription happens
! indicates that unsubscription happens
() is for grouping events in the same frame

expectSubscriptions is method, which gets subscription log as a paramter and performs assertion on it. To summarize above emits false, false, false, true, false (aaaba, keys from values, so a = false, b = true). Then component.setFlagOnTrue is called on that stream. The expected behavior is '^--!', so it means that the method subscribed to it at the beginning (^), two virtual frames were passed (--), and at the end, it was unsubscribed (!).

Next test checks if values before subscription are taken to result.

it('should ignore values before subscription', () => {
    testScheduler.run((helpers) => {
      const { cold, hot, expectObservable } = helpers;
      const list1 = hot('a^b', { a: [1], b: [2] });
      const list2 = cold('a', { a: [3] });
      const list3 = cold('a', { a: [4] });
      const expected = '-a';
      expectObservable(component.combineStreams$(list1, list2, list3)).toBe(
        expected,
        {
          a: [2, 3, 4],
        }
      );
    });
  });

This time, one of the observables is hot, so additionally, We can use ^ indicator, which shows the moment when a subscription happens. In given tests, value [1] is ignored because it was emitted before subscription.

Last test check if returned list is an empty array, when error occurs.

it('should return empty list on error', () => {
    testScheduler.run((helpers) => {
      const { cold, expectObservable } = helpers;
      const list = cold('#', { a: ['value1', 'value2', 'value3'] });
      dataService.getList$ = () => list;
      const expected = '(a|)';
      expectObservable(component.getList()).toBe(expected, { a: [] });
    });
  });

In this test, dataService.getList$ is changed to the method which returns observable with error (# indicator, values are set just for proper typing). Assertion expects an empty array, and the stream is completed in a single frame ((a|), a as a key of value, | indicates that stream is completed).

Summary

Marble tests are a nice feature when we are thing about testing RxJS streams synchronously. It is worth trying.