java类io.reactivex.observers.TestObserver的实例源码

@Test
public void ribActivity_onSaveInstanceStateAndCallbackFlagEnabled_shouldEmitToCallbacks() {
  ActivityController<EmptyActivity> activityController = buildActivity(EmptyActivity.class);
  RibActivity activity = activityController.setup().get();
  TestObserver<ActivityCallbackEvent.SaveInstanceState> testSub = new TestObserver<>();
  activity.callbacks(ActivityCallbackEvent.SaveInstanceState.class).subscribe(testSub);

  android.os.Bundle state = new android.os.Bundle();
  state.putString("hello", "seattle");
  activity.onSaveInstanceState(state);

  testSub.assertValueCount(1);
  ActivityCallbackEvent.SaveInstanceState receivedEvent = testSub.values().get(0);
  assertThat(receivedEvent.getType()).isEqualTo(ActivityCallbackEvent.Type.SAVE_INSTANCE_STATE);
  assertThat(receivedEvent.getOutState()).isNotNull();
  assertThat(receivedEvent.getOutState().getString("hello")).isEqualTo("seattle");
}

@Test
public void oneEvent() {
    TestObserver<String> testObserver = Single.just("1")
        .delay(1, TimeUnit.MILLISECONDS, testScheduler)
        .compose(RxLifecycle.<String, String>bind(lifecycle))
        .test();

    testObserver.assertNoValues();
    testObserver.assertNoErrors();

    lifecycle.onNext("stop");
    testScheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS);

    testObserver.assertNoValues();
    testObserver.assertError(CancellationException.class);
}

@Test
public void start_user_shouldDisplayUser() {
  // given
  TestObserver<UserSelectedEvent> userSelected$ = TestObserver.create();
  User user = mock(User.class);
  given(view.userSelection$()).willReturn(noTriggers());
  UserPresenter presenter = new UserPresenter(Sink.of(userSelected$));

  // when
  presenter.start(user, view);

  // then
  then(view).should().displayUser(user);
  then(view).should().userSelection$();
  then(view).shouldHaveNoMoreInteractions();
  userSelected$.assertNoValues();
}

@Test public void testSignInWithEmailAndPassword() {
  mockSuccessfulAuthResult();

  when(mockFirebaseAuth.signInWithEmailAndPassword("email", "password")).thenReturn(
      mockAuthResultTask);

  TestObserver<FirebaseUser> obs = TestObserver.create();

  RxFirebaseAuth.signInWithEmailAndPassword(mockFirebaseAuth, "email", "password").subscribe(obs);

  callOnComplete(mockAuthResultTask);
  obs.dispose();

  // Ensure no more values are emitted after unsubscribe
  callOnComplete(mockAuthResultTask);

  obs.assertNoErrors();
  obs.assertComplete();
  obs.assertValueCount(1);
}

@Test
public void test_error() {
    Observable<String> observable = Observable.error(new Exception("We got an Exception"));
    TestObserver<String> testObserver = new TestObserver<>();

    observable.subscribe(testObserver);

    testObserver.assertError(Exception.class);
    testObserver.assertNotComplete();
    testObserver.assertErrorMessage("We got an Exception");
    testObserver.assertFailure(exception -> {
        return exception instanceof Exception;
    });

    testObserver.assertFailureAndMessage(Exception.class, "We got an Exception");
}

@Test public void testPutStream() {
  mockSuccessfulResultForTask(mockUploadTask, mockUploadTaskSnapshot);
  when(mockStorageReference.putStream(mockInputStream)).thenReturn(mockUploadTask);
  when(mockUploadTaskSnapshot.getBytesTransferred()).thenReturn(1000L);
  when(mockUploadTaskSnapshot.getTotalByteCount()).thenReturn(1000L);

  TestObserver<UploadTask.TaskSnapshot> obs = TestObserver.create();

  RxFirebaseStorage.putStream(mockStorageReference, mockInputStream).subscribe(obs);

  verifyAddOnCompleteListenerForTask(mockUploadTask);

  callOnComplete(mockUploadTask);
  obs.dispose();

  callOnComplete(mockUploadTask);

  obs.assertNoErrors();
  obs.assertComplete();
  obs.assertValue(new Predicate<UploadTask.TaskSnapshot>() {
    @Override public boolean test(UploadTask.TaskSnapshot taskSnapshot) throws Exception {
      return taskSnapshot.getBytesTransferred() == taskSnapshot.getTotalByteCount()
          && taskSnapshot.getTotalByteCount() == 1000L;
    }
  });
}

