Testing asynchronous RxJava code - Android

Question

I am in the process of getting to know RxJava. I started using it in one of my personal apps and would like to unit test the code but having some difficulties and would like some help.

The scenario is simple.

1. I get UserInfo object by making a REST call
2. If the retuned UserInfo object is not null return true else return false.

For the above scenario my RxJava code looks like this

public LiveData<Boolean> doesUserExists(String userName) {
    UserExistsObserver observer= new UserExistsObserver ();
    getUserInfo(userName).subscribeWith(subscriber);
    disposable.add(observer);
    return userExists;
}

public Observable<Boolean> getUserInfo(String userName) {
    return repository.getUserInfo(userName)
            .flatMap(new Function<UserInfo, Observable<Boolean>>() {
                @Override
                public Observable<Boolean> apply(UserInfo userInfo) throws Exception {
                      return Observable.just(userInfo != null);
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
}

So I wanted to write a simple unit test to check if getUserInfo() returns correct boolean value or not. Below is my unit test.

 @Test
public void getUserInfo_returns_true(){
    UserInfo userInfo = new UserInfo(); //Dummy data - non null userInfo object
    when(repository.getUserInfo("username")).thenReturn(Observable.just(userInfo));

    TestObserver<Boolean> testObserver = new TestObserver<>();
    //the flatMap operator should return true since userInfo is not null
    viewModel.getUserInfo("username").subscribeWith(testObserver); 
    testObserver.assertValue(true);
}

And below is my log

java.lang.AssertionError: Expected: true (class: Boolean), Actual: [] (latch = 1, values = 0, errors = 0, completions = 0)

at io.reactivex.observers.BaseTestConsumer.fail(BaseTestConsumer.java:163)
at io.reactivex.observers.BaseTestConsumer.assertValue(BaseTestConsumer.java:328)
at com.ik.githubbrowser.search_user.SearchUserViewModelTest.getUserInfo_returns_true(SearchUserViewModelTest.java:51)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.mockito.internal.runners.JUnit45AndHigherRunnerImpl.run(JUnit45AndHigherRunnerImpl.java:37)
at org.mockito.runners.MockitoJUnitRunner.run(MockitoJUnitRunner.java:62)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:117)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:42)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:262)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:84)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

java.lang.NullPointerException
at io.reactivex.android.schedulers.HandlerScheduler$HandlerWorker.schedule(HandlerScheduler.java:70)
at io.reactivex.Scheduler$Worker.schedule(Scheduler.java:272)
at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.schedule(ObservableObserveOn.java:161)
at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.onNext(ObservableObserveOn.java:119)
at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onNext(ObservableSubscribeOn.java:58)
at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarXMapObservable.subscribeActual(ObservableScalarXMap.java:164)
at io.reactivex.Observable.subscribe(Observable.java:10903)
at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Exception in thread "RxCachedThreadScheduler-1" java.lang.NullPointerException
at io.reactivex.android.schedulers.HandlerScheduler$HandlerWorker.schedule(HandlerScheduler.java:70)
at io.reactivex.Scheduler$Worker.schedule(Scheduler.java:272)
at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.schedule(ObservableObserveOn.java:161)
at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.onNext(ObservableObserveOn.java:119)
at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onNext(ObservableSubscribeOn.java:58)
at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarXMapObservable.subscribeActual(ObservableScalarXMap.java:164)
at io.reactivex.Observable.subscribe(Observable.java:10903)
at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Process finished with exit code -1

Sedillo

Answer 1

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.

Gurdwara

I used the awaitTerminalEvent() before the assertValue but the unit test doesn't finish. Its stuck in the awaitTerminalEvent call doesnt reach the assertValue line. – Sedillo

This is probably because the Android scheduler never gets an opportunity to run because it is synchronized with the view output. You either need to flush the looper so the scheduler can run or replace the schedulers with a more appropriate type (Usually TestScheduler or TrampolineScheduler) – Gurdwara

Thanks I had to replace both subscribeOn and observerOn schedulers. Have posted the working code as ans as a reference for others – Sedillo

Answer 2

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

}

Sedillo

Answer 3

By defining a new test rule you can keep your test class clean and use this rule in other test classes again

public class RxSchedulerRule implements TestRule {
 @Override
 public Statement apply(Statement base, Description description) {

    return new Statement() {
        @Override
        public void evaluate() throws Throwable {
            RxAndroidPlugins.setInitMainThreadSchedulerHandler(schedulerCallable
                    -> TrampolineScheduler.instance());
            RxJavaPlugins.setIoSchedulerHandler(scheduler
                    -> TrampolineScheduler.instance());
            RxJavaPlugins.setComputationSchedulerHandler(scheduler ->
                    TrampolineScheduler.instance());

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

        }
    };
 }
}

in your test classes

@Rule
public RxSchedulerRule rxSchedulerRule=new RxSchedulerRule();

Knuth