flush() Function

The RunHelpers object includes a property flush, which is a function that executes the currently queued assertions for the TestScheduler.

It is important to note that the assertions returned by either expectObservable or expectSubscriptions are not executed until the callback function provided to the run() method of the TestScheduler is returned. If we need to manually execute assertions in the queue, then we need to manually invoke the flush() method.

This is helpful for testing scenarios where an Observable performs side effects.

Example

import { interval } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';
import { map, tap } from 'rxjs/operators';

describe('getting started with RxJS testing with marbles', () => {
  let testScheduler: TestScheduler;

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

  test('produce side effect', () => {
    testScheduler.run(({ expectObservable, flush }) => {
      // this is probably not a good idea :shrug:
      let counter = 0;

      // produces a side effect
      const source = interval(1000).pipe(
        tap((i) => (counter = i)),
        map((i) => i.toString())
      );
      const subscription = '5001ms !';
      const expected = '1000ms 0 999ms 1 999ms 2 999ms 3 999ms 4';
      expectObservable(source, subscription).toBe(expected);

      // flush the scheduler
      flush();

      expect(counter).toBe(4);
    });
  });
});

See example on codesandbox

If an Observable is producing side effects, its probably not a good idea. However, let's learn how we can test for a potential side effect.

• In the test above I have a counter number with an initial value of 0.
• The source Observable emits a value every 1000 milliseconds using the interval() operator.
• The side effect is in the tap() operator that sets the counter value for each next notification. In order to test for this, we need to run the assertion produced by the expectObservable() before the callback function returns.
• Note that we are providing a subscription string to the expectObservable() method in order to unsubscribe from the source Observable after 5000 milliseconds.
• We use the flush() function to execute any queued tests.
• Finally, we can use Jest's expect() and toBe() function to assert that the counter value is incremented to 4.