Suppress UndeliverableException handling in tests (ReactiveX#6987) (ReactiveX#6996)

* Suppress UndeliverableException handling in tests (ReactiveX#6987)

Use a JUnit TestRule to suppress UndeliverableException handling in test
methods annotated with SuppressUndeliverable.

* Write requested changes (ReactiveX#6996)

* remove changes to RxJavaPlugins

* move @rule field into RxJavaTest

* use lambda SuppressUndeliverableRule

Author: akluball

src/test/java/io/reactivex/rxjava3/core/RxJavaTest.java

@@ -20,9 +20,13 @@
 import org.junit.*;
 import org.junit.rules.Timeout;
 
+import io.reactivex.rxjava3.testsupport.SuppressUndeliverableRule;
+
 public abstract class RxJavaTest {
     @Rule
     public Timeout globalTimeout = new Timeout(5, TimeUnit.MINUTES);
+    @Rule
+    public final SuppressUndeliverableRule suppressUndeliverableRule = new SuppressUndeliverableRule();
 
     /**
      * Announce creates a log print preventing Travis CI from killing the build.

src/test/java/io/reactivex/rxjava3/internal/jdk8/ParallelCollectorTest.java

@@ -148,6 +148,7 @@ public Set<Characteristics> characteristics() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void collectorCombinerCrash() {
         Flowable.range(1, 5)
         .parallel()

src/test/java/io/reactivex/rxjava3/internal/operators/completable/CompletableFromCallableTest.java

@@ -29,9 +29,10 @@
 import io.reactivex.rxjava3.exceptions.TestException;
 import io.reactivex.rxjava3.observers.TestObserver;
 import io.reactivex.rxjava3.schedulers.Schedulers;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class CompletableFromCallableTest extends RxJavaTest {
+
     @Test
     public void fromCallable() {
         final AtomicInteger atomicInteger = new AtomicInteger();
@@ -161,6 +162,7 @@ public String answer(InvocationOnMock invocation) throws Throwable {
     }
 
     @Test
+    @SuppressUndeliverable
     public void fromActionErrorsDisposed() {
         final AtomicInteger calls = new AtomicInteger();
         Completable.fromCallable(new Callable<Object>() {

src/test/java/io/reactivex/rxjava3/internal/operators/completable/CompletableFromSupplierTest.java

@@ -30,7 +30,7 @@
 import io.reactivex.rxjava3.functions.Supplier;
 import io.reactivex.rxjava3.observers.TestObserver;
 import io.reactivex.rxjava3.schedulers.Schedulers;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class CompletableFromSupplierTest extends RxJavaTest {
 
@@ -163,6 +163,7 @@ public String answer(InvocationOnMock invocation) throws Throwable {
     }
 
     @Test
+    @SuppressUndeliverable
     public void fromActionErrorsDisposed() {
         final AtomicInteger calls = new AtomicInteger();
         Completable.fromSupplier(new Supplier<Object>() {

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java

@@ -112,6 +112,7 @@ public void empty() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void error() {
         Flowable<String> sourceStrings = Flowable.just("one", "two", "three", "four", "five", "six");
         Flowable<String> errorSource = Flowable.error(new TestException("forced failure"));
@@ -1196,6 +1197,7 @@ public void keySelectorThrows() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void valueSelectorThrows() {
         Flowable<Integer> source = Flowable.just(0, 1, 2, 3, 4, 5, 6);
 
@@ -1251,6 +1253,7 @@ public void accept(GroupedFlowable<Integer, Integer> t1) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void error2() {
         Flowable<Integer> source = Flowable.concat(Flowable.just(0),
                 Flowable.<Integer> error(new TestException("Forced failure")));
@@ -1686,6 +1689,7 @@ public void accept(GroupedFlowable<Integer, Integer> g) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void keySelectorAndDelayError() {
         Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
         .groupBy(Functions.<Integer>identity(), true)
@@ -1700,6 +1704,7 @@ public Flowable<Integer> apply(GroupedFlowable<Integer, Integer> g) throws Excep
     }
 
     @Test
+    @SuppressUndeliverable
     public void keyAndValueSelectorAndDelayError() {
         Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
         .groupBy(Functions.<Integer>identity(), Functions.<Integer>identity(), true)
@@ -1853,6 +1858,7 @@ public Publisher<Integer> apply(GroupedFlowable<Object, Integer> g) throws Excep
     }
 
     @Test
+    @SuppressUndeliverable
     public void groupError() {
         Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
         .groupBy(Functions.justFunction(1), true)

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithFlowableTest.java

@@ -623,6 +623,7 @@ public void run() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void disposeMainBoundaryErrorRace() {
         final TestException ex = new TestException();
 

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithStartEndFlowableTest.java

@@ -32,7 +32,7 @@
 import io.reactivex.rxjava3.processors.*;
 import io.reactivex.rxjava3.schedulers.TestScheduler;
 import io.reactivex.rxjava3.subscribers.*;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class FlowableWindowWithStartEndFlowableTest extends RxJavaTest {
 
@@ -324,6 +324,7 @@ public Flowable<Integer> apply(Integer v) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void endError() {
         PublishProcessor<Integer> source = PublishProcessor.create();
         PublishProcessor<Integer> start = PublishProcessor.create();

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowWithTimeTest.java

@@ -31,7 +31,7 @@
 import io.reactivex.rxjava3.processors.*;
 import io.reactivex.rxjava3.schedulers.*;
 import io.reactivex.rxjava3.subscribers.*;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class FlowableWindowWithTimeTest extends RxJavaTest {
 
@@ -556,6 +556,7 @@ public void restartTimer() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void exactBoundaryError() {
         Flowable.error(new TestException())
         .window(1, TimeUnit.DAYS, Schedulers.single(), 2, true)
@@ -973,6 +974,7 @@ public void accept(Flowable<Integer> v) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void exactTimeBoundNoInterruptWindowOutputOnError() throws Exception {
         final AtomicBoolean isInterrupted = new AtomicBoolean();
 
@@ -1053,6 +1055,7 @@ public void accept(Flowable<Integer> v) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void exactTimeAndSizeBoundNoInterruptWindowOutputOnError() throws Exception {
         final AtomicBoolean isInterrupted = new AtomicBoolean();
 
@@ -1133,6 +1136,7 @@ public void accept(Flowable<Integer> v) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void skipTimeAndSizeBoundNoInterruptWindowOutputOnError() throws Exception {
         final AtomicBoolean isInterrupted = new AtomicBoolean();
 

src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeUsingTest.java

@@ -481,6 +481,7 @@ public void run() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void errorDisposeRace() {
         for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
 

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCreateTest.java

@@ -33,6 +33,7 @@
 public class ObservableCreateTest extends RxJavaTest {
 
     @Test
+    @SuppressUndeliverable
     public void basic() {
         final Disposable d = Disposable.empty();
 
@@ -58,6 +59,7 @@ public void subscribe(ObservableEmitter<Integer> e) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void basicWithCancellable() {
         final Disposable d1 = Disposable.empty();
         final Disposable d2 = Disposable.empty();
@@ -91,6 +93,7 @@ public void cancel() throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void basicWithError() {
         final Disposable d = Disposable.empty();
 
@@ -115,6 +118,7 @@ public void subscribe(ObservableEmitter<Integer> e) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void basicSerialized() {
         final Disposable d = Disposable.empty();
 
@@ -142,6 +146,7 @@ public void subscribe(ObservableEmitter<Integer> e) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void basicWithErrorSerialized() {
         final Disposable d = Disposable.empty();
 
@@ -209,6 +214,7 @@ public void unsafeWithObservable() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void createNullValue() {
         final Throwable[] error = { null };
 
@@ -232,6 +238,7 @@ public void subscribe(ObservableEmitter<Integer> e) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void createNullValueSerialized() {
         final Throwable[] error = { null };
 

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableGroupByTest.java

@@ -96,6 +96,7 @@ public void empty() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void error() {
         Observable<String> sourceStrings = Observable.just("one", "two", "three", "four", "five", "six");
         Observable<String> errorSource = Observable.error(new RuntimeException("forced failure"));
@@ -1178,6 +1179,7 @@ public void keySelectorThrows() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void valueSelectorThrows() {
         Observable<Integer> source = Observable.just(0, 1, 2, 3, 4, 5, 6);
 
@@ -1233,6 +1235,7 @@ public void accept(GroupedObservable<Integer, Integer> t1) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void error2() {
         Observable<Integer> source = Observable.concat(Observable.just(0),
                 Observable.<Integer> error(new TestException("Forced failure")));
@@ -1446,6 +1449,7 @@ public Integer apply(Integer i) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void keySelectorAndDelayError() {
         Observable.just(1).concatWith(Observable.<Integer>error(new TestException()))
         .groupBy(Functions.<Integer>identity(), true)
@@ -1460,6 +1464,7 @@ public ObservableSource<Integer> apply(GroupedObservable<Integer, Integer> g) th
     }
 
     @Test
+    @SuppressUndeliverable
     public void keyAndValueSelectorAndDelayError() {
         Observable.just(1).concatWith(Observable.<Integer>error(new TestException()))
         .groupBy(Functions.<Integer>identity(), Functions.<Integer>identity(), true)

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableGroupJoinTest.java

@@ -479,6 +479,7 @@ public Observable<Integer> apply(Integer r, Observable<Integer> l) throws Except
     }
 
     @Test
+    @SuppressUndeliverable
     public void innerErrorRight() {
         Observable.just(1)
         .groupJoin(

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableLiftTest.java

@@ -19,10 +19,12 @@
 
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.testsupport.SuppressUndeliverable;
 
 public class ObservableLiftTest extends RxJavaTest {
 
     @Test
+    @SuppressUndeliverable
     public void callbackCrash() {
         try {
             Observable.just(1)

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableRefCountTest.java

@@ -855,6 +855,7 @@ protected void subscribeActual(Observer<? super Object> observer) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void badSourceSubscribe() {
         BadObservableSubscribe bo = new BadObservableSubscribe();
 
@@ -881,6 +882,7 @@ public void badSourceDispose() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void badSourceConnect() {
         BadObservableConnect bo = new BadObservableConnect();
 
@@ -922,6 +924,7 @@ protected void subscribeActual(Observer<? super Object> observer) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void badSourceSubscribe2() {
         BadObservableSubscribe2 bo = new BadObservableSubscribe2();
 
@@ -959,6 +962,7 @@ protected void subscribeActual(Observer<? super Object> observer) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void badSourceCompleteDisconnect() {
         BadObservableConnect2 bo = new BadObservableConnect2();
 

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableRetryWithPredicateTest.java

@@ -37,6 +37,7 @@
 import io.reactivex.rxjava3.testsupport.*;
 
 public class ObservableRetryWithPredicateTest extends RxJavaTest {
+
     BiPredicate<Integer, Throwable> retryTwice = new BiPredicate<Integer, Throwable>() {
         @Override
         public boolean test(Integer t1, Throwable t2) {
@@ -390,6 +391,7 @@ public void dontRetry() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void retryDisposeRace() {
         for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
             final PublishSubject<Integer> ps = PublishSubject.create();
@@ -438,6 +440,7 @@ public boolean test(Integer n, Throwable e) throws Exception {
     }
 
     @Test
+    @SuppressUndeliverable
     public void retryBiPredicateDisposeRace() {
         for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
             final PublishSubject<Integer> ps = PublishSubject.create();

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableTakeTest.java

@@ -34,7 +34,7 @@
 import io.reactivex.rxjava3.plugins.RxJavaPlugins;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import io.reactivex.rxjava3.subjects.PublishSubject;
-import io.reactivex.rxjava3.testsupport.TestHelper;
+import io.reactivex.rxjava3.testsupport.*;
 
 public class ObservableTakeTest extends RxJavaTest {
 
@@ -111,6 +111,7 @@ public Integer apply(Integer t1) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void takeDoesntLeakErrors() {
         Observable<String> source = Observable.unsafeCreate(new ObservableSource<String>() {
             @Override

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableTakeWhileTest.java

@@ -100,6 +100,7 @@ public boolean test(String input) {
     }
 
     @Test
+    @SuppressUndeliverable
     public void takeWhileDoesntLeakErrors() {
         Observable<String> source = Observable.unsafeCreate(new ObservableSource<String>() {
             @Override

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableTimeoutWithSelectorTest.java

@@ -40,6 +40,7 @@
 import io.reactivex.rxjava3.testsupport.*;
 
 public class ObservableTimeoutWithSelectorTest extends RxJavaTest {
+
     @Test
     public void timeoutSelectorNormal1() {
         PublishSubject<Integer> source = PublishSubject.create();
@@ -493,6 +494,7 @@ public void withOtherMainError() {
     }
 
     @Test
+    @SuppressUndeliverable
     public void badSourceTimeout() {
         new Observable<Integer>() {
             @Override