PostsPresenter { public void loadPostsFromAPI() { postsAPI.getPostsObservable().retryWhen(exponentialBackoffForExceptions(1,4,TimeUnit.SECONDS,IOException.class)) .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()) .subscribe(new DisposableObserver<List<Post>>() { @Override public void onNext(List<Post> newPosts) { EventBus.getDefault().post(new NewPostsEvent(newPosts)); } @Override public void onError(Throwable e) { EventBus.getDefault().post(new ErrorEvent()); } @Override public void onComplete() { Log.d("PostsPresenter","Completed"); } }); } @Inject  PostsPresenter(PostsAPI postsAPI); void loadPostsFromAPI(); @SafeVarargs static Function<Observable<? extends Throwable>, Observable<?>> exponentialBackoffForExceptions(final long initialDelay, final int numRetries,
                                    final TimeUnit unit, final Class<? extends Throwable>... errorTypes);  }
@Test public void testShouldSchedulePostsLoadFromAPIOnBackgroundThread() { Observable<List<Post>> postsObservable = (Observable<List<Post>>) mock(Observable.class); when(postsPresenter.postsAPI.getPostsObservable()).thenReturn(postsObservable); when(postsObservable.subscribeOn(Schedulers.io())).thenReturn(postsObservable); when(postsObservable.observeOn(AndroidSchedulers.mainThread())).thenReturn(postsObservable); postsPresenter.loadPostsFromAPI(); verify(postsPresenter.postsAPI).getPostsObservable(); verify(postsObservable).subscribeOn(Schedulers.io()); verify(postsObservable).observeOn(AndroidSchedulers.mainThread()); verify(postsObservable).subscribe(Matchers.<Subscriber<List<Post>>>any()); }