(RxJava2)Single.blockingGet() don't work on unit test

Question

I creating unit test for method of interactor that return Rx Single. The Single must to return a bean of user asynchronously. I tried to call blockingGet() for Single but the bean don't return since the test works endlessly. I checked breakpoints and I saw that bean was created but it didn't return. Also I tried to call test().values().get(0) for Single but it didn't contain the bean. The breakpoints didn't call.

I can to understand what it is problem. I need help for resolve it.

The method in the interactor.

    public Single<UserBean> verifyUser() {
    String jwt = getJWTToken();
    if (TextUtils.isEmpty(jwt)) {
        return Single.error(new EmptyJWTException());
    }
    if (!checkConnectionInfo()) {
        return Single.error(new NotNetworkConnectedException());
    }
    return Single.create((SingleOnSubscribe<UserBean>) emitter -> {
        UserBean resultBean = mAuthenticationRepository.verifyUserByJwt(jwt);
        if (resultBean != null && !TextUtils.isEmpty(resultBean.getId())) {
            emitter.onSuccess(resultBean);
        } else {
            emitter.onError(new RequestToServerException());
        }
    })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
}

The unit test.

@RunWith(MockitoJUnitRunner.class)
public class AuthenticationInteractorTest {

@Mock
private Application mApplication;
@Mock
private IAuthenticationRepository mAuthenticationRepository;
@Mock
private SharedPreferences mSharedPreferences;
@Mock
private ConnectivityManager mConnectivityManager;
@Mock
private NetworkInfo mNetworkInfo;
private AuthenticationInteractor mInteractor;

@Before
public void setUp() {
    mInteractor = new AuthenticationInteractor(mAuthenticationRepository, mApplication);
}

@Test
public void verifyUserTest() {
    String jwt = "jwt";
    UserBean userBean = createUserBean();
    when(mApplication.getSharedPreferences(AuthenticationInteractor.SHARED_PREFS, Context.MODE_PRIVATE)).thenReturn(mSharedPreferences);
    when(mSharedPreferences.getString(AuthenticationInteractor.TOKEN_KEY, null)).thenReturn(jwt);
    when((ConnectivityManager) mApplication.getSystemService(Context.CONNECTIVITY_SERVICE)).thenReturn(mConnectivityManager);
    when(mConnectivityManager.getActiveNetworkInfo()).thenReturn(mNetworkInfo);
    when(mNetworkInfo.isConnected()).thenReturn(true);
    when(mAuthenticationRepository.verifyUserByJwt(jwt)).thenReturn(userBean);
    UserBean bean = mInteractor.verifyUser().blockingGet();
    assertEquals(userBean, bean);
}

private UserBean createUserBean() {
    UserBean userBean = new UserBean();
    userBean.setMetaId("1539762652616:940c29e6d716:15:jmt4z02l:11182");
    userBean.setJWT("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6Im1hZ2F6b240aWtAeWFuZGV4LnJ1IiwiY3MiOiI1OWEwNDA2MTgzNDAwMDAwIiwiaWF0IjoxNTM5NzYyNjUyLCJhdWQiOiIqLmxvY2FsaG9zdCIsImlzcyI6Im1zLXVzZXJzIn0.2AcuLZjR_-GyWAFen7JczJ5-W37zQWGbnRRErGLqDEY");
    userBean.setId("test@test.test");
    userBean.setType("user");
    userBean.setUserName("test@test.test");
    userBean.setCreated(1535970916310L);
    userBean.setLatestResult("1539676474685");
    userBean.setOnDiet("No");
    return userBean;
}

}

Answer 1

It looks like you're not overriding the Schedulers being used. Without doing this you're almost certainly going to get the following error when trying to subscribe using AndroidSchedulers.mainThread

Method getMainLooper in android.os.Looper not mocked

To overcome this you could create a TestRule which allows you to specify the Schedulers for your test, e.g. the following would reset Schedulers.io and AndroidSchedulers.mainThread with Schedulers.trampoline.

public class RxSchedulersRule implements TestRule {
    @Override
    public Statement apply(final Statement base, Description description) {
        return new Statement() {
            @Override
            public void evaluate() throws Throwable {
                RxAndroidPlugins.reset();
                RxJavaPlugins.reset();

                RxAndroidPlugins.setInitMainThreadSchedulerHandler(s -> Schedulers.trampoline());
                RxJavaPlugins.setIoSchedulerHandler(s -> Schedulers.trampoline());

                try {
                    base.evaluate();
                } finally {
                    RxAndroidPlugins.reset();
                    RxJavaPlugins.reset();
                }
            }
        };
    }
}

To use in your test you would then need to declare the rule

@Rule public final RxSchedulersRule schedulersRule = new RxSchedulersRule();