Remove assertTrustedObservable

- Was no longer working once onError(Exception) changed to onError(Throwable) as the assertions throwing java.lang.Error no longer bypassed the composed Observables.
- It was confusing while locally contextual JUnit verifications as used everywhere else are clear, readable and localized to each test

Author: benjchristensen

rxjava-core/src/main/java/rx/operators/OperationTake.java

@@ -158,7 +158,7 @@ public static class UnitTest {
         @Test
         public void testTake1() {
             Observable<String> w = Observable.from("one", "two", "three");
-            Observable<String> take = Observable.create(assertTrustedObservable(take(w, 2)));
+            Observable<String> take = Observable.create(take(w, 2));
 
             @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
@@ -173,7 +173,7 @@ public void testTake1() {
         @Test
         public void testTake2() {
             Observable<String> w = Observable.from("one", "two", "three");
-            Observable<String> take = Observable.create(assertTrustedObservable(take(w, 1)));
+            Observable<String> take = Observable.create(take(w, 1));
 
             @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
@@ -197,7 +197,17 @@ public Subscription call(Observer<String> observer)
                     return Subscriptions.empty();
                 }
             });
-            Observable.create(assertTrustedObservable(take(source, 1))).toBlockingObservable().last();
+
+            @SuppressWarnings("unchecked")
+            Observer<String> aObserver = mock(Observer.class);
+
+            Observable.create(take(source, 1)).subscribe(aObserver);
+
+            verify(aObserver, times(1)).onNext("one");
+            // even though onError is called we take(1) so shouldn't see it
+            verify(aObserver, never()).onError(any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+            verifyNoMoreInteractions(aObserver);
         }
 
         @Test
@@ -221,9 +231,19 @@ public void unsubscribe()
                     };
                 }
             });
-            Observable.create(assertTrustedObservable(take(source, 0))).toBlockingObservable().lastOrDefault("ok");
+
+            @SuppressWarnings("unchecked")
+            Observer<String> aObserver = mock(Observer.class);
+
+            Observable.create(take(source, 0)).subscribe(aObserver);
             assertTrue("source subscribed", subscribed.get());
             assertTrue("source unsubscribed", unSubscribed.get());
+
+            verify(aObserver, never()).onNext(anyString());
+            // even though onError is called we take(0) so shouldn't see it
+            verify(aObserver, never()).onError(any(Throwable.class));
+            verify(aObserver, times(1)).onCompleted();
+            verifyNoMoreInteractions(aObserver);
         }
 
         @Test
