验证在RxJava订阅服务器的onNext中调用该方法

我有以下方法使用Retrofit服务接口从API获取一些数据，然后与 view 接口进行交互 .

@Override
@VisibleForTesting
public void fetchPhotos(@Nullable PhotosService service, @Nullable Scheduler subscribeOn) {
    view.showLoading();

    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(Constants.PLACEHOLDER_API_BASE_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

    if (service == null) service = retrofit.create(PhotosService.class);
    if (subscribeOn == null) subscribeOn = Schedulers.newThread();

    service.listPhotos()
            .subscribeOn(subscribeOn)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(photoList -> {
                Log.d(TAG, "got photos " + photoList.toString());
                view.unshowLoading();
            }, throwable -> {
                Log.d(TAG, "error " + throwable.toString());
                view.unshowLoading();
                view.displayError(throwable.toString(), v -> fetchPhotos());
            });
}

我想测试在onNext中调用 view.unshowLoading() .

这是我的测试：

@Test
public void viewUnshowsLoadingAfterFetchingPhotos() {

    PhotosListView view = Mockito.mock(PhotosListView.class);

    PhotosListPresenter presenter = new PhotosListPresenterImpl(view);

    presenter.fetchPhotos(() -> Observable.create(new Observable.OnSubscribe<List<Photo>>() {
        @Override
        public void call(Subscriber<? super List<Photo>> subscriber) {

            subscriber.onNext(new ArrayList<Photo>());

        }
    }), Schedulers.immediate());

    Mockito.verify(view).unshowLoading();


}

我明确传入 Scheduler Schedulers.immediate() 以确保在订阅线程上立即调用 onNext() .

当我通过我的方法调试时， onNext() 未被调用 . 我做错了什么或者我怎么能最好地测试这个？

EDIT: This article带给我一些东西：

如果要更改执行操作的线程，可以调用subscribeOn（） . 要返回主线程，请使用observeOn（AndroidSchedulers.mainThread（）） . 但是，请注意，每当您将操作强制到特定线程时，它总是会使订阅异步 .

当我省略了

.subscribeOn(subscribeOn)
        .observeOn(AndroidSchedulers.mainThread())

部分，测试按预期工作 . 当没有传入调度程序时，我已将我的方法重新排列为 observeOn() 或 subscribeOn() 中没有调用：

public void fetchPhotos(@Nullable PhotosService service, @Nullable Scheduler subscribeOn, @Nullable Scheduler observeOn) {
    view.showLoading();

    if (service == null) service = createService();

    Observable<List<Photo>> observable = service.listPhotos();

    if (subscribeOn != null) observable = observable.subscribeOn(subscribeOn);
    if (observeOn != null) observable = observable.observeOn(observeOn);

    observable.subscribe(photoList -> {
        Log.d(TAG, "got photos " + photoList.toString());
        view.unshowLoading();
    }, throwable -> {
        Log.d(TAG, "error " + throwable.toString());
        view.unshowLoading();
        view.displayError(throwable.toString(), v -> fetchPhotos());
    });
}

看起来有点笨拙，但有效 .

任何想法仍然欢迎:)