@Test
public void deleteSelectedArtists() throws Exception {
    when(repository.deleteArtists(any())).thenReturn(Completable.complete());
    ArtistsPresenter presenter = presenterBuilder.uiModel(twoSelectedArtistsUiModel).build();
    TestObserver<ArtistsUiModel> observer = presenter.getUiModels().skip(1).test();

    presenter.sendUiEvent(DeleteArtistsEvent.create(twoSelectedArtistsUiModel));

    verify(repository).deleteArtists(argThat(selected -> {
        assertThat(selected).hasSize(2).containsOnly(artist1, artist2);

        return true;
    }));
    observer
            .awaitCount(2)
            .assertNoErrors()
            .assertValues(
                    twoSelectedArtistsUiModel.copy()
                            .deletion(true)
                            .build(),
                    withArtistsUiModel
            );
}

@Test
public void testCompleteSuccessful() {
    final String hello = "Hello, world!";
    when(mockHelloTask.getResult())
            .thenReturn(hello);
    when(mockHelloTask.isSuccessful())
            .thenReturn(true);
    when(mockHelloTask.addOnCompleteListener(onComplete.capture()))
            .thenReturn(mockHelloTask);

    TestObserver<String> obs = TestObserver.create();
    RxTask.completes(mockHelloTask).subscribe(obs);
    verify(mockHelloTask).addOnCompleteListener(onComplete.capture());
    onComplete.getValue().onComplete(mockHelloTask);
    obs.assertNoErrors()
       .assertComplete();

    verify(mockHelloTask).addOnCompleteListener(any(OnCompleteListener.class));
}

/**
 * Observable.fromIterable() is another way to create an Observable. It's
 * different than .just() - it is specifically designed to work with
 * Collections. When just is given a collection, it converts it into an
 * Observable that emits each item from the list. Let's understand how the
 * two are different more clearly.
 */
@Test
public void fromIterableShouldCreateAnObservableThatEmitsEachElementFromAnIterable() {

    List<String> sandwichIngredients = Arrays.asList("bread (one)", "bread (two)", "cheese", "mayo", "turkey",
            "lettuce", "pickles", "jalapenos", "Sriracha sauce");
    Observable<String> favoriteFoodsObservable = Observable.fromIterable(sandwichIngredients);
    TestObserver<String> testObserverIteratable = null; // TODO create new TestObserver and subscribe to favoriteFoodsObservable

    assertThat(testObserverIteratable.values()).hasSize(____);
    assertThat(testObserverIteratable.values()).containsAll(null);


    TestObserver<List<String>> testObserverForJust = new TestObserver<>();
    // TODO create obseverable for sandwichIngredients with just operator and subscribe testObserverForJust to it
    assertThat(testObserverForJust.values()).hasSize(1);
    assertThat(testObserverForJust.values()).contains(sandwichIngredients);
    /**
     * ^^ As you can see here, fromIterable() & just() do very different things!
     */
}

private void testBindUntilEvent(LifecycleProvider<FragmentEvent> provider) {
    Fragment fragment = (Fragment) provider;
    startFragment(fragment);

    TestObserver<Object> testObserver = observable.compose(provider.bindUntilEvent(STOP)).test();

    fragment.onAttach(null);
    testObserver.assertNotComplete();
    fragment.onCreate(null);
    testObserver.assertNotComplete();
    fragment.onViewCreated(null, null);
    testObserver.assertNotComplete();
    fragment.onStart();
    testObserver.assertNotComplete();
    fragment.onResume();
    testObserver.assertNotComplete();
    fragment.onPause();
    testObserver.assertNotComplete();
    fragment.onStop();
    testObserver.assertComplete();
}

@Test public void testUnlink() {
  mockSuccessfulAuthResult();
  when(mockFirebaseUser.unlink("provider")).thenReturn(mockAuthTaskResult);

  TestObserver<AuthResult> obs = TestObserver.create();

  RxFirebaseUser.unlink(mockFirebaseUser, "provider").subscribe(obs);

  callOnComplete(mockAuthTaskResult);
  obs.dispose();

  // Ensure no more values are emitted after unsubscribe
  callOnComplete(mockAuthTaskResult);

  obs.assertNoErrors();
  obs.assertComplete();
  obs.assertValueCount(1);
}

