Android unit testing how to test observable and Subscriber

Question

Hi Following is my part of code which I want to test with various requests but I am unable to test as it is observable subscriber patter any idea how to mock subscriber or how to wait for a result in unit testing

api.loadData(request)
            .observeOn(schedulers.main())
            .doOnNext(this::addDataCheck)
            .subscribe(
                    data -> onDataLodaded(listener, data),
                    e -> onAdFailedToLoad(listener, Api.error(request), e)
            )
            .addTo(disposables);

Answer 1

If you have access to the Observable you could use the test() method of Observable to get a TestObserver or use blocking*() methods like blockingSubscribe(), blockingFirst(), etc.

In most cases you won't have access to them or you have already called subscribe() on them. Then the answer depends on the JUnit version you are using.
In general you will have to change the Handler for the RxJava and RxAndroid schedulers like this:

RxJavaPlugins.setComputationSchedulerHandler(scheduler -> Schedulers.trampoline());
RxJavaPlugins.setIoSchedulerHandler         (scheduler -> Schedulers.trampoline());
RxJavaPlugins.setSingleSchedulerHandler     (scheduler -> Schedulers.trampoline());
RxJavaPlugins.setNewThreadSchedulerHandler  (scheduler -> Schedulers.trampoline());

RxAndroidPlugins.setMainThreadSchedulerHandler(scheduler -> Schedulers.trampoline());

With this all your tasks will be run on the test thread of the JUnit test.

JUnit 4

JUnit 4 uses Rules to change behaviour for all test methods, like changing the Schedulers of RxJava, for this extend TestRule and implement the methods as needed. Rules are used like this:

public class TestClass {
    @Rule public TestRule testRule = new MyTestRule();

    // your tests
}

See the official wiki or for example here for more information about Rules.

JUnit 5

JUnit 5 uses Extensions instead of Rules to apply modifications to all test methods. You could for example use this class:

import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.ExtensionContext;

import io.reactivex.rxjava3.android.plugins.RxAndroidPlugins;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxInstantExecutorExtension implements BeforeAllCallback {
    @Override
    public void beforeAll(ExtensionContext context) {
        RxJavaPlugins.setComputationSchedulerHandler(scheduler -> Schedulers.trampoline());
        RxJavaPlugins.setIoSchedulerHandler(scheduler -> Schedulers.computation());
        RxJavaPlugins.setSingleSchedulerHandler(scheduler -> Schedulers.trampoline());
        RxJavaPlugins.setNewThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
        RxAndroidPlugins.setMainThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
    }
}

in your test like this:

@ExtendWith(RxInstantExecutorExtension.class)
public class TestClass { // your tests }

More information about extensions are in the official user guide or for example here.

Because the question is tagged with "android", here a reminder that a Rule/Extension for the scheduling of LiveData has to be added if it is used.