Have unit tests extends RxJavaTest - 2 (#6594)

This commit updates the unit tests of the following operators:
* flowable
* maybe
* observable
* disposable
* observer
* parallel
* processors
* schedulers

and RxJava plugin.

Related: #6583

Author: RomanWuattier

src/test/java/io/reactivex/flowable/FlowableBackpressureTests.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 
+import io.reactivex.RxJavaTest;
 import org.junit.*;
 import org.junit.rules.TestName;
 import org.reactivestreams.*;
@@ -31,7 +32,7 @@
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subscribers.*;
 
-public class FlowableBackpressureTests {
+public class FlowableBackpressureTests extends RxJavaTest {
 
     static final class FirehoseNoBackpressure extends AtomicBoolean implements Subscription {
 
@@ -235,36 +236,6 @@ public Publisher<Integer> apply(Integer i) {
         assertTrue(c.get() < Flowable.bufferSize());
     }
 
-    @Test
-    @Ignore("The test is non-deterministic and can't be made deterministic")
-    public void flatMapAsync() {
-        int num = (int) (Flowable.bufferSize() * 2.1);
-        AtomicInteger c = new AtomicInteger();
-        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
-
-        incrementingIntegers(c)
-        .subscribeOn(Schedulers.computation())
-        .flatMap(new Function<Integer, Publisher<Integer>>() {
-            @Override
-            public Publisher<Integer> apply(Integer i) {
-                return incrementingIntegers(new AtomicInteger())
-                        .take(10)
-                        .subscribeOn(Schedulers.computation());
-            }
-        }
-        )
-        .take(num).subscribe(ts);
-
-        ts.awaitDone(5, TimeUnit.SECONDS);
-        ts.assertNoErrors();
-        System.out.println("testFlatMapAsync => Received: " + ts.values().size() + "  Emitted: " + c.get() + " Size: " + Flowable.bufferSize());
-        assertEquals(num, ts.values().size());
-        // even though we only need 10, it will request at least Flowable.bufferSize(), and then as it drains keep requesting more
-        // and then it will be non-deterministic when the take() causes the unsubscribe as it is scheduled on 10 different schedulers (threads)
-        // normally this number is ~250 but can get up to ~1200 when Flowable.bufferSize() == 1024
-        assertTrue(c.get() <= Flowable.bufferSize() * 2);
-    }
-
     @Test
     public void zipSync() {
         int num = (int) (Flowable.bufferSize() * 4.1);
@@ -479,7 +450,7 @@ public void onNext(Integer t) {
         assertEquals(20, batches.get());
     }
 
-    @Test(timeout = 2000)
+    @Test
     public void firehoseFailsAsExpected() {
         AtomicInteger c = new AtomicInteger();
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
@@ -515,7 +486,7 @@ public void firehoseFailsAsExpectedLoop() {
         }
     }
 
-    @Test(timeout = 10000)
+    @Test
     public void onBackpressureDrop() {
         long t = System.currentTimeMillis();
         for (int i = 0; i < 100; i++) {
@@ -544,7 +515,7 @@ public void onBackpressureDrop() {
         }
     }
 
-    @Test(timeout = 20000)
+    @Test
     public void onBackpressureDropWithAction() {
         for (int i = 0; i < 100; i++) {
             final AtomicInteger emitCount = new AtomicInteger();
@@ -586,7 +557,7 @@ public void accept(Integer v) {
         }
     }
 
-    @Test(timeout = 10000)
+    @Test
     public void onBackpressureDropSynchronous() {
         for (int i = 0; i < 100; i++) {
             int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
@@ -608,7 +579,7 @@ public void onBackpressureDropSynchronous() {
         }
     }
 
-    @Test(timeout = 10000)
+    @Test
     public void onBackpressureDropSynchronousWithAction() {
         for (int i = 0; i < 100; i++) {
             final AtomicInteger dropCount = new AtomicInteger();
@@ -639,7 +610,7 @@ public void accept(Integer j) {
         }
     }
 
-    @Test(timeout = 2000)
+    @Test
     public void onBackpressureBuffer() {
         int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
         AtomicInteger c = new AtomicInteger();

src/test/java/io/reactivex/flowable/FlowableCollectTest.java

@@ -28,7 +28,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.testsupport.TestHelper;
 
-public final class FlowableCollectTest {
+public final class FlowableCollectTest extends RxJavaTest {
 
     @Test
     public void collectToListFlowable() {

src/test/java/io/reactivex/flowable/FlowableCombineLatestTests.java

@@ -15,17 +15,14 @@
  */
 package io.reactivex.flowable;
 
-import static io.reactivex.Flowable.combineLatest;
-import static org.junit.Assert.assertNull;
-
+import io.reactivex.RxJavaTest;
 import org.junit.*;
 
 import io.reactivex.Flowable;
 import io.reactivex.flowable.FlowableCovarianceTest.*;
 import io.reactivex.functions.*;
-import io.reactivex.processors.BehaviorProcessor;
 
-public class FlowableCombineLatestTests {
+public class FlowableCombineLatestTests extends RxJavaTest {
     /**
      * This won't compile if super/extends isn't done correctly on generics.
      */
@@ -63,25 +60,4 @@ public void accept(ExtendedResult t1) {
             System.out.println("Result: " + t1);
         }
     };
-
-    @Ignore("No longer allowed")
-    @Test
-    public void nullEmitting() throws Exception {
-        // FIXME this is no longer allowed
-        Flowable<Boolean> nullObservable = BehaviorProcessor.createDefault((Boolean) null);
-        Flowable<Boolean> nonNullObservable = BehaviorProcessor.createDefault(true);
-        Flowable<Boolean> combined =
-                combineLatest(nullObservable, nonNullObservable, new BiFunction<Boolean, Boolean, Boolean>() {
-                    @Override
-                    public Boolean apply(Boolean bool1, Boolean bool2) {
-                        return bool1 == null ? null : bool2;
-                    }
-                });
-        combined.subscribe(new Consumer<Boolean>() {
-            @Override
-            public void accept(Boolean aBoolean) {
-                assertNull(aBoolean);
-            }
-        });
-    }
 }

src/test/java/io/reactivex/flowable/FlowableConcatTests.java

@@ -16,13 +16,14 @@
 
 import java.util.*;
 
+import io.reactivex.RxJavaTest;
 import org.junit.Test;
 import org.reactivestreams.*;
 
 import io.reactivex.Flowable;
 import io.reactivex.flowable.FlowableCovarianceTest.*;
 
-public class FlowableConcatTests {
+public class FlowableConcatTests extends RxJavaTest {
 
     @Test
     public void concatSimple() {

src/test/java/io/reactivex/flowable/FlowableConversionTest.java

@@ -28,7 +28,7 @@
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subscribers.DefaultSubscriber;
 
-public class FlowableConversionTest {
+public class FlowableConversionTest extends RxJavaTest {
 
     public static class Cylon { }
 

src/test/java/io/reactivex/flowable/FlowableCovarianceTest.java

@@ -33,7 +33,7 @@
  *
  * See https://github.com/Netflix/RxJava/pull/331
  */
-public class FlowableCovarianceTest {
+public class FlowableCovarianceTest extends RxJavaTest {
 
     /**
      * This won't compile if super/extends isn't done correctly on generics.

src/test/java/io/reactivex/flowable/FlowableDoAfterNextTest.java

@@ -19,11 +19,12 @@
 
 import java.util.concurrent.atomic.AtomicInteger;
 
+import io.reactivex.RxJavaTest;
 import org.junit.Test;
 
 import io.reactivex.functions.Consumer;
 
-public class FlowableDoAfterNextTest {
+public class FlowableDoAfterNextTest extends RxJavaTest {
 
     @Test
     public void ifFunctionThrowsThatNoMoreEventsAreProcessed() {

src/test/java/io/reactivex/flowable/FlowableDoOnTest.java

@@ -23,7 +23,7 @@
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
 
-public class FlowableDoOnTest {
+public class FlowableDoOnTest extends RxJavaTest {
 
     @Test
     public void doOnEach() {

src/test/java/io/reactivex/flowable/FlowableErrorHandlingTests.java

@@ -18,14 +18,15 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.RxJavaTest;
 import org.junit.Test;
 import org.reactivestreams.Subscriber;
 
 import io.reactivex.Flowable;
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subscribers.DefaultSubscriber;
 
-public class FlowableErrorHandlingTests {
+public class FlowableErrorHandlingTests extends RxJavaTest {
 
     /**
      * Test that an error from a user provided Observer.onNext

src/test/java/io/reactivex/flowable/FlowableEventStreamTest.java

@@ -16,11 +16,12 @@
 
 package io.reactivex.flowable;
 
+import io.reactivex.RxJavaTest;
 import org.junit.Test;
 
 import io.reactivex.testsupport.TestHelper;
 
-public class FlowableEventStreamTest {
+public class FlowableEventStreamTest extends RxJavaTest {
     @Test
     public void constructorShouldBePrivate() {
         TestHelper.checkUtilityClass(FlowableEventStream.class);