Error subscribing to epics in Jest test with message "You provided an invalid object where a stream was expected."

Question

I'm pretty new to redux-observable so I'm not sure whether is a real problem or just a silly mistake.

I'm trying to add test to a epics I wrote for a React applications, but I'm not able to test them. This is the example.

My example epic:

const example = action$ =>
     action$.pipe(
        ofType('my_type'),
        mergeMap(() => {
            return { type: 'result_type'}
        })
    );

I was trying to test it in the simplest possible way:

it('simple test', done => {
    const action$ = ActionsObservable.of({ type: 'my_type'});
    const state$ = null;

    return myEpics.example(action$, state$)
        .subscribe(actions => {
            expect(actions).toEqual({ type: 'result_type'});
            done();
        })

});

Using this code I get the following error:

TypeError: You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Iterable.

      12 | 
      13 |     return myEpics.example(action$, state$)
    > 14 |         .subscribe(actions => {
         |          ^

I'm using redux-observable 1.2.0, rxjs 6.5.3

Did I miss something? I saw many examples of this type and they work.

EDIT

What I'm really doing (apart from the dummy example) is replacing a redux-thunk promise-based api middleware (with many async call to a ReST endpoint). I was able to make it work, but about testing I'm a little bit confused. I saw the marble-diagram approach, and the subscribe-to-epic one adding expectations e.g. on api calls, resulting actions. I'm using the latter, but, for example, for an epic returning multiple actions, I get only the first one in the subscribe.

A 'complex' example:

export const handleLogIn = (action$, state$) =>
    action$.pipe(
        ofType(authTypes.LOG_IN),
        withLatestFrom(state$),
        mergeMap(([{ payload: credentials}, { router }]) =>
            from(ApiService.logIn(credentials))
                .pipe(
                    mergeMap(authToken => of(authActions.set_auth_token(authToken), navigationActions.goTo(router.location.state ? router.location.state.pathname : baseRoutes.START),
                    catchError(message => of(authActions.set_auth_error(message)))
                )
        )
    );

Test

it('should handle LOG_IN navigating to base path', done => {
    const token = 'aToken';
    const credentials = { username: 'user', password: 'password' };
    const action$ = ActionsObservable.of(authActions.log_in(credentials));
    const state$ = new StateObservable(new Subject(), { router: { location: { state: null }}});

    spyOn(ApiService, 'logIn').and.returnValue(Promise.resolve(token));

    authEpics.handleLogIn(action$, state$)
        .subscribe(actions => {
            expect(ApiService.logIn).toHaveBeenCalledWith(credentials);
            expect(actions).toEqual([
                authActions.set_auth_token(token),
                navigationActions.go_to(baseRoutes.START)
            ]);
            done();
        });

});

the test fails because only authActions.set_auth_token is the only action returned. Am I missing something?

jayphelps jayphelps · Accepted Answer · 2019-10-11T23:14:58

Welcome!

The issue is an RxJS error. mergeMap is an operator that is basically a map() plus a mergeAll(). That is, inside the callback you provide, you're expected to return something stream-like. Usually that's another Observable, but it can also be a Promise, Array, or Iterable instead.

mergeMap is one of the one-to-many operators, with the difference between it and concatMap, switchMap, and exhaustMap that each of them have difference strategies for handling what should happen if a new source value is emitted before the returned inner stream you projected to hasn't finished yet.

In the case of mergeMap, if another source value is emitted before the inner stream has finished it will call your callback again ("project" in functional programming terms) and subscribe to it while merging any resulting values with the previously projected inner stream.

In your case, you're returning a Plain Old JavaScript Object inside mergeMap, which is not stream-like (Observable, Promise, Array, or Iterable). If in fact you just want to map one action to another action 1:1, you could just use map:

import { map } from 'rxjs/operators';

const example = action$ =>
  action$.pipe(
    ofType('my_type'),
    map(() => {
      return { type: 'result_type' };
    })
);

But be warned that this isn't usually idiomatic Redux (some rare exceptions) the original action is probably what your reducers should be handling. But you might have just being doing that for a "hello world" style thing, which is totally cool.

That said, it's still possible to return one (or more) values synchronously from inside a mergeMap. You can use of()

import { of } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

const example = action$ =>
  action$.pipe(
    ofType('my_type'),
    map(() => {
      return of({ type: 'result_type' });
      // or more than one
      // return of({ type: 'first'}, { type: 'second' }, ...etc);
    })
);

Remember: redux-observable is a small helper library for doing RxJS + Redux. That means that really you'll be learning RxJS and Redux, as there's very little to learn about redux-observable itself outside of an "Epic" being a pattern of structuring your RxJS code and the ofType() operator that is sugar on-top of filter()

If you're confused--you're not alone--it might be best to play around in a Stackblitz and/or take a peek at some RxJS tutorials that touch on this more. While it can be hard to understand, once you do, the abilities RxJS unlocks become much more apparent for complex async code.

Error subscribing to epics in Jest test with message "You provided an invalid object where a stream was expected."

1 Answers