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 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");
}

Unit testing clients of Observables

Sign up