RxJava+Retrofit 2 unit test weird error

Question

I am creating Android application using MVP pattern. For that I am using Retrofit 2 and RxJava. App works fine

But in unit testing I am getting weird error.Same test code sometimes passes, sometimes fails.

Error displays with this message

Wanted but not invoked:
albumView.showProgress();
-> at kz.afckairat.kairat.media.AlbumPresenterTest.checkGetPhotoAlbums(AlbumPresenterTest.java:66)
Actually, there were zero interactions with this mock.

Test class

public class AlbumPresenterTest {

enter code here
private MediaService mediaService;

private AlbumView albumView;

private AlbumPresenterImpl photoAlbumPresenter;

@Before
public void setUp() throws Exception {

    albumView = mock(AlbumView.class);
    mediaService = mock(MediaService.class);

    photoAlbumPresenter = new AlbumPresenterImpl(albumView, mediaService, MediaType.PHOTO);

    RxAndroidPlugins.getInstance().registerSchedulersHook(new RxAndroidSchedulersHook() {
        @Override
        public Scheduler getMainThreadScheduler() {
            return Schedulers.immediate();
        }
    });
}

@After
public void tearDown() {
    RxAndroidPlugins.getInstance().reset();
}

@Test
public void checkGetPhotoAlbums() {
    List<Album> albums = getAlbumList();
    when(mediaService.getPhotoAlbums()).thenReturn(Observable.just(albums));

    photoAlbumPresenter.getAlbums();

    verify(albumView).showProgress();
    verify(albumView).showAlbums(albums);
    verify(albumView).hideProgress();

}

@Test
public void checkGetPhotoAlbumError() {
    String msg = "Error";
    when(mediaService.getPhotoAlbums()).thenReturn(Observable.error(new IOException(msg)));

    photoAlbumPresenter.getAlbums();

    verify(albumView).showProgress();
    verify(albumView).showError(msg);
    verify(albumView).hideProgress();
}


private List<Album> getAlbumList() {
    List<Album> albums = new ArrayList<>();
    Album album = new Album(1, "Test1", "test1.jpg", "01.01.2016", 2);
    albums.add(album);
    album = new Album(2, "Test2", "test2.jpg", "01.01.2016", 2);
    albums.add(album);
    return albums;
}
}

Presenter class which is tested

public class AlbumPresenterImpl implements AlbumPresenter {

private AlbumView view;
private MediaType type;
private List<Album> albums;
private MediaService mediaService;

public AlbumPresenterImpl(AlbumView view, MediaService mediaService, MediaType type) {
    this.view = view;
    this.mediaService = mediaService;
    this.type = type;
}

@Override
public void getAlbums() {
    Observable<List<Album>> observable;

    if (type.equals(MediaType.VIDEO)) {
        observable = mediaService.getVideoAlbums();
    } else {
        observable = mediaService.getPhotoAlbums();
    }

    observable.doOnSubscribe(view::showProgress)
            .doAfterTerminate(view::hideProgress)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(items -> {
                albums = items;
                view.showAlbums(albums);
            }, throwable -> {
                view.showError(throwable.getLocalizedMessage());
            });
}

@Override
public void onResume() {
    if (albums == null) {
        getAlbums();
    }
}

@Override
public void onDestroy() {

}
}

Why sometimes test don't pass?

Thanks a lot!

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

Update

As @Fred wrote problem was in Schedulers

public class RxSchedulersOverrideRule implements TestRule {

private final RxJavaSchedulersHook mRxJavaSchedulersHook = new RxJavaSchedulersHook() {
    @Override
    public Scheduler getIOScheduler() {
        return Schedulers.immediate();
    }

    @Override
    public Scheduler getNewThreadScheduler() {
        return Schedulers.immediate();
    }
};

private final RxAndroidSchedulersHook mRxAndroidSchedulersHook = new RxAndroidSchedulersHook() {
    @Override
    public Scheduler getMainThreadScheduler() {
        return Schedulers.immediate();
    }
};

// Hack to get around RxJavaPlugins.reset() not being public
// See https://github.com/ReactiveX/RxJava/issues/2297
// Hopefully the method will be public in new releases of RxAndroid and we can remove the hack.
private void callResetViaReflectionIn(RxJavaPlugins rxJavaPlugins)
        throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
    Method method = rxJavaPlugins.getClass().getDeclaredMethod("reset");
    method.setAccessible(true);
    method.invoke(rxJavaPlugins);
}

@Override
public Statement apply(final Statement base, Description description) {
    return new Statement() {
        @Override
        public void evaluate() throws Throwable {
            RxAndroidPlugins.getInstance().reset();
            RxAndroidPlugins.getInstance().registerSchedulersHook(mRxAndroidSchedulersHook);
            callResetViaReflectionIn(RxJavaPlugins.getInstance());
            RxJavaPlugins.getInstance().registerSchedulersHook(mRxJavaSchedulersHook);

            base.evaluate();

            RxAndroidPlugins.getInstance().reset();
            callResetViaReflectionIn(RxJavaPlugins.getInstance());
        }
    };
}

}

Code taken from Github a link!

And in Test class

@Rule
public final RxSchedulersOverrideRule mOverrideSchedulersRule = new RxSchedulersOverrideRule();

Answer 1

It seems you override the main thread scheduler with:

RxAndroidPlugins.getInstance().registerSchedulersHook(new RxAndroidSchedulersHook() {
    @Override
    public Scheduler getMainThreadScheduler() {
        return Schedulers.immediate();
    }
});

But from the code, the observables still run on the Schedulers.io() scheduler:

observable.doOnSubscribe(view::showProgress)
        .doAfterTerminate(view::hideProgress)
        .subscribeOn(Schedulers.io())
        // ...

As you may know, the immediate scheduler executes code in the current thread, which I guess since you jump to the io scheduler it's a different one from the one the tests run on.

This will make the test run in one thread and the subscribers/observables in another. This would explain why sometimes the tests pass and sometimes they don't. There's a race condition.

Essential the easiest way is to make sure that at test time you have both observeOn and subscribeOn on Schedulers.immediate() and at run time you have the correct ones, i.e., Schedulers.io() and AndroidSchedulers.mainThread().

You can do this by overriding the schedulers, by passing them as constructors or you could even take a look at this where Dan Lew explains how to use compose to create scheduler transformers. You can then make sure your classes at run time use a proper scheduler transformer and at test time they use some transformer that puts everything on the immediate thread.