2.x: BaseTestConsumer add assertValueAt(index, Predicate<T>)

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

@@ -312,26 +312,45 @@ public final U assertValue(T value) {
      */
     @SuppressWarnings("unchecked")
     public final U assertValue(Predicate<T> valuePredicate) {
+        assertValueAt(0, valuePredicate);
+
+        if (values.size() > 1) {
+            throw fail("Value present but other values as well");
+        }
+
+        return (U)this;
+    }
+
+    /**
+     * Asserts that this TestObserver/TestSubscriber received an onNext value at the given index
+     * for the provided predicate returns true.
+     * @param valuePredicate
+     *            the predicate that receives the onNext value
+     *            and should return true for the expected value.
+     * @return this
+     */
+    @SuppressWarnings("unchecked")
+    public final U assertValueAt(int index, Predicate<T> valuePredicate) {
         int s = values.size();
         if (s == 0) {
             throw fail("No values");
         }
 
+        if (index >= values.size()) {
+            throw fail("Invalid index: " + index);
+        }
+
         boolean found = false;
 
         try {
-            if (valuePredicate.test(values.get(0))) {
+            if (valuePredicate.test(values.get(index))) {
                 found = true;
             }
         } catch (Exception ex) {
             throw ExceptionHelper.wrapOrThrow(ex);
         }
 
-        if (found) {
-            if (s != 1) {
-                throw fail("Value present but other values as well");
-            }
-        } else {
+        if (!found) {
             throw fail("Value not present");
         }
         return (U)this;

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

@@ -1263,4 +1263,62 @@ public void assertValuePredicateMatchButMore() {
             }
         });
     }
+
+    @Test
+    public void assertValueAtPredicateEmpty() {
+        TestObserver<Object> ts = new TestObserver<Object>();
+
+        Observable.empty().subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No values");
+        ts.assertValueAt(0, new Predicate<Object>() {
+            @Override public boolean test(final Object o) throws Exception {
+                return false;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtPredicateMatch() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1, 2).subscribe(ts);
+
+        ts.assertValueAt(1, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 2;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtPredicateNoMatch() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1, 2, 3).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value not present");
+        ts.assertValueAt(2, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o != 3;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtInvalidIndex() {
+        TestObserver<Integer> ts = new TestObserver<Integer>();
+
+        Observable.just(1, 2).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Invalid index: 2 (latch = 0, values = 2, errors = 0, completions = 1)");
+        ts.assertValueAt(2, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
 }

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

@@ -1649,4 +1649,62 @@ public void assertValuePredicateMatchButMore() {
             }
         });
     }
+
+    @Test
+    public void assertValueAtPredicateEmpty() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+
+        Flowable.empty().subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No values");
+        ts.assertValueAt(0, new Predicate<Object>() {
+            @Override public boolean test(final Object o) throws Exception {
+                return false;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtPredicateMatch() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1, 2).subscribe(ts);
+
+        ts.assertValueAt(1, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 2;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtPredicateNoMatch() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1, 2, 3).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Value not present");
+        ts.assertValueAt(2, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o != 3;
+            }
+        });
+    }
+
+    @Test
+    public void assertValueAtInvalidIndex() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        Flowable.just(1, 2).subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Invalid index: 2 (latch = 0, values = 2, errors = 0, completions = 1)");
+        ts.assertValueAt(2, new Predicate<Integer>() {
+            @Override public boolean test(final Integer o) throws Exception {
+                return o == 1;
+            }
+        });
+    }
 }