2.x: Common test classes for Processors. (#4882)

Author: artem-zinnatullin

src/test/java/io/reactivex/processors/AsyncProcessorTest.java

@@ -13,29 +13,37 @@
 
 package io.reactivex.processors;
 
-import static org.junit.Assert.*;
-import static org.mockito.ArgumentMatchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.junit.*;
-import org.mockito.*;
-import org.reactivestreams.Subscriber;
-
 import io.reactivex.TestHelper;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.Consumer;
 import io.reactivex.internal.fuseable.QueueSubscription;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subscribers.*;
+import io.reactivex.subscribers.SubscriberFusion;
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+import org.reactivestreams.Subscriber;
 
-public class AsyncProcessorTest {
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.*;
+
+public class AsyncProcessorTest extends DelayedFlowableProcessorTest<Object> {
 
     private final Throwable testException = new Throwable();
 
+    @Override
+    protected FlowableProcessor<Object> create() {
+        return AsyncProcessor.create();
+    }
+
     @Test
     public void testNeverCompleted() {
         AsyncProcessor<String> subject = AsyncProcessor.create();
@@ -422,58 +430,6 @@ public void fusionOfflie() {
         .assertResult(1);
     }
 
-    @Test
-    public void onNextNull() {
-        final AsyncProcessor<Object> p = AsyncProcessor.create();
-
-        p.onNext(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onErrorNull() {
-        final AsyncProcessor<Object> p = AsyncProcessor.create();
-
-        p.onError(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onNextNullDelayed() {
-        final AsyncProcessor<Object> p = AsyncProcessor.create();
-
-        TestSubscriber<Object> ts = p.test();
-
-        p.onNext(null);
-
-        ts
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onErrorNullDelayed() {
-        final AsyncProcessor<Object> p = AsyncProcessor.create();
-
-        TestSubscriber<Object> ts = p.test();
-
-        p.onError(null);
-
-        ts
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
     @Test
     public void onSubscribeAfterDone() {
         AsyncProcessor<Object> p = AsyncProcessor.create();

src/test/java/io/reactivex/processors/BehaviorProcessorTest.java

@@ -13,30 +13,41 @@
 
 package io.reactivex.processors;
 
-import static org.junit.Assert.*;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
-
-import java.util.List;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.junit.*;
-import org.mockito.*;
-import org.reactivestreams.Subscriber;
-
-import io.reactivex.*;
-import io.reactivex.exceptions.*;
+import io.reactivex.Flowable;
+import io.reactivex.Scheduler;
+import io.reactivex.TestHelper;
+import io.reactivex.exceptions.MissingBackpressureException;
+import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subscribers.*;
+import io.reactivex.subscribers.DefaultSubscriber;
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+import org.reactivestreams.Subscriber;
 
-public class BehaviorProcessorTest {
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+public class BehaviorProcessorTest extends DelayedFlowableProcessorTest<Object> {
 
     private final Throwable testException = new Throwable();
 
+    @Override
+    protected FlowableProcessor<Object> create() {
+        return BehaviorProcessor.create();
+    }
+
     @Test
     public void testThatSubscriberReceivesDefaultValueAndSubsequentEvents() {
         BehaviorProcessor<String> subject = BehaviorProcessor.createDefault("default");
@@ -563,66 +574,6 @@ public void testCurrentStateMethodsError() {
         assertTrue(as.getThrowable() instanceof TestException);
     }
 
-    @Test
-    public void onNextNull() {
-        final BehaviorProcessor<Object> p = BehaviorProcessor.create();
-
-        p.onNext(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onErrorNull() {
-        final BehaviorProcessor<Object> p = BehaviorProcessor.create();
-
-        p.onError(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onNextNullDelayed() {
-        final BehaviorProcessor<Object> p = BehaviorProcessor.create();
-
-        TestSubscriber<Object> ts = p.test();
-
-        assertTrue(p.hasSubscribers());
-
-        p.onNext(null);
-
-        assertFalse(p.hasSubscribers());
-
-        ts
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
-    @Test
-    public void onErrorNullDelayed() {
-        final BehaviorProcessor<Object> p = BehaviorProcessor.create();
-
-        TestSubscriber<Object> ts = p.test();
-
-        assertTrue(p.hasSubscribers());
-
-        p.onError(null);
-
-        assertFalse(p.hasSubscribers());
-
-        ts
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
-
     @Test
     public void cancelOnArrival() {
         BehaviorProcessor<Object> p = BehaviorProcessor.create();

src/test/java/io/reactivex/processors/DelayedFlowableProcessorTest.java

@@ -0,0 +1,40 @@
+package io.reactivex.processors;
+
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+
+public abstract class DelayedFlowableProcessorTest<T> extends FlowableProcessorTest<T> {
+
+    @Test
+    public void onNextNullDelayed() {
+        final FlowableProcessor<T> p = create();
+
+        TestSubscriber<T> ts = p.test();
+
+        p.onNext(null);
+
+
+
+        ts
+                .assertNoValues()
+                .assertError(NullPointerException.class)
+                .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
+    }
+
+    @Test
+    public void onErrorNullDelayed() {
+        final FlowableProcessor<T> p = create();
+
+        TestSubscriber<T> ts = p.test();
+
+        p.onError(null);
+        assertFalse(p.hasSubscribers());
+
+        ts
+                .assertNoValues()
+                .assertError(NullPointerException.class)
+                .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+    }
+}

src/test/java/io/reactivex/processors/FlowableProcessorTest.java

@@ -0,0 +1,32 @@
+package io.reactivex.processors;
+
+import org.junit.Test;
+
+public abstract class FlowableProcessorTest<T> {
+
+    protected abstract FlowableProcessor<T> create();
+
+    @Test
+    public void onNextNull() {
+        final FlowableProcessor<T> p = create();
+
+        p.onNext(null);
+
+        p.test()
+                .assertNoValues()
+                .assertError(NullPointerException.class)
+                .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
+    }
+
+    @Test
+    public void onErrorNull() {
+        final FlowableProcessor<T> p = create();
+
+        p.onError(null);
+
+        p.test()
+                .assertNoValues()
+                .assertError(NullPointerException.class)
+                .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
+    }
+}

src/test/java/io/reactivex/processors/PublishProcessorTest.java

@@ -13,24 +13,34 @@
 
 package io.reactivex.processors;
 
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
+import io.reactivex.Flowable;
+import io.reactivex.TestHelper;
+import io.reactivex.exceptions.MissingBackpressureException;
+import io.reactivex.exceptions.TestException;
+import io.reactivex.functions.Consumer;
+import io.reactivex.functions.Function;
+import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subscribers.DefaultSubscriber;
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
 import java.util.ArrayList;
-import java.util.concurrent.*;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.Test;
-import org.mockito.*;
-import org.reactivestreams.*;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
-import io.reactivex.*;
-import io.reactivex.exceptions.*;
-import io.reactivex.functions.*;
-import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subscribers.*;
+public class PublishProcessorTest extends FlowableProcessorTest<Object> {
 
-public class PublishProcessorTest {
+    @Override
+    protected FlowableProcessor<Object> create() {
+        return PublishProcessor.create();
+    }
 
     @Test
     public void testCompleted() {
@@ -615,27 +625,5 @@ public void run() {
         }
     }
 
-    @Test
-    public void onNextNull() {
-        final PublishProcessor<Object> p = PublishProcessor.create();
-
-        p.onNext(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onNext called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
 
-    @Test
-    public void onErrorNull() {
-        final PublishProcessor<Object> p = PublishProcessor.create();
-
-        p.onError(null);
-
-        p.test()
-            .assertNoValues()
-            .assertError(NullPointerException.class)
-            .assertErrorMessage("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
-    }
 }