io.reactivex.observers.TestObserver#create ( )源码实例Demo

@Before
public void setUp() throws Exception {
  cConsumer = new TestConsumer<>();
  dAction = new TestAction();

  ObservableTransformer<TestEffect, TestEvent> router =
      RxMobius.<TestEffect, TestEvent>subtypeEffectHandler()
          .addTransformer(A.class, (Observable<A> as) -> as.map(a -> AEvent.create(a.id())))
          .addTransformer(B.class, (Observable<B> bs) -> bs.map(b -> BEvent.create(b.id())))
          .addConsumer(C.class, cConsumer)
          .addAction(D.class, dAction)
          .addFunction(E.class, e -> AEvent.create(e.id()))
          .build();

  publishSubject = PublishSubject.create();
  testSubscriber = TestObserver.create();

  publishSubject.compose(router).subscribe(testSubscriber);
}
 

@Test
public void testReload_notSuccessful() {
    mockNotSuccessfulVoidResult(new IllegalStateException());
    when(mockFirebaseUser.reload())
            .thenReturn(mockVoidTaskResult);

    TestObserver obs = TestObserver.create();

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

    callOnComplete(mockVoidTaskResult);
    obs.dispose();

    obs.assertError(IllegalStateException.class);
}
 

@Test public void testGetStream() {
  mockSuccessfulResultForTask(mockStreamDownloadTask, mockStreamDownloadTaskSnapshot);
  when(mockStorageReference.getStream()).thenReturn(mockStreamDownloadTask);
  when(mockStreamDownloadTaskSnapshot.getTotalByteCount()).thenReturn(1000L);
  when(mockStreamDownloadTaskSnapshot.getBytesTransferred()).thenReturn(1000L);

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

  RxFirebaseStorage.getStream(mockStorageReference).subscribe(obs);

  verifyAddOnCompleteListenerForTask(mockStreamDownloadTask);

  callOnComplete(mockStreamDownloadTask);
  obs.dispose();

  callOnComplete(mockStreamDownloadTask);

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

@Test
public void testRemoveValue() {
    when(mockDatabaseReference.removeValue())
            .thenReturn(mockTask);

    TestObserver sub = TestObserver.create();

    RxFirebaseDatabase.removeValue(mockDatabaseReference)
            .subscribe(sub);

    verifyAddOnCompleteListenerForTask();
    callTaskOnComplete();

    sub.assertComplete();
    sub.assertNoErrors();

    sub.dispose();
}
 

@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();

    obs.assertError(IllegalStateException.class);
}
 

@Test
public void testRemoveValue_Unsuccessful() {
    when(mockDatabaseReference.removeValue())
            .thenReturn(mockTask);

    TestObserver sub = TestObserver.create();

    RxFirebaseDatabase.removeValue(mockDatabaseReference)
            .subscribe(sub);

    verifyAddOnCompleteListenerForTask();
    callTaskOnCompleteWithError(new IllegalStateException());

    sub.assertNotComplete();
    sub.assertError(IllegalStateException.class);

    sub.dispose();
}
 

@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 testRunTransaction() throws Exception {
    TestObserver sub = TestObserver.create();

    RxFirebaseDatabase
            .runTransaction(mockDatabaseReference, mockTransactionTask)
            .subscribe(sub);

    verifyRunTransaction();

    callTransactionOnComplete();
    verifyTransactionTaskCall();

    sub.assertComplete();
    sub.assertNoErrors();

    sub.dispose();
}
 

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

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

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

  callOnComplete(mockUploadTask);
  obs.dispose();

  callOnComplete(mockUploadTask);

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

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

    TestObserver obs = TestObserver.create();

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

    callOnComplete(mockVoidTaskResult);
    obs.dispose();

    obs.assertComplete();
}
 

@Test
public void testSetPriority() {
    when(mockDatabaseReference.setPriority(1))
            .thenReturn(mockTask);

    TestObserver sub = TestObserver.create();

    RxFirebaseDatabase.setPriority(mockDatabaseReference, 1)
            .subscribe(sub);

    verifyAddOnCompleteListenerForTask();
    callTaskOnComplete();

    sub.assertComplete();
    sub.assertNoErrors();

    sub.dispose();
}
 

@Test
public void testDataChanges_DataReference() {
    TestObserver<DataSnapshot> sub = TestObserver.create();

    RxDatabaseReference.changes(mockDatabaseReference)
            .subscribe(sub);

    verifyDataReferenceAddValueEventListener();
    callValueEventOnDataChange("Foo");

    sub.assertNotComplete();
    sub.assertValueCount(1);

    sub.dispose();

    callValueEventOnDataChange("Foo");

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

@Test
public void testRunTransactionLocal() throws Exception {
    TestObserver sub = TestObserver.create();

    RxFirebaseDatabase
            .runTransaction(mockDatabaseReference, mockTransactionTask, true)
            .subscribe(sub);

    verifyRunTransactionLocal();

    callTransactionOnComplete();
    verifyTransactionTaskCall();

    sub.assertComplete();
    sub.assertNoErrors();

    sub.dispose();
}
 

@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 testDataChanges_DataReference() {
    TestObserver<DataSnapshot> sub = TestObserver.create();

    RxFirebaseDatabase.dataChanges(mockDatabaseReference)
            .subscribe(sub);

    verifyDataReferenceAddValueEventListener();
    callValueEventOnDataChange("Foo");

    sub.assertNotComplete();
    sub.assertValueCount(1);

    sub.dispose();

    callValueEventOnDataChange("Foo");

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

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

    when(mockFirebaseAuth.signInWithCredential(mockAuthCredential))
            .thenReturn(mockAuthResultTask);

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

    RxFirebaseAuth
            .signInWithCredential(mockFirebaseAuth, mockAuthCredential)
            .subscribe(obs);

    callOnComplete(mockAuthResultTask);
    obs.dispose();

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

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

@Test
public void testPropagateActionsToEpic() {
    store = Store.create(reducer, epicMiddleware);

    TestObserver<Action> testObserver = TestObserver.create();
    actionsCaptor.getValue().subscribe(testObserver);

    Action testAction = Action.create("TEST");
    store.dispatch(testAction);

    testObserver.assertValue(testAction);
}
 

@Test
public void testChildEvents_DataReference_remove() {
    TestObserver<ChildEvent> sub = TestObserver.create();

    RxFirebaseDatabase.childEvents(mockDatabaseReference)
            .subscribe(sub);

    verifyDataReferenceAddChildEventListener();
    callOnChildRemoved();
    callOnChildRemoved();

    sub.assertNotComplete();
    sub.assertNoErrors();
    sub.assertValueCount(2);

    List events = sub.getEvents().get(0);
    for (Object event : events) {
        assertThat(event)
                .isInstanceOf(ChildRemoveEvent.class);
    }

    sub.dispose();

    callOnChildRemoved();

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

@Test public void testPutFileWithMetadataAndUri() {
  mockSuccessfulResultForTask(mockUploadTask, mockUploadTaskSnapshot);
  when(mockStorageReference.putFile(mockUri, mockStorageMetadata, mockUri)).thenReturn(
      mockUploadTask);
  when(mockUploadTaskSnapshot.getBytesTransferred()).thenReturn(1000L);
  when(mockUploadTaskSnapshot.getTotalByteCount()).thenReturn(1000L);

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

  RxFirebaseStorage.putFile(mockStorageReference, mockUri, mockStorageMetadata, mockUri)
      .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 testAuthStateChanges() {
    TestObserver<FirebaseAuth> obs = TestObserver.create();

    RxFirebaseAuth.changes(mockFirebaseAuth)
            .subscribe(obs);

    callOnAuthStateChanged();

    obs.assertNotComplete();
    obs.assertValueCount(1);

    obs.dispose();

    callOnAuthStateChanged();

    // Assert no more values are emitted
    obs.assertValueCount(1);
}