@Test public void testDataChangesOfClazz() {
  TestObserver<Optional<String>> sub = TestObserver.create();

  RxFirebaseDatabase.dataChangesOf(mockDatabaseReference, String.class).subscribe(sub);

  verifyAddValueEventListener();
  callValueEventOnDataChange("Foo");

  sub.assertNotComplete();
  sub.assertValueCount(1);
  sub.assertValue(new Predicate<Optional<String>>() {
    @Override public boolean test(Optional<String> stringOptional) throws Exception {
      return stringOptional.isPresent() && "Foo".equals(stringOptional.get());
    }
  });

  sub.dispose();

  callValueEventOnDataChange("Foo");

  // Ensure no more values are emitted after unsubscribe
  sub.assertValueCount(1);
}

@Test
public void states_should_stopObservingStateChanges() {
  //GIVEN
  Store<Integer> store = new Store<>(0);
  TestObserver<Integer> testSubscriber = new TestObserver<>();
  states(store).subscribe(testSubscriber);

  //WHEN
  testSubscriber.dispose();
  store.dispatch(integer -> integer + 1);
  final Integer state = store.getState();

  //THEN
  testSubscriber.assertNotComplete();
  testSubscriber.assertValue(0);
  assertThat(testSubscriber.isDisposed(), is(true));
  assertThat(state, is(1));
}

@Test public void testUpdateMetadata() {
  when(mockStorageMetadata.getName()).thenReturn("metadata");
  mockSuccessfulResultForTask(mockStorageMetadataTask, mockStorageMetadata);

  TestObserver<StorageMetadata> obs = TestObserver.create();
  when(mockStorageReference.updateMetadata(mockStorageMetadata)).thenReturn(
      mockStorageMetadataTask);

  RxFirebaseStorage.updateMetadata(mockStorageReference, mockStorageMetadata).subscribe(obs);

  verifyAddOnCompleteListenerForTask(mockStorageMetadataTask);

  callOnComplete(mockStorageMetadataTask);
  obs.dispose();

  callOnComplete(mockStorageMetadataTask);

  obs.assertNoErrors();
  obs.assertComplete();
  obs.assertValue(new Predicate<StorageMetadata>() {
    @Override public boolean test(StorageMetadata metadata) throws Exception {
      return "metadata".equals(metadata.getName());
    }
  });
}

@Test
public void loadConcerts() throws Exception {
    when(repository.getConcerts()).thenReturn(Observable.just(concerts));
    TestObserver<ConcertsUiModel> testObserver = presenter.getUiModels()
            .skip(1)
            .test();

    presenter.sendUiEvent(LoadConcertsEvent.INSTANCE);

    testObserver
            .awaitCount(2)
            .assertNoErrors()
            .assertNotComplete()
            .assertValueAt(0, model -> {
                assertThat(model.isLoading()).isTrue();

                return true;
            })
            .assertValueAt(1, model -> {
                assertThat(model.isLoading()).isFalse();
                assertThat(model.isLoadingError()).isFalse();
                assertThat(model.getConcerts()).isEqualTo(sortedConcerts);

                return true;
            });
}

@Test
public void testCommand_input_error() throws IOException {
    processor.attach(session);
    session.isAlive().test().awaitDone(1, TimeUnit.SECONDS).assertNoTimeout().assertValue(true);
    doThrow(new IOException()).when(session).writeLine(anyString(), anyBoolean());

    Cmd cmd = Cmd.builder("echo straw", "error berry").build();
    final TestObserver<Cmd.Result> observer = processor.submit(cmd).test().awaitDone(3, TimeUnit.SECONDS).assertNoTimeout();
    final Cmd.Result result = observer.assertValueCount(1).values().get(0);

    session.close().test().awaitDone(1, TimeUnit.SECONDS).assertNoTimeout().assertValue(1);

    assertThat(result.getExitCode(), is(Cmd.ExitCode.SHELL_DIED));

    assertThat(mockSession.getOutputPub().hasSubscribers(), is(false));
    assertThat(mockSession.getErrorPub().hasSubscribers(), is(false));
}

