Add assertValueAt(int, value) to TestObserver

.travis.yml

@@ -1,6 +1,6 @@
 language: java
 jdk:
-- oraclejdk7
+- oraclejdk8
 
 # force upgrade Java8 as per https://github.com/travis-ci/travis-ci/issues/4042 (fixes compilation issue)
 #addons:

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

@@ -17,6 +17,7 @@
 import java.util.concurrent.*;
 
 import io.reactivex.Notification;
+import io.reactivex.annotations.Experimental;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.CompositeException;
 import io.reactivex.functions.Predicate;
@@ -335,7 +336,7 @@ public final U assertValue(T value) {
 
     /**
      * Assert that this TestObserver/TestSubscriber did not receive an onNext value which is equal to
-     * the given value with respect to Objects.equals.
+     * the given value with respect to null-safe Object.equals.
      *
      * <p>History: 2.0.5 - experimental
      * @param value the value to expect not being received
@@ -401,6 +402,33 @@ public final U assertNever(Predicate<? super T> valuePredicate) {
         return (U)this;
     }
 
+    /**
+     * Asserts that this TestObserver/TestSubscriber received an onNext value at the given index
+     * which is equal to the given value with respect to null-safe Object.equals.
+     * @param index the position to assert on
+     * @param value the value to expect
+     * @return this
+     * @since 2.1.3 - experimental
+     */
+    @SuppressWarnings("unchecked")
+    @Experimental
+    public final U assertValueAt(int index, T value) {
+        int s = values.size();
+        if (s == 0) {
+            throw fail("No values");
+        }
+
+        if (index >= s) {
+            throw fail("Invalid index: " + index);
+        }
+
+        T v = values.get(index);
+        if (!ObjectHelper.equals(value, v)) {
+            throw fail("Expected: " + valueAndClass(value) + ", Actual: " + valueAndClass(v));
+        }
+        return (U)this;
+    }
+
     /**
      * Asserts that this TestObserver/TestSubscriber received an onNext value at the given index
      * for the provided predicate returns true.

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

@@ -1382,6 +1382,48 @@ public void assertValueAtInvalidIndex() {
         });
     }
 
+    @Test
+    public void assertValueAtIndexEmpty() {
+        TestObserver<Object> ts = new TestObserver<Object>();
+
+        Observable.empty().subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("No values");
+        ts.assertValueAt(0, "a");
+    }
+
+    @Test
+    public void assertValueAtIndexMatch() {
+        TestObserver<String> ts = new TestObserver<String>();
+
+        Observable.just("a", "b").subscribe(ts);
+
+        ts.assertValueAt(1, "b");
+    }
+
+    @Test
+    public void assertValueAtIndexNoMatch() {
+        TestObserver<String> ts = new TestObserver<String>();
+
+        Observable.just("a", "b", "c").subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Expected: b (class: String), Actual: c (class: String) (latch = 0, values = 3, errors = 0, completions = 1)");
+        ts.assertValueAt(2, "b");
+    }
+
+    @Test
+    public void assertValueAtIndexInvalidIndex() {
+        TestObserver<String> ts = new TestObserver<String>();
+
+        Observable.just("a", "b").subscribe(ts);
+
+        thrown.expect(AssertionError.class);
+        thrown.expectMessage("Invalid index: 2 (latch = 0, values = 2, errors = 0, completions = 1)");
+        ts.assertValueAt(2, "c");
+    }
+
     @Test
     public void withTag() {
         try {