diff --git a/src/main/java/io/reactivex/observers/TestObserver.java b/src/main/java/io/reactivex/observers/TestObserver.java
index 0a599863bb..1f615dca3d 100644
--- a/src/main/java/io/reactivex/observers/TestObserver.java
+++ b/src/main/java/io/reactivex/observers/TestObserver.java
@@ -21,6 +21,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.CompositeException;
 import io.reactivex.functions.Consumer;
+import io.reactivex.functions.Predicate;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.fuseable.QueueDisposable;
@@ -424,10 +425,12 @@ public final TestObserver<T> assertNoErrors() {
      *
      * <p>The comparison is performed via Objects.equals(); since most exceptions don't
      * implement equals(), this assertion may fail. Use the {@link #assertError(Class)}
-     * overload to test against the class of an error instead of an instance of an error.
+     * overload to test against the class of an error instead of an instance of an error
+     * or {@link #assertError(Predicate)} to test with different condition.
      * @param error the error to check
      * @return this;
      * @see #assertError(Class)
+     * @see #assertError(Predicate)
      */
     public final TestObserver<T> assertError(Throwable error) {
         int s = errors.size();
@@ -475,6 +478,43 @@ public final TestObserver<T> assertError(Class<? extends Throwable> errorClass)
         return this;
     }
 
+    /**
+     * Asserts that this TestObserver received exactly one onError event for which
+     * the provided predicate returns true.
+     * @param errorPredicate
+     *            the predicate that receives the error Throwable
+     *            and should return true for expected errors.
+     * @return this
+     */
+    public final TestObserver<T> assertError(Predicate<Throwable> errorPredicate) {
+        int s = errors.size();
+        if (s == 0) {
+            throw fail("No errors");
+        }
+
+        boolean found = false;
+
+        for (Throwable e : errors) {
+            try {
+                if (errorPredicate.test(e)) {
+                    found = true;
+                    break;
+                }
+            } catch (Exception ex) {
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+
+        if (found) {
+            if (s != 1) {
+                throw fail("Error present but other errors as well");
+            }
+        } else {
+            throw fail("Error not present");
+        }
+        return this;
+    }
+
     /**
      * Assert that this TestObserver received exactly one onNext value which is equal to
      * the given value with respect to Objects.equals.
@@ -825,6 +865,7 @@ public final TestObserver<T> assertOf(Consumer<? super TestObserver<T>> check) {
      * @param values the expected values, asserted in order
      * @return this
      * @see #assertFailure(Class, Object...)
+     * @see #assertFailure(Predicate, Object...)
      * @see #assertFailureAndMessage(Class, String, Object...)
      */
     public final TestObserver<T> assertResult(T... values) {
@@ -848,6 +889,22 @@ public final TestObserver<T> assertFailure(Class<? extends Throwable> error, T..
                 .assertNotComplete();
     }
 
+    /**
+     * Assert that the upstream signalled the specified values in order and then failed
+     * with a Throwable for which the provided predicate returns true.
+     * @param errorPredicate
+     *            the predicate that receives the error Throwable
+     *            and should return true for expected errors.
+     * @param values the expected values, asserted in order
+     * @return this
+     */
+    public final TestObserver<T> assertFailure(Predicate<Throwable> errorPredicate, T... values) {
+        return assertSubscribed()
+                .assertValues(values)
+                .assertError(errorPredicate)
+                .assertNotComplete();
+    }
+
     /**
      * Assert that the upstream signalled the specified values in order,
      * then failed with a specific class or subclass of Throwable
diff --git a/src/main/java/io/reactivex/subscribers/TestSubscriber.java b/src/main/java/io/reactivex/subscribers/TestSubscriber.java
index 1b2832634e..670e2d9ad7 100644
--- a/src/main/java/io/reactivex/subscribers/TestSubscriber.java
+++ b/src/main/java/io/reactivex/subscribers/TestSubscriber.java
@@ -22,6 +22,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.CompositeException;
 import io.reactivex.functions.Consumer;
+import io.reactivex.functions.Predicate;
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.fuseable.QueueSubscription;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
@@ -468,10 +469,12 @@ public final TestSubscriber<T> assertNoErrors() {
      *
      * <p>The comparison is performed via Objects.equals(); since most exceptions don't
      * implement equals(), this assertion may fail. Use the {@link #assertError(Class)}
-     * overload to test against the class of an error instead of an instance of an error.
+     * overload to test against the class of an error instead of an instance of an error
+     * or {@link #assertError(Predicate)} to test with different condition.
      * @param error the error to check
      * @return this
      * @see #assertError(Class)
+     * @see #assertError(Predicate)
      */
     public final TestSubscriber<T> assertError(Throwable error) {
         int s = errors.size();
@@ -519,6 +522,43 @@ public final TestSubscriber<T> assertError(Class<? extends Throwable> errorClass
         return this;
     }
 
+    /**
+     * Asserts that this TestSubscriber received exactly one onError event for which
+     * the provided predicate returns true.
+     * @param errorPredicate
+     *            the predicate that receives the error Throwable
+     *            and should return true for expected errors.
+     * @return this
+     */
+    public final TestSubscriber<T> assertError(Predicate<Throwable> errorPredicate) {
+        int s = errors.size();
+        if (s == 0) {
+            throw fail("No errors");
+        }
+
+        boolean found = false;
+
+        for (Throwable e : errors) {
+            try {
+                if (errorPredicate.test(e)) {
+                    found = true;
+                    break;
+                }
+            } catch (Exception ex) {
+                throw ExceptionHelper.wrapOrThrow(ex);
+            }
+        }
+
+        if (found) {
+            if (s != 1) {
+                throw fail("Error present but other errors as well");
+            }
+        } else {
+            throw fail("Error not present");
+        }
+        return this;
+    }
+
     /**
      * Assert that this TestSubscriber received exactly one onNext value which is equal to
      * the given value with respect to Objects.equals.
@@ -869,6 +909,7 @@ public final TestSubscriber<T> assertOf(Consumer<? super TestSubscriber<T>> chec
      * @param values the expected values, asserted in order
      * @return this
      * @see #assertFailure(Class, Object...)
+     * @see #assertFailure(Predicate, Object...)
      * @see #assertFailureAndMessage(Class, String, Object...)
      */
     public final TestSubscriber<T> assertResult(T... values) {
@@ -892,6 +933,22 @@ public final TestSubscriber<T> assertFailure(Class<? extends Throwable> error, T
                 .assertNotComplete();
     }
 
+    /**
+     * Assert that the upstream signalled the specified values in order and then failed
+     * with a Throwable for which the provided predicate returns true.
+     * @param errorPredicate
+     *            the predicate that receives the error Throwable
+     *            and should return true for expected errors.
+     * @param values the expected values, asserted in order
+     * @return this
+     */
+    public final TestSubscriber<T> assertFailure(Predicate<Throwable> errorPredicate, T... values) {
+        return assertSubscribed()
+                .assertValues(values)
+                .assertError(errorPredicate)
+                .assertNotComplete();
+    }
+
     /**
      * Assert that the upstream signalled the specified values in order,
      * then failed with a specific class or subclass of Throwable
diff --git a/src/test/java/io/reactivex/observers/TestObserverTest.java b/src/test/java/io/reactivex/observers/TestObserverTest.java
index 6c1a85c288..5f32528cff 100644
--- a/src/test/java/io/reactivex/observers/TestObserverTest.java
+++ b/src/test/java/io/reactivex/observers/TestObserverTest.java
@@ -31,6 +31,7 @@
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
+import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.fuseable.QueueDisposable;
 import io.reactivex.internal.operators.observable.ObservableScalarXMap.ScalarDisposable;
 import io.reactivex.internal.subscriptions.EmptySubscription;
@@ -332,6 +333,13 @@ public void assertError() {
             // expected
         }
 
+        try {
+            ts.assertError(Functions.<Throwable>alwaysTrue());
+            throw new RuntimeException("Should have thrown");
+        } catch (AssertionError ex) {
+            // expected
+        }
+
         try {
             ts.assertErrorMessage("");
             throw new RuntimeException("Should have thrown");
@@ -367,6 +375,15 @@ public void assertError() {
 
         ts.assertError(TestException.class);
 
+        ts.assertError(Functions.<Throwable>alwaysTrue());
+
+        ts.assertError(new Predicate<Throwable>() {
+            @Override
+            public boolean test(Throwable t) throws Exception {
+                return t.getMessage() != null && t.getMessage().contains("Forced");
+            }
+        });
+
         ts.assertErrorMessage("Forced failure");
 
         try {
@@ -390,6 +407,13 @@ public void assertError() {
             // expected
         }
 
+        try {
+            ts.assertError(Functions.<Throwable>alwaysFalse());
+            throw new RuntimeException("Should have thrown");
+        } catch (AssertionError exc) {
+            // expected
+        }
+
         try {
             ts.assertNoErrors();
             throw new RuntimeException("Should have thrown");
@@ -428,12 +452,16 @@ public void assertFailure() {
 
         ts.assertFailure(TestException.class);
 
+        ts.assertFailure(Functions.<Throwable>alwaysTrue());
+
         ts.assertFailureAndMessage(TestException.class, "Forced failure");
 
         ts.onNext(1);
 
         ts.assertFailure(TestException.class, 1);
 
+        ts.assertFailure(Functions.<Throwable>alwaysTrue(), 1);
+
         ts.assertFailureAndMessage(TestException.class, "Forced failure", 1);
     }
 
@@ -965,6 +993,12 @@ public void assertErrorMultiple() {
         } catch (AssertionError ex) {
             // expected
         }
+        try {
+            ts.assertError(Functions.<Throwable>alwaysTrue());
+            throw new RuntimeException("Should have thrown!");
+        } catch (AssertionError ex) {
+            // expected
+        }
         try {
             ts.assertErrorMessage("");
             throw new RuntimeException("Should have thrown!");
@@ -973,6 +1007,24 @@ public void assertErrorMultiple() {
         }
     }
 
+    @Test
+    public void testErrorInPredicate() {
+        TestObserver<Object> ts = new TestObserver<Object>();
+        ts.onError(new RuntimeException());
+        try {
+            ts.assertError(new Predicate<Throwable>() {
+                @Override
+                public boolean test(Throwable throwable) throws Exception {
+                    throw new TestException();
+                }
+            });
+        } catch (TestException ex) {
+            // expected
+            return;
+        }
+        fail("Error in predicate but not thrown!");
+    }
+
     @Test
     public void assertComplete() {
         TestObserver<Integer> ts = new TestObserver<Integer>();
diff --git a/src/test/java/io/reactivex/subscribers/TestSubscriberTest.java b/src/test/java/io/reactivex/subscribers/TestSubscriberTest.java
index 70886c3f97..75591fd95c 100644
--- a/src/test/java/io/reactivex/subscribers/TestSubscriberTest.java
+++ b/src/test/java/io/reactivex/subscribers/TestSubscriberTest.java
@@ -30,6 +30,7 @@
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
+import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.fuseable.QueueSubscription;
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.processors.*;
@@ -346,6 +347,27 @@ public void testMultipleErrors3() {
         fail("Multiple Error present but no assertion error!");
     }
 
+    @Test
+    public void testMultipleErrors4() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.onSubscribe(EmptySubscription.INSTANCE);
+        ts.onError(new TestException());
+        ts.onError(new TestException());
+        try {
+            ts.assertError(Functions.<Throwable>alwaysTrue());
+        } catch (AssertionError ex) {
+            Throwable e = ex.getCause();
+            if (!(e instanceof CompositeException)) {
+                fail("Multiple Error present but the reported error doesn't have a composite cause!");
+            }
+            CompositeException ce = (CompositeException)e;
+            assertEquals(2, ce.size());
+            // expected
+            return;
+        }
+        fail("Multiple Error present but no assertion error!");
+    }
+
     @Test
     public void testDifferentError() {
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
@@ -378,6 +400,20 @@ public void testDifferentError3() {
         ts.onError(new RuntimeException());
         try {
             ts.assertError(TestException.class);
+        }
+        catch (AssertionError ex) {
+            // expected
+            return;
+        }
+        fail("Different Error present but no assertion error!");
+    }
+
+    @Test
+    public void testDifferentError4() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.onError(new RuntimeException());
+        try {
+            ts.assertError(Functions.<Throwable>alwaysFalse());
         } catch (AssertionError ex) {
             // expected
             return;
@@ -385,6 +421,24 @@ public void testDifferentError3() {
         fail("Different Error present but no assertion error!");
     }
 
+    @Test
+    public void testErrorInPredicate() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.onError(new RuntimeException());
+        try {
+            ts.assertError(new Predicate<Throwable>() {
+                @Override
+                public boolean test(Throwable throwable) throws Exception {
+                    throw new TestException();
+                }
+            });
+        } catch (TestException ex) {
+            // expected
+            return;
+        }
+        fail("Error in predicate but not thrown!");
+    }
+
     @Test
     public void testNoError() {
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
@@ -409,6 +463,18 @@ public void testNoError2() {
         fail("No present but no assertion error!");
     }
 
+    @Test
+    public void testNoError3() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        try {
+            ts.assertError(Functions.<Throwable>alwaysTrue());
+        } catch (AssertionError ex) {
+            // expected
+            return;
+        }
+        fail("No present but no assertion error!");
+    }
+
     @Test
     public void testInterruptTerminalEventAwait() {
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
@@ -717,6 +783,13 @@ public void assertError() {
 
         }
 
+        try {
+            ts.assertError(Functions.<Throwable>alwaysTrue());
+            throw new RuntimeException("Should have thrown");
+        } catch (AssertionError ex) {
+            // expected
+        }
+
         try {
             ts.assertSubscribed();
             throw new RuntimeException("Should have thrown");
@@ -747,6 +820,15 @@ public void assertError() {
 
         ts.assertErrorMessage("Forced failure");
 
+        ts.assertError(Functions.<Throwable>alwaysTrue());
+
+        ts.assertError(new Predicate<Throwable>() {
+            @Override
+            public boolean test(Throwable t) {
+                return t.getMessage() != null && t.getMessage().contains("Forced");
+            }
+        });
+
         try {
             ts.assertErrorMessage("");
             throw new RuntimeException("Should have thrown");
@@ -775,6 +857,13 @@ public void assertError() {
             // expected
         }
 
+        try {
+            ts.assertError(Functions.<Throwable>alwaysFalse());
+            throw new RuntimeException("Should have thrown");
+        } catch (AssertionError exc) {
+            // expected
+        }
+
         ts.assertTerminated();
 
         ts.assertValueCount(0);
@@ -806,12 +895,16 @@ public void assertFailure() {
 
         ts.assertFailure(TestException.class);
 
+        ts.assertFailure(Functions.<Throwable>alwaysTrue());
+
         ts.assertFailureAndMessage(TestException.class, "Forced failure");
 
         ts.onNext(1);
 
         ts.assertFailure(TestException.class, 1);
 
+        ts.assertFailure(Functions.<Throwable>alwaysTrue(), 1);
+
         ts.assertFailureAndMessage(TestException.class, "Forced failure", 1);
     }