Unit testing clients of Observables

Question

I have the following method go() I'd like to test:

private Pair<String, String> mPair;

public void go() {
    Observable.zip(
            mApi.webCall(),
            mApi.webCall2(),
            new Func2<String, String, Pair<String, String>>() {
                @Override
                public Pair<String, String> call(String s, String s2) {
                    return new Pair(s, s2);
                }
            }
    )
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Pair<String, String>>() {
                @Override
                public void call(Pair<String, String> pair) {
                    mApi.webCall3(pair.first, pair.second);
                }
            });
}

This method uses Observable.zip() to execute to http requests asynchronously, and merge them together in one Pair. In the end, another http request is executed with the result of these previous requests.

I'd like to verify that calling the go() method makes the webCall() and webCall2() requests, followed by the webCall3(String, String) request. Therefore, I'd like the following test to pass (using Mockito to spy the Api):

@Test
public void testGo() {
    /* Given */
    Api api = spy(new Api() {
        @Override
        public Observable<String> webCall() {
            return Observable.just("First");
        }

        @Override
        public Observable<String> webCall2() {
            return Observable.just("second");
        }

        @Override
        public void webCall3() {
        }
    });

    Test test = new Test(api);

    /* When */
    test.go();

    /* Then */
    verify(api).webCall();
    verify(api).webCall2();
    verify(api).webCall3("First", "second");
}

However when running this, web calls are executed asynchronously, and my test executes the assertion before the subscriber is done causing my test to fail.

I have read that you can use RxJavaSchedulersHook and RxAndroidSchedulersHook to return Schedulers.immediate() for all methods, but this results in the test running indefinitely.

I am running my unit tests on a local JVM.

How can I achieve this, preferably without having to modify the signature of go()?

Answer 1

I had a similar issue that took one more step in order to be solved.:

existingObservable
     .zipWith(Observable.interval(100, TimeUnit.MILLISECONDS), new Func1<> ...)
     .subscribeOn(schedulersProvider.computation())

Was still not using the provided TestScheduler schedulersProvider returned. It was necessary to specify .subscribeOn() on the individual streams that i was zipping in order to work.:

existingObservable.subscribeOn(schedulersProvider.computation())
     .zipWith(Observable.interval(100, TimeUnit.MILLISECONDS).subscribeOn(schedulersProvider.computation()), new Func1<> ...)
     .subscribeOn(schedulersProvider.computation())

Note that schedulersProvider is a mock returning the TestScheduler of my Test!

Answer 2

I have found out that I can retrieve my Schedulers in a non-static way, basically injecting them into my client class. The SchedulerProvider replaces the static calls to Schedulers.x():

public interface SchedulerProvider {

    Scheduler io();

    Scheduler mainThread();
}

The production implementation delegates back to Schedulers:

public class SchedulerProviderImpl implements SchedulerProvider {

    public static final SchedulerProvider INSTANCE = new SchedulerProviderImpl();

    @Override
    public Scheduler io() {
        return Schedulers.io();
    }

    @Override
    public Scheduler mainThread() {
        return AndroidSchedulers.mainThread();
    }
}

However, during tests I can create a TestSchedulerProvider:

public class TestSchedulerProvider implements SchedulerProvider {

    private final TestScheduler mIOScheduler = new TestScheduler();

    private final TestScheduler mMainThreadScheduler = new TestScheduler();

    @Override
    public TestScheduler io() {
        return mIOScheduler;
    }

    @Override
    public TestScheduler mainThread() {
        return mMainThreadScheduler;
    }
}

Now I can inject the SchedulerProvider in to the Test class containing the go() method:

class Test {

    /* ... */

    Test(Api api, SchedulerProvider schedulerProvider) {
        mApi = api;
        mSchedulerProvider = schedulerProvider;
    }

    void go() {
        Observable.zip(
            mApi.webCall(),
            mApi.webCall2(),
            new Func2<String, String, Pair<String, String>>() {
                @Override
                public Pair<String, String> call(String s, String s2) {
                    return new Pair(s, s2);
                }
            }
      )
            .subscribeOn(mSchedulerProvider.io())
            .observeOn(mSchedulerProvider.mainThread())
            .subscribe(new Action1<Pair<String, String>>() {
                @Override
                public void call(Pair<String, String> pair) {
                    mApi.webCall3(pair.first, pair.second);
                }
            });
    }
}

Testing this works as follows:

@Test
public void testGo() {
    /* Given */
    TestSchedulerProvider testSchedulerProvider = new TestSchedulerProvider();

    Api api = spy(new Api() {
        @Override
        public Observable<String> webCall() {
            return Observable.just("First");
        }

        @Override
        public Observable<String> webCall2() {
            return Observable.just("second");
        }

        @Override
        public void webCall3() {
        }
    });

    Test test = new Test(api, testSchedulerProvider);

    /* When */
    test.go();
    testSchedulerProvider.io().triggerActions();
    testSchedulerProvider.mainThread().triggerActions();

    /* Then */
    verify(api).webCall();
    verify(api).webCall2();
    verify(api).webCall3("First", "second");
}

Answer 3

I would use the TestScheduler and TestSubscriber in your tests. In order to use this you'll have to receive the observable that composes the zip so you can subscribe to that work with the testScheduler. Also you'll have to parameterize your schedulers. You won't have to change your go method's signature but you would have to parameterize the schedulers in underlying functionality. You could inject the schedulers by constructor, override a protected field by inheritance, or call to a package protected overload. I have written my examples assuming an overload that accepts the schedulers as arguments and returns the Observable.

The TestScheduler gives you a synchronous way to trigger async operator behavior in a predictable reproducible way. The TestSubscriber gives you a way to await termination and assert over values and signals received. Also you might want to be aware that the delay(long, TimeUnit) operator by default schedules work on the computation scheduler. You'll need to use the testScheduler there as well.

Scheduler ioScheduler = Schedulers.io();
Scheduler mainThreadScheduler =  AndroidSchedulers.mainThread();

public void go() {
    go(subscribeOnScheduler, mainThreadScheduler).toBlocking().single();
}

/*package*/ Observable<Pair<String, String>> go(Scheduler ioScheduler, Scheduler mainThreadScheduler) {
    return Observable.zip(
            mApi.webCall(),
            mApi.webCall2(),
            new Func2<String, String, Pair<String, String>>() {
                @Override
                public Pair<String, String> call(String s, String s2) {
                    return new Pair(s, s2);
                }
            })
            .doOnNext(new Action1<Pair<String, String>>() {
                @Override
                public void call(Pair<String, String>() {
                    mApi.webCall3(pair.first, pair.second);
                })
            })
            .subscribeOn(ioScheduler)
            .observeOn(mainThreadScheduler);
}

Test code

@Test
public void testGo() {
    /* Given */
    TestScheduler testScheduler = new TestScheduler();

    Api api = spy(new Api() {
        @Override
        public Observable<String> webCall() {
            return Observable.just("First").delay(1, TimeUnit.SECONDS, testScheduler);
        }

        @Override
        public Observable<String> webCall2() {
            return Observable.just("second");
        }

        @Override
        public void webCall3() {
        }
    });

    Test test = new Test(api);

    /* When */
    test.go(testScheduler, testScheduler).subscribe(subscriber);
    testScheduler.triggerActions();
    subscriber.awaitTerminalEvent();

    /* Then */
    verify(api).webCall();
    verify(api).webCall2();
    verify(api).webCall3("First", "second");
}

Answer 4

(Lambdas thanks to retrolambda)

For starters, I would rephrase go as:

private Pair<String, String> mPair;

public Observable<Pair<String, String>> go() {
    return Observable.zip(
                mApi.webCall(),
                mApi.webCall2(),
                (String s, String s2) -> new Pair(s, s2)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext(pair -> mPair = pair);
}

public Pair<String, String> getPair() {
    return mPair;
}

doOnNext allows you to intercept the value that is being processed in the chain whenever someone will subscribe to the Observable

Then, I would call the test like that:

Pair result = test.go().toBlocking().lastOrDefault(null);

Then you can test what result is.