@@ -233,7 +253,7 @@ public void testUnsubscribeAfterTake() {
 
             @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
-            Observable<String> take = Observable.create(assertTrustedObservable(take(w, 1)));
+            Observable<String> take = Observable.create(take(w, 1));
             take.subscribe(aObserver);
 
             // wait for the Observable to complete
@@ -248,7 +268,9 @@ public void testUnsubscribeAfterTake() {
             verify(aObserver, times(1)).onNext("one");
             verify(aObserver, never()).onNext("two");
             verify(aObserver, never()).onNext("three");
+            verify(aObserver, times(1)).onCompleted();
             verify(s, times(1)).unsubscribe();
+            verifyNoMoreInteractions(aObserver);
         }
 
         private static class TestObservable extends Observable<String> {

rxjava-core/src/main/java/rx/operators/OperatorTester.java

@@ -55,18 +55,6 @@ private OperatorTester() {
 
     public static class UnitTest {
 
-        public static <T> Func1<Observer<T>, Subscription> assertTrustedObservable(final Func1<Observer<T>, Subscription> source)
-        {
-            return new Func1<Observer<T>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<T> observer)
-                {
-                    return source.call(new TestingObserver<T>(observer));
-                }
-            };
-        }
-
         /**
          * Used for mocking of Schedulers since many Scheduler implementations are static/final.
          * 
@@ -77,221 +65,6 @@ public static Scheduler forwardingScheduler(Scheduler underlying) {
             return new ForwardingScheduler(underlying);
         }
 
-        public static class TestingObserver<T> implements Observer<T> {
-
-            private final Observer<T> actual;
-            private final AtomicBoolean isFinished = new AtomicBoolean(false);
-            private final AtomicBoolean isInCallback = new AtomicBoolean(false);
-
-            public TestingObserver(Observer<T> actual) {
-                this.actual = actual;
-            }
-
-            @Override
-            public void onCompleted() {
-                assertFalse("previous call to onCompleted() or onError()", !isFinished.compareAndSet(false, true));
-                assertFalse("concurrent callback pending", !isInCallback.compareAndSet(false, true));
-                actual.onCompleted();
-                isInCallback.set(false);
-            }
-
-            @Override
-            public void onError(Exception e) {
-                assertFalse("previous call to onCompleted() or onError()", !isFinished.compareAndSet(false, true));
-                assertFalse("concurrent callback pending", !isInCallback.compareAndSet(false, true));
-                actual.onError(e);
-                isInCallback.set(false);
-            }
-
-            @Override
-            public void onNext(T args) {
-                assertFalse("previous call to onCompleted() or onError()", isFinished.get());
-                assertFalse("concurrent callback pending", !isInCallback.compareAndSet(false, true));
-                actual.onNext(args);
-                isInCallback.set(false);
-            }
-
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testDoubleCompleted() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onCompleted();
-                    observer.onCompleted();
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testCompletedError() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onCompleted();
-                    observer.onError(new Exception());
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testCompletedNext() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onCompleted();
-                    observer.onNext("one");
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testErrorCompleted() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onError(new Exception());
-                    observer.onCompleted();
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testDoubleError() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onError(new Exception());
-                    observer.onError(new Exception());
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test(expected = AssertionError.class)
-        public void testErrorNext() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onError(new Exception());
-                    observer.onNext("one");
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test
-        public void testNextCompleted() {
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(Observer<String> observer)
-                {
-                    observer.onNext("one");
-                    observer.onCompleted();
-                    return Subscriptions.empty();
-                }
-            })).toBlockingObservable().lastOrDefault("end");
-        }
-
-        @Test
-        public void testConcurrentNextNext() {
-            final List<Thread> threads = new ArrayList<Thread>();
-            final AtomicReference<Throwable> threadFailure = new AtomicReference<Throwable>();
-            Observable.create(assertTrustedObservable(new Func1<Observer<String>, Subscription>()
-            {
-                @Override
-                public Subscription call(final Observer<String> observer)
-                {
-                    threads.add(new Thread(new Runnable()
-                    {
-                        @Override
-                        public void run()
-                        {
-                            observer.onNext("one");
-                        }
-                    }));
-                    threads.add(new Thread(new Runnable()
-                    {
-                        @Override
-                        public void run()
-                        {
-                            observer.onNext("two");
-                        }
-                    }));
-                    return Subscriptions.empty();
-                }
-            })).subscribe(new SlowObserver());
-            for (Thread thread : threads) {
-                thread.setUncaughtExceptionHandler(new UncaughtExceptionHandler()
-                {
-                    @Override
-                    public void uncaughtException(Thread thread, Throwable throwable)
-                    {
-                        threadFailure.set(throwable);
-                    }
-                });
-                thread.start();
-            }
-            for (Thread thread : threads) {
-                try {
-                    thread.join();
-                } catch (InterruptedException ignored) {
-                }
-            }
-            // Junit seems pretty bad about exposing test failures inside of created threads.
-            assertNotNull("exception thrown by thread", threadFailure.get());
-            assertEquals("class of exception thrown by thread", AssertionError.class, threadFailure.get().getClass());
-        }
-
-        private static class SlowObserver implements Observer<String>
-        {
-            @Override
-            public void onCompleted()
-            {
-                try {
-                    Thread.sleep(10);
-                } catch (InterruptedException ignored) {
-                }
-            }
-
-            @Override
-            public void onError(Exception e)
-            {
-                try {
-                    Thread.sleep(10);
-                } catch (InterruptedException ignored) {
-                }
-            }
-
-            @Override
-            public void onNext(String args)
-            {
-                try {
-                    Thread.sleep(10);
-                } catch (InterruptedException ignored) {
-                }
-            }
-        }
-
         public static class ForwardingScheduler extends Scheduler {
             private final Scheduler underlying;