observables_marble_testing.js

import { TestScheduler } from 'rxjs/testing';
import { map, concatWith, take, delay } from 'rxjs/operators';
import { from, of, catchError, interval } from 'rxjs';

describe('Marble testing', () => {
  let testScheduler;

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

  it('should convert ASCII diagrams into observables', () => {
    testScheduler.run(helpers => {
      const { cold, expectObservable } = helpers;
      const source$ = cold('--a-b---c');
      const expected =     '--a-b---c';

      expectObservable(source$).toBe(expected);
    })
  })

  it('should callow configuration of emitted values', () => {
    testScheduler.run(helpers => {
      const { cold, expectObservable } = helpers;
      const source$ = cold('--a-b---c', {a: 1, b: 2, c: 3});
      const final$ = source$.pipe(map(value => value * 10))
      const expected =     '--a-b---c';

      expectObservable(final$).toBe(expected, {a: 10, b: 20, c:30});
    })
  })

  it('should let you identify subscription points', () => {
    testScheduler.run(helpers => {
      const { cold, expectObservable, expectSubscriptions } = helpers;
      const source1$ = cold('-a---b-|');
      const source2$ = cold('-c---d-|');
      const final$ = source1$.pipe(concatWith(source2$));
      // ^ is a symbol for subscription
      // it is helpful to write streams under each other to know where exaclty subscription happens
      const expected =             '-a---b--c---d-|';
      const sourceOneExpectedSub = '^------!';
      const sourceTwoExpectedSub = '-------^------!';

      expectObservable(final$).toBe(expected);
      expectSubscriptions(source1$.subscriptions).toBe(sourceOneExpectedSub)
      expectSubscriptions(source2$.subscriptions).toBe(sourceTwoExpectedSub)
    })
  })

  it('should let you test hot observables', () => {
    testScheduler.run(helpers => {
    // hot: useful to confirm that pre-subscription emissions don't affect the outcome of observable
    // you can set subscription point with a `^` character (if you won;t add it, it acts the same as cold)
    const { hot, expectObservable } = helpers;
    const source$ = hot('--a-b-^-c');
    const final$ = source$.pipe(take(1));
    const expected =          '--(c|)';
    expectObservable(final$).toBe(expected);
    })
  })

  it('should let you test sync operations', () => {
    testScheduler.run(helpers => {
      const { expectObservable } = helpers;
      const source$ = from([1,2,3,4,5])
      // all values are expected in the same frame - we can wrap it with () 
      // we need to include also the completion with |
      const expected = '(abcde|)'
      expectObservable(source$).toBe(expected, {a:1, b:2, c:3, d:4, e:5})});
  })

  it('should let you test async operations', () => {
    testScheduler.run(helper => {
      const {expectObservable} = helper
      const source$ = from([1,2,3,4,5])
      const final$ = source$.pipe(delay(200))
      // 200ms/1s/... can be used to test delays
      const expected = '200ms (abcde|)';
      expectObservable(final$).toBe(expected, {a:1, b:2, c:3, d:4, e:5})});
  })

    it('should let you test errors and error messages', () => {
    testScheduler.run(helper => {
      const {expectObservable} = helper
      const source$ =of({firstName: 'John', lastName: 'Doe'}, null).pipe(
        map(o => `${o.firstName} ${o.lastName}`),
        catchError(() => {
          throw { message: 'Invalid user!'} 
        })
      );

      // we can use # to indicate the error message
      const expected = '(a#)';
      expectObservable(source$).toBe(expected, {a: 'John Doe'}, {message: 'Invalid user!'})});
  })

  it('should let you test snapshots of streams that do not complete', () => {
    testScheduler.run(helper => {
      const {expectObservable} = helper
      const source$ = interval(1000).pipe(
        map(value => `${value + 1} sec`)
      )
      // 999ms + 1ms for "a"
      const expected = '1s a 999ms b 999ms c';
      // When to unsubscribe
      const unsubscribe = '4s !'
      expectObservable(source$,unsubscribe).toBe(expected, {a: '1 sec', b: '2 sec', c: '3 sec'});
    })
  })
})
  

observables_marble_testing_custom_custom_pipe

import { of, delay, throwError } from 'rxjs';
import { TestScheduler } from 'rxjs/testing'
import { breweryTypeahead } from './index_breweries'

// Observable that we are testing: 
const BASE_URL = 'https://api.openbrewerydb.org/breweries';
const inputBox = document.querySelector('input-box');
const showBox = document.querySelector('show-box');
const input$ = fromEvent(inputBox, 'input')


export const breweryTypeahead = (ajaxHelper = ajax) => sourceObservable => {
  return sourceObservable.pipe(
      debounceTime(200),
      pluck('target', 'value'),
      distinctUntilChanged(),
      switchMap(searchTerm => 
      ajaxHelper.getJSON(`${BASE_URL}?by_name=${searchTerm}`).pipe(catchError(() => EMPTY))
    )
  )
}

input$.pipe(
 breweryTypeahead()
).subscribe(response => {  
  showBox.innerHtml = response.data
})

// Tests:
describe('breweries', () => {
  let testScheduler;

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

  it('should debounce input by 200ms', () => {
    testScheduler.run(helpers => {
      const {cold, expectObservable } = helpers;
      const searchTerm = 'testing'
      const source$ = cold('a', { a: { target: { value: searchTerm }}});
      const final$ = source$.pipe(
        breweryTypeahead({
          getJSON: () => of(searchTerm).pipe(delay(300))
        })
      );
      const expected = '500ms a'; 
      expectObservable(final$).toBe(expected, { a: searchTerm });
    })
  })

  it('should cancel active request if no value emitted', () => {
    testScheduler.run(helpers => {
      const {cold, expectObservable } = helpers;
      const searchTerm = 'testing'
      const source$ = cold('a 250ms b', { a: { target: { value: 'first' }}, b: { target: { value: 'second' }}});
      const final$ = source$.pipe(
        breweryTypeahead({
          getJSON: () => of(searchTerm).pipe(delay(300))
        })
      );
      const expected = '751ms b'; 
      expectObservable(final$).toBe(expected, { b: searchTerm });
    })
  })

  it('should not emit duplicate values in a row', () => {
    testScheduler.run(helpers => {
      const {cold, expectObservable } = helpers;
      const searchTerm = 'testing'
      const source$ = cold('a 250ms b', { a: { target: { value: 'first' }}, b: { target: { value: 'first' }}});
      const final$ = source$.pipe(
        breweryTypeahead({
          getJSON: () => of(searchTerm).pipe(delay(300))
        })
      );
      const expected = '500ms a'; 
      expectObservable(final$).toBe(expected, { a: searchTerm });
    })
  })
  
  it('should ignore ajax errors', () => {
    testScheduler.run(helpers => {
      const {cold, expectObservable } = helpers;
      const source$ = cold('a 250ms b', { a: { target: { value: 'first' }}, b: { target: { value: 'first' }}});
      const final$ = source$.pipe(
        breweryTypeahead({
          getJSON: () => throwError(() =>'error')
        })
      );
      const expected = ''; 
      expectObservable(final$).toBe(expected);
    })
  })
})