Mock Retrofit Observable<T> response in Android Unit tests

Question

I have an API interface and I'm testing a View that involves network calls.

@Config(emulateSdk = 18)
public class SampleViewTest extends RobolectricTestBase {

    ServiceApi apiMock;

    @Inject
    SampleView fixture;

    @Override
    public void setUp() {
        super.setUp(); //injection is performed in super
        apiMock = mock(ServiceApi.class);
        fixture = new SampleView(activity);
        fixture.setApi(apiMock);
    }

    @Test
    public void testSampleViewCallback() {
        when(apiMock.requestA()).thenReturn(Observable.from(new ResponseA());
        when(apiMock.requestB()).thenReturn(Observable.from(new ResponseB());

        AtomicReference<Object> testResult = new AtomicReference<>();
        fixture.updateView(new Callback() {

            @Override
            public void onSuccess(Object result) {
                testResult.set(result);
            }

            @Override
            public void onError(Throwable error) {
                throw new RuntimeException(error);
            }
        });

        verify(apiMock, times(1)).requestA();
        verify(apiMock, times(1)).requestB();

        assertNotNull(testResult.get());
    }
}

For some reason apiMock methods are never called and verification always fails.

In my view I'm calling my api like this

apiV2.requestA()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer());

What am I missing here?

Update #1:
After some investigation it appears that when in my implementation (sample above) I observeOn(AndroidSchedulers.mainThread()) subscriber is not called. Still do not know why.

Update #2:
When subscribing just like that apiV2.requestA().subscribe(new Observer()); everything works just fine - mock api is called and test passes.
Advancing ShadowLooper.idleMainLooper(5000) did nothing. Even grabbed looper from handler in HandlerThreadScheduler and advanced it. Same result.

Update #3: Adding actual code where API is used.

public void updateView(final Callback) {
    Observable.zip(wrapObservable(api.requestA()), wrapObservable(api.requestB()),
        new Func2<ResponseA, ResponseB, Object>() {
            @Override
            public Object call(ResponseA responseA, ResponseB responseB) {
                return mergeBothResponses(responseA, responseB);
            }
        }
    ).subscribe(new EndlessObserver<Object>() {

        @Override
        public void onError(Throwable e) {
            Log.e(e);
            listener.onError(e);
        }

        @Override
        public void onNext(Object config) {
            Log.d("Configuration updated [%s]", config.toString());
            listener.onSuccess(config);
        }
    });
}

protected <T> Observable<T> wrapObservable(Observable<T> observable) {
    return observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

Answer 1

I'm still wrapping my head around how to properly use rxjava myself but I would try to modify your code so that you only observeOn(mainThread) on the final zipped Observable instead of doing it on both of the original request response's Observable. I would then verify if this affect the fact that you have to advance both Loopers or not.

To simply your tests and remove the need for Looper idling I would take the threading out of the equation since you don't need background processing when running tests. You can do that by having your Schedulers injected instead of creating them statically. When running your production code you'd have the AndroidSchedulers.mainThread and Schedulers.io injected and when running tests code you would inject Schedulers.immediate where applicable.

@Inject
@UIScheduler /* Inject Schedulers.immediate for tests and AndroidSchedulers.mainThread for production code */
private Scheduler mainThreadSched;

@Inject
@IOScheduler /* Inject Scheduler.immediate for tests and Schedulers.io for production code */
private Scheduler ioSched;

public void updateView(final Callback) {
    Observable.zip(wrapObservable(api.requestA()), wrapObservable(api.requestB()),
        new Func2<ResponseA, ResponseB, Object>() {
            @Override
            public Object call(ResponseA responseA, ResponseB responseB) {
                return mergeBothResponses(responseA, responseB);
            }
        }
    ).observeOn(mainThreadSched)
    .subscribe(new EndlessObserver<Object>() {

        @Override
        public void onError(Throwable e) {
            Log.e(e);
            listener.onError(e);
        }

        @Override
        public void onNext(Object config) {
            Log.d("Configuration updated [%s]", config.toString());
            listener.onSuccess(config);
        }
    });
}

protected <T> Observable<T> wrapObservable(Observable<T> observable) {
    return observable.subscribeOn(ioSched);
}

Answer 2

what version of rxjava are you using? I know there was some changes in the 0.18.* version regarding the ExecutorScheduler. I had a similar issue as you when using 0.18.3 where I wouldn't get the onComplete message because my subscription would be unsubscribe ahead of time. The only reason I'm mentioning this to you is that a fix in 0.19.0 fixed my issue.

Unfortunately I can't really explain the details of what was fixed, it's beyond my understanding at this point but if it turns out to be the same cause maybe someone with more understand could explain. Here's the link of what I'm talking about https://github.com/Netflix/RxJava/issues/1219.

This isn't much of an answer but more a heads up in case it could help you.

Answer 3

As @champ016 stated there were issues with RxJava versions that are lower than 0.19.0.

When using 0.19.0 the following approach works. Although still don't quite get why I have to advance BOTH loopers.

@Test
public void testSampleViewCallback() {
    when(apiMock.requestA()).thenReturn(Observable.from(new ResponseA());
    when(apiMock.requestB()).thenReturn(Observable.from(new ResponseB());

    AtomicReference<Object> testResult = new AtomicReference<>();
    fixture.updateView(new Callback() {

        @Override
        public void onSuccess(Object result) {
            testResult.set(result);
        }

        @Override
        public void onError(Throwable error) {
            throw new RuntimeException(error);
        }
    });

    ShadowLooper.idleMainLooper(5000);
    Robolectric.shadowOf(
        Reflection.field("handler")
            .ofType(Handler.class)
            .in(AndroidSchedulers.mainThread())
            .get().getLooper())
            .idle(5000);

    verify(apiMock, times(1)).requestA();
    verify(apiMock, times(1)).requestB();

    assertNotNull(testResult.get());
}