private void testBindUntilEvent(ActivityController<? extends LifecycleProvider<ActivityEvent>> controller) {
    LifecycleProvider<ActivityEvent> activity = controller.get();

    TestObserver<Object> testObserver = observable.compose(activity.bindUntilEvent(STOP)).test();

    controller.create();
    testObserver.assertNotComplete();
    controller.start();
    testObserver.assertNotComplete();
    controller.resume();
    testObserver.assertNotComplete();
    controller.pause();
    testObserver.assertNotComplete();
    controller.stop();
    testObserver.assertComplete();
}

@Test public void testUpdateProfile() {
  mockVoidResult(true);
  when(mockFirebaseUser.updateProfile(mockProfileChangeRequest)).thenReturn(mockVoidTaskResult);

  TestObserver obs = TestObserver.create();

  RxFirebaseUser.updateProfile(mockFirebaseUser, mockProfileChangeRequest).subscribe(obs);

  callOnComplete(mockVoidTaskResult);
  obs.dispose();

  // Ensure no more values are emitted after unsubscribe
  callOnComplete(mockVoidTaskResult);

  obs.assertNoErrors();
  obs.assertComplete();
}

@Test
public void executionObservables_noErrors() {
    RxCommand<String> command = RxCommand.create(o -> {
        if (o == null) {
            return Observable.error(new Exception("something wrong"));
        } else {
            return Observable.just((String) o);
        }
    });


    TestObserver<Observable<String>> testObserver = new TestObserver<>();
    command.executionObservables().subscribe(testObserver);

    command.execute(null);
    command.execute("1");
    command.execute("2");

    testObserver.assertValueCount(3);
    testObserver.assertNoErrors();
    testObserver.assertNotComplete();
}

@Test public void testUpdatePassword_notSuccessful() {
  mockNotSuccessfulVoidResult(new IllegalStateException());
  when(mockFirebaseUser.updatePassword("password")).thenReturn(mockVoidTaskResult);

  TestObserver obs = TestObserver.create();

  RxFirebaseUser.updatePassword(mockFirebaseUser, "password").subscribe(obs);

  callOnComplete(mockVoidTaskResult);
  obs.dispose();

  // Ensure no more values are emitted after unsubscribe
  callOnComplete(mockVoidTaskResult);

  obs.assertError(IllegalStateException.class);
  obs.assertNotComplete();
}

@Test
public void testBindUntilEvent() {
    NaviEmitter fragment = createFragmentEmitter();
    LifecycleProvider<FragmentEvent> provider = createFragmentLifecycleProvider(fragment);

    Observable<Object> observable = create().hide();
    TestObserver<Object> testObserver = observable.compose(provider.bindUntilEvent(STOP)).test();

    fragment.onAttach(null);
    testObserver.assertNotComplete();
    fragment.onCreate(null);
    testObserver.assertNotComplete();
    fragment.onCreate(null);
    testObserver.assertNotComplete();
    fragment.onStart();
    testObserver.assertNotComplete();
    fragment.onResume();
    testObserver.assertNotComplete();
    fragment.onPause();
    testObserver.assertNotComplete();
    fragment.onStop();
    testObserver.assertComplete();
}

@Test
public void loadLocations() throws Exception {
    when(repository.getLocations()).thenReturn(Observable.just(locations));
    TestObserver<LocationsUiModel> testObserver = presenter.getUiModels()
            .skip(1)
            .test();

    presenter.sendUiEvent(LoadLocationsEvent.INSTANCE);

    testObserver
            .awaitCount(2)
            .assertNoErrors()
            .assertNotComplete()
            .assertValueAt(0, model -> {
                assertThat(model.isLoading()).isTrue();

                return true;
            })
            .assertValueAt(1, model -> {
                assertThat(model.isLoading()).isFalse();
                assertThat(model.isLoadingError()).isFalse();
                assertThat(model.getLocations()).isEqualTo(locations);

                return true;
            });
}

@Test public void testDelete() {
  mockVoidResult(true);
  when(mockFirebaseUser.delete()).thenReturn(mockVoidTaskResult);

  TestObserver obs = TestObserver.create();

  RxFirebaseUser.delete(mockFirebaseUser).subscribe(obs);

  callOnComplete(mockVoidTaskResult);
  obs.dispose();

  // Ensure no more values are emitted after unsubscribe
  callOnComplete(mockVoidTaskResult);

  obs.assertNoErrors();
  obs.assertComplete();
}

/**
 * {@link NetworkManager} integration test: get AppData from local REST API mock server,
 * compere parsed response to AppData parsed directly from .json file
 */
@Test
public void getAppDataTest() throws Exception {
    networkManager.configAuth(dummyApiUrl);

    Gson gson = new Gson();
    AppData appData = gson.fromJson(
            readAssetFile(testContext, "api/app_data.json"), AppData.class);

    TestObserver<AppData> testObserver = new CustomTestObserver<>(appData);
    networkManager.getAppData().subscribe(testObserver);

    testObserver.assertSubscribed();
    testObserver.assertValueCount(1);
    testObserver.assertNoErrors();
    testObserver.onComplete();
}

@Test public void testPutBytes_notSuccessful() {
  mockNotSuccessfulResultForTask(mockUploadTask, new IllegalStateException());
  when(mockStorageReference.putBytes(new byte[] { 1, 2, 3 })).thenReturn(mockUploadTask);

  TestObserver<UploadTask.TaskSnapshot> obs = TestObserver.create();

  RxFirebaseStorage.putBytes(mockStorageReference, new byte[] { 1, 2, 3 }).subscribe(obs);
  verifyAddOnCompleteListenerForTask(mockUploadTask);

  callOnComplete(mockUploadTask);
  obs.dispose();

  callOnComplete(mockUploadTask);

  obs.assertError(IllegalStateException.class);
  obs.assertNoValues();
}

@Test
public void eventOutOfLifecycle() {
    PublishSubject<String> stream = PublishSubject.create();
    PublishSubject<String> lifecycle = PublishSubject.create();

    TestObserver<String> testObserver = stream
        .compose(RxLifecycle.<String, String>bind(lifecycle, CORRESPONDING_EVENTS))
        .test();

    // Event is out of lifecycle, but this just results in completing the stream
    lifecycle.onNext("destroy");
    stream.onNext("1");

    testObserver.assertNoValues();
    testObserver.assertComplete();
}

private void testLifecycle(ActivityController<? extends LifecycleOwner> controller) {
    LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(controller.get());

    TestObserver<Lifecycle.Event> testObserver = provider.lifecycle().test();

    controller.create();
    controller.start();
    controller.resume();
    controller.pause();
    controller.stop();
    controller.destroy();

    testObserver.assertValues(
            Lifecycle.Event.ON_CREATE,
            Lifecycle.Event.ON_START,
            Lifecycle.Event.ON_RESUME,
            Lifecycle.Event.ON_PAUSE,
            Lifecycle.Event.ON_STOP,
            Lifecycle.Event.ON_DESTROY
    );
}

@SuppressWarnings("Duplicates") @Test public void testGetFileUri_notSuccessful() {
  mockNotSuccessfulResultForTask(mockFileDownloadTask, new IllegalStateException());
  when(mockStorageReference.getFile(mockUri)).thenReturn(mockFileDownloadTask);

  TestObserver<FileDownloadTask.TaskSnapshot> obs = TestObserver.create();

  RxFirebaseStorage.getFile(mockStorageReference, mockUri).subscribe(obs);
  verifyAddOnCompleteListenerForTask(mockFileDownloadTask);

  callOnComplete(mockFileDownloadTask);
  obs.dispose();

  callOnComplete(mockFileDownloadTask);

  obs.assertError(IllegalStateException.class);
  obs.assertNoValues();
}

@Test
public void abstractButtonAction() {
    EventQueue.invokeLater(new Runnable() {
        @Override
        public void run() {
            JButton button = new JButton();

            TestObserver<ActionEvent> to = actions(button)
            .test();

            button.doClick();

            to
              .assertSubscribed()
              .assertValueCount(1)
              .assertNotTerminated();

            button.doClick(3);

            to.assertValueCount(4)
            .assertNotTerminated();

            to.dispose();

            button.doClick();

            to.assertValueCount(4)
            .assertNotTerminated();
        }
    });
}

@Test
public void testIsViewReady_BeforeAttachView_ShouldCallValueFalse() throws Exception {
    mPresenter.create();

    final TestObserver<Boolean> test = RxTiPresenterUtils.isViewReady(mPresenter).test();

    test.assertValue(false);
}