RxAndroid, Retrofit 2 unit test Schedulers.io

Question

I am newly learned the RxAndroid but unfortunately the book I studied did not covered any unit test. I have searched a lot on google but failed to find any simple tutorial that cover the RxAndroid unit test in precise way.

I have basically wrote a small REST API using RxAndroid and Retrofit 2. Here is the ApiManager class:

public class MyAPIManager {
    private final MyService myService;

    public MyAPIManager() {
        HttpLoggingInterceptor logging = new HttpLoggingInterceptor();
        // set your desired log level
        logging.setLevel(HttpLoggingInterceptor.Level.BODY);

        OkHttpClient.Builder b = new OkHttpClient.Builder();
        b.readTimeout(35000, TimeUnit.MILLISECONDS);
        b.connectTimeout(35000, TimeUnit.MILLISECONDS);
        b.addInterceptor(logging);
        OkHttpClient client = b.build();

        Retrofit retrofit = new Retrofit.Builder()
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .baseUrl("http://192.168.1.7:8000")
                .client(client)
                .build();

        myService = retrofit.create(MyService.class);
    }

    public Observable<Token> getToken(String username, String password) {
        return myService.getToken(username, password)
                .subscribeOn(Schedulers.io());
                .observeOn(AndroidSchedulers.mainThread());
    }
}

I am trying to create a unit test for getToken. Here is my sample test:

public class MyAPIManagerTest {
    private MyAPIManager myAPIManager;
    @Test
    public void getToken() throws Exception {
        myAPIManager = new MyAPIManager();

        Observable<Token> o = myAPIManager.getToken("hello", "mytoken");
        o.test().assertSubscribed();
        o.test().assertValueCount(1);
    }

}

Due to subscribeOn(Schedulers.io) the above test does not run on main thread due to which it returns 0 value. If I remove subscribeOn(Schedulers.io) from MyAPIManager then it run well and return 1 value. Is there any way to test with Schedulers.io?

Answer 1

Great question and certainly one topic that is lacking a lot of coverage in the community. I would like to share a couple of solutions I personally used and were splendid. These are thought for RxJava 2 but they're available with RxJava 1 just under different names. You will for sure find it if you need it.

1. RxPlugins and RxAndroidPlugins (this is my favourite so far)

So Rx actually provides a mechanism to change the schedulers provided by the static methods inside Schedulers and AndroidSchedulers. These are for example:

RxJavaPlugins.setComputationSchedulerHandler
RxJavaPlugins.setIoSchedulerHandler
RxJavaPlugins.setNewThreadSchedulerHandler
RxJavaPlugins.setSingleSchedulerHandler
RxAndroidPlugins.setInitMainThreadSchedulerHandler

What these do is very simple. They make sure that when you call i.e. Schedulers.io() the returned scheduler is the one you provide in the handler set in setIoSchedulerHandler. Which scheduler do you want to use? Well you want Schedulers.trampoline(). This means that the code will run on the same thread as it was before. If all schedulers are in the trampoline scheduler, then all will be running on the JUnit thread. After the tests are run, you can just clean the whole thing by calling:

RxJavaPlugins.reset()
RxAndroidPlugins.reset()

I think the best approach to this is to use a JUnit rule. Here's a possible one (sorry for the kotlin syntax):

class TrampolineSchedulerRule : TestRule {
  private val scheduler by lazy { Schedulers.trampoline() }

  override fun apply(base: Statement?, description: Description?): Statement =
        object : Statement() {
            override fun evaluate() {
                try {
                    RxJavaPlugins.setComputationSchedulerHandler { scheduler }
                    RxJavaPlugins.setIoSchedulerHandler { scheduler }
                    RxJavaPlugins.setNewThreadSchedulerHandler { scheduler }
                    RxJavaPlugins.setSingleSchedulerHandler { scheduler }
                    RxAndroidPlugins.setInitMainThreadSchedulerHandler { scheduler }
                    base?.evaluate()
                } finally {
                    RxJavaPlugins.reset()
                    RxAndroidPlugins.reset()
                }
            }
        }
}

At the top of your unit test you just need to declare a public attribute annotated with @Rule and instantiated with this class:

@Rule
public TrampolineSchedulerRule rule = new TrampolineSchedulerRule()

in kotlin

@get:Rule
val rule = TrampolineSchedulerRule()

2. Injecting schedulers (a.k.a. dependency injection)

Another possibility is to inject the schedulers in your classes so at test time you can inject again the Schedulers.trampoline() and in your app you can inject the normal schedulers. This might work for a while, but it will soon become cumbersome when you need to inject loads of schedulers just for a simple class. Here's one way of doing this

public class MyAPIManager {
  private final MyService myService;
  private final Scheduler io;
  private final Scheduler mainThread;

  public MyAPIManager(Scheduler io, Scheduler mainThread) {
    // initialise everything
    this.io = io;
    this.mainThread = mainThread;
  }

  public Observable<Token> getToken(String username, String password) {
    return myService.getToken(username, password)
            .subscribeOn(io);
            .observeOn(mainThread);
  }
}

As you can see we can now tell the class the actual schedulers. In your tests you'd do something like:

public class MyAPIManagerTest {
  private MyAPIManager myAPIManager;
  @Test
  public void getToken() throws Exception {
    myAPIManager = new MyAPIManager(
          Schedulers.trampoline(),
          Schedulers.trampoline());

    Observable<Token> o = myAPIManager.getToken("hello", "mytoken");
    o.test().assertSubscribed();
    o.test().assertValueCount(1);
  }
}

The key points are:

• You want it on the Schedulers.trampoline() scheduler to make sure everything's run on the JUnit thread

• You need to be able to modify the schedulers while testing.

That's all. Hope it helps.

=========================================================

Here is Java version which I have used after following above Kotlin example:

public class TrampolineSchedulerRule implements TestRule {
    @Override
    public Statement apply(Statement base, Description description) {
        return new MyStatement(base);
    }

    public class MyStatement extends Statement {
        private final Statement base;

        @Override
        public void evaluate() throws Throwable {
            try {
                RxJavaPlugins.setComputationSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setIoSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setNewThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setSingleSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxAndroidPlugins.setInitMainThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
                base.evaluate();
            } finally {
                RxJavaPlugins.reset();
                RxAndroidPlugins.reset();
            }
        }

        public MyStatement(Statement base) {
            this.base = base;
        }
    }
}