expectObservable().toBe() only receives last value of stream

Question

I want to test a service that uses internally a BehaviorSubject to hold the state and exposes Observables with a distinctUntilChanged() in the pipe. When I run the following test, than the actual steam that is compared with my expectations only 'contains' the last value. What do I have to understand to fix that?

import { BehaviorSubject } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';

describe('My exposed stream', () => {
  let testScheduler;

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

  it('does not propagate if the current value equals the last one', () => {
    testScheduler.run(({ expectObservable }) => {
      const internalStream$ = new BehaviorSubject<string>(null);
      const exposedStream$ = internalStream$.pipe(distinctUntilChanged());

      expectObservable(exposedStream$).toBe('012', [null, 'foo', 'bar']);

      internalStream$.next('foo');
      internalStream$.next('foo');
      internalStream$.next('bar');
    });
  });
});

Result:

Expected $.length = 1 to equal 3.
Expected $[0].notification.value = 'bar' to equal null.
Expected $[1] = undefined to equal Object({ frame: 1, notification: Notification({ kind: 'N', value: 'foo', error: undefined, hasValue: true }) }).
Expected $[2] = undefined to equal Object({ frame: 2, notification: Notification({ kind: 'N', value: 'bar', error: undefined, hasValue: true }) }).

You can run and modify the code here: https://stackblitz.com/edit/typescript-moydq7?file=test.ts

Answer 1

My code had 2 issues:

1. I needed to catch the events with a ReplySubject (Thank you, @AliF50)
2. The correct marble notation to verify the behavior is (012) instead if 012, because the 3 events are sent within the same time frame synchronously.

Here is my solution. Still very cumbersome and if you are aware of better ways, please let me know.

import { BehaviorSubject, ReplaySubject } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';

describe('My exposed stream', () => {
  let testScheduler;

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

  it('does not propagate if the current value equals the last one', () => {
    testScheduler.run(({ expectObservable }) => {

      // given
      const internalStream$ = new BehaviorSubject<string>(null);
      const exposedStream$ = internalStream$.pipe(distinctUntilChanged());

      const observedValues$ = new ReplaySubject<string>();
      exposedStream$.subscribe(observedValues$);

      // when
      internalStream$.next('foo');
      internalStream$.next('foo');
      internalStream$.next('bar');

      // then
      expectObservable(observedValues$).toBe('(012)', [null, 'foo', 'bar']);
    });
  });
});

You can play around with it here: https://stackblitz.com/edit/typescript-sa1n8v?file=test.ts

Answer 2

https://stackoverflow.com/a/62773431/7365461, I think that post can answer your question.

That being said, I would test it like so (without the testScheduler):

it('does not propagate if the current value equals the last one', () => {
  const internalStream$ = new BehaviorSubject<string>(null);
  const exposedStream$ = internalStream$.pipe(distinctUntilChanged());
  
  let subscribeCount = 0;
  const subscribeValues = [];
  exposedStream$.subscribe(value => {
    if (subscribeCount > 3) {
      fail();
    }

    subscribeValues.push(value);
  });

  internalStream$.next('foo');
  internalStream$.next('foo');
  internalStream$.next('bar');
  
  expect(subscribeValues).toEqual([null, 'foo', 'bar']);
});

Edit

It seems like our syntax for expectObservable is a bit wrong, I don't have experience with expectObservable though. Check this out, it works in the link you have provided me:

import { BehaviorSubject, ReplaySubject } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';

describe('My exposed stream', () => {
  let testScheduler;

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

  it('does not propagate if the current value equals the last one', () => {
    testScheduler.run(({ expectObservable }) => {
      const internalStream$ = new BehaviorSubject<string>(null);
      const exposedStream$ = internalStream$.pipe(distinctUntilChanged());

      const observedValues$ = new ReplaySubject<string>();
      exposedStream$.subscribe(observedValues$);
      expectObservable(observedValues$).toBe('(abc)', {'a': null, 'b': 'foo', 'c': 'bar' });

      internalStream$.next('foo');
      internalStream$.next('foo');
      internalStream$.next('bar');
    });
  });
});

As for as the assertions not traversing inside of the subscribe, I understand and you're right. I have two defense strategies for this. One is using the done callback of jasmine and the other is using failSpecWithNoExpectations of jasmine. Check out this link https://testing-angular.com/angular-testing-principles/ towards the end on how to do it. This will fail a test if it has no expectations or no expectations were travelled/traversed.

For the done callback, you can do:

it('does stuff', done => {
  myObservable$.subscribe(value => {
    expect(value).toBe('abc');
    // call done to tell jasmine you're done with this test
    // if this done is not called, the test will hang and fail
    done();
  });
});