Testing asynchronous RxJava code - Android

As @kiskae suggested I had to replace the schedulers. I replaced both subscribeOn and observeOn schedulers. The idea is to perform the operation on the same thread thereby making it synchronous. As this unit test runs on JVM, JVM won't have access to Android specific AndroidSchedulers.mainThread() which is passed as a scheduler to observeOn. So we replace this scheduler with the help of RxAndroidPlugins class. We do the same to replace the scheduler passed to subscribeOn using the RxJavaPlugins class.

For more info read this medium post.

Below is my working code.

import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.MockitoAnnotations;
import org.mockito.runners.MockitoJUnitRunner;

import java.util.concurrent.Callable;

import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.android.plugins.RxAndroidPlugins;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Function;
import io.reactivex.observers.TestObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;

import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@RunWith(MockitoJUnitRunner.class) 
public class SearchUserViewModelTest {

    private RepositoryImpl repository;
    private SearchUserViewModel viewModel;

@BeforeClass
public static void before(){
    RxAndroidPlugins.reset();
    RxJavaPlugins.reset();
    RxJavaPlugins.setIoSchedulerHandler(new Function<Scheduler, Scheduler>() {
        @Override
        public Scheduler apply(@NonNull Scheduler scheduler) throws Exception {
            return Schedulers.trampoline();
        }
    });
    RxAndroidPlugins.setInitMainThreadSchedulerHandler(new Function<Callable<Scheduler>, Scheduler>() {
        @Override
        public Scheduler apply(@NonNull Callable<Scheduler> schedulerCallable) throws Exception {
            return Schedulers.trampoline();
        }
    });
}


@Before
public void setup(){

    MockitoAnnotations.initMocks(this);
    repository = mock(RepositoryImpl.class);
    viewModel = new SearchUserViewModel(repository);
}

@Test
public void getUserInfo_returns_true(){
    UserInfo userInfo = new UserInfo();
    userInfo.setName("");
    when(repository.getUserInfo(anyString())).thenReturn(Observable.just(userInfo));
    TestObserver<Boolean> testObserver = new TestObserver<>();
    viewModel.getUserInfo(anyString()).subscribe(testObserver);
    testObserver.assertValue(true);
}

@AfterClass
public static void after(){
    RxAndroidPlugins.reset();
    RxJavaPlugins.reset();
}

}

Your method has to go through 2 different threads to produce a result (due to the subscribeOn and observeOn calls). This means the observer needs time to actually produce a result. Use TestObserver.awaitTerminalEvent() before checking assertValue to ensure the observable actually produced a value.

Alternatively you should use different schedulers when testing code, since the Android scheduler might require additional code to function properly in a testing environment.