2.x: TestSubscriber & TestObserver add assertValue(Predicate)

src/main/java/io/reactivex/observers/TestObserver.java

@@ -533,6 +533,40 @@ public final TestObserver<T> assertValue(T value) {
         return this;
     }
 
+    /**
+     * Asserts that this TestObserver received exactly one onNext value for which
+     * the provided predicate returns true.
+     * @param valuePredicate
+     *            the predicate that receives the onNext value
+     *            and should return true for the expected value.
+     * @return this
+     */
+    public final TestObserver<T> assertValue(Predicate<T> valuePredicate) {
+        int s = values.size();
+        if (s == 0) {
+            throw fail("No values");
+        }
+
+        boolean found = false;
+
+        try {
+            if (valuePredicate.test(values.get(0))) {
+                found = true;
+            }
+        } catch (Exception ex) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+
+        if (found) {
+            if (s != 1) {
+                throw fail("Value present but other values as well");
+            }
+        } else {
+            throw fail("Value not present");
+        }
+        return this;
+    }
+
     /** Appends the class name to a non-null value. */
     static String valueAndClass(Object o) {
         if (o != null) {

src/main/java/io/reactivex/subscribers/TestSubscriber.java

@@ -577,6 +577,40 @@ public final TestSubscriber<T> assertValue(T value) {
         return this;
     }
 
+    /**
+     * Asserts that this TestSubscriber received exactly one onNext value for which
+     * the provided predicate returns true.
+     * @param valuePredicate
+     *            the predicate that receives the onNext value
+     *            and should return true for the expected value.
+     * @return this
+     */
+    public final TestSubscriber<T> assertValue(Predicate<T> valuePredicate) {
+        int s = values.size();
+        if (s == 0) {
+            throw fail("No values");
+        }
+
+        boolean found = false;
+
+        try {
+            if (valuePredicate.test(values.get(0))) {
+                found = true;
+            }
+        } catch (Exception ex) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+
+        if (found) {
+            if (s != 1) {
+                throw fail("Value present but other values as well");
+            }
+        } else {
+            throw fail("Value not present");
+        }
+        return this;
+    }
+
     /** Appends the class name to a non-null value. */
     static String valueAndClass(Object o) {
         if (o != null) {

src/test/java/io/reactivex/observers/TestObserverTest.java

@@ -178,7 +178,7 @@ public void testNullExpected() {
         to.onNext(1);
 
         try {
-            to.assertValue(null);
+            to.assertValue((Integer) null);
         } catch (AssertionError ex) {
             // this is expected
             return;
@@ -1205,4 +1205,62 @@ public void asyncFusion() {
         .assertFusionMode(QueueDisposable.ASYNC)
         .assertResult(1);
     }
-}
+
+    @Test
+    public void assertValuePredicateEmpty() {
+        TestObserver<Object> ts = new TestObserver<Object>();
+
+        Observable.empty().subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No values");
+        ts.assertValue(new Predicate<Object>() {
+            @Override public boolean test(final Object o) throws Exception {
+                return false;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateMatch() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1).subscribe(ts);
+
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateNoMatch() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value not present");
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o != 1;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateMatchButMore() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1, 2).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value present but other values as well");
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
+}

src/test/java/io/reactivex/subscribers/TestSubscriberTest.java

@@ -1592,4 +1592,61 @@ public void asyncQueueThrows() {
         .assertFailure(TestException.class);
     }
 
-}
+    @Test
+    public void assertValuePredicateEmpty() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+
+        Flowable.empty().subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No values");
+        ts.assertValue(new Predicate<Object>() {
+            @Override public boolean test(final Object o) throws Exception {
+                return false;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateMatch() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1).subscribe(ts);
+
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateNoMatch() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value not present");
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o != 1;
+            }
+        });
+    }
+
+    @Test
+    public void assertValuePredicateMatchButMore() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1, 2).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value present but other values as well");
+        ts.assertValue(new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
+}