fix tests

Author: davidmoten

src/main/java/rx/exceptions/OnCompletedFailedException.java

@@ -23,4 +23,7 @@ public OnCompletedFailedException(Throwable throwable) {
         super(throwable);
     }
 
+    public OnCompletedFailedException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
 }

src/main/java/rx/exceptions/UnsubscribeFailedException.java

@@ -23,4 +23,8 @@ public UnsubscribeFailedException(Throwable throwable) {
         super(throwable);
     }
 
+    public UnsubscribeFailedException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
+    
 }

src/main/java/rx/observers/SafeSubscriber.java

@@ -86,15 +86,15 @@ public void onCompleted() {
                 // which can prevent it from being able to handle StackOverflow
                 Exceptions.throwIfFatal(e);
                 RxJavaPluginUtils.handleException(e);
-                throw new OnCompletedFailedException(e);
+                throw new OnCompletedFailedException(e.getMessage(), e);
             } finally {
                 try {
                     // Similarly to onError if failure occurs in unsubscribe then Rx contract is broken
                     // and we throw an UnsubscribeFailureException.
                     unsubscribe();
                 } catch (Throwable e2) {
                     RxJavaPluginUtils.handleException(e2);
-                    throw new UnsubscribeFailedException(e2);
+                    throw new UnsubscribeFailedException(e2.getMessage(), e2);
                 }
             }
         }

src/test/java/rx/observers/SafeObserverTest.java

@@ -22,6 +22,7 @@
 
 import org.junit.Test;
 
+import junit.framework.Assert;
 import rx.Subscriber;
 import rx.exceptions.*;
 import rx.functions.Action0;
@@ -68,19 +69,6 @@ public void onCompletedFailure() {
         }
     }
 
-    @Test
-    public void onCompletedFailureSafe() {
-        AtomicReference<Throwable> onError = new AtomicReference<Throwable>();
-        try {
-            new SafeSubscriber<String>(OBSERVER_ONCOMPLETED_FAIL(onError)).onCompleted();
-            assertNotNull(onError.get());
-            assertTrue(onError.get() instanceof SafeObserverTestException);
-            assertEquals("onCompletedFail", onError.get().getMessage());
-        } catch (Exception e) {
-            fail("expects exception to be passed to onError");
-        }
-    }
-
     @Test
     public void onErrorFailure() {
         try {
@@ -184,8 +172,8 @@ public void call() {
             e.printStackTrace();
 
             assertTrue(o.isUnsubscribed());
-
-            assertTrue(e instanceof SafeObserverTestException);
+            assertTrue(e instanceof UnsubscribeFailedException);
+            assertTrue(e.getCause() instanceof SafeObserverTestException);
             assertEquals("failure from unsubscribe", e.getMessage());
             // expected since onError fails so SafeObserver can't help
         }
@@ -475,9 +463,12 @@ public void onCompleted() {
             }
         });
 
-        s.onCompleted();
-        
-        assertTrue("Error not received", error.get() instanceof TestException);
+        try {
+            s.onCompleted();
+            Assert.fail();
+        } catch (OnCompletedFailedException e) {
+           assertNull(error.get());
+        }
     }
 
     @Test

src/test/java/rx/observers/SafeSubscriberTest.java

@@ -26,9 +26,11 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import rx.exceptions.OnCompletedFailedException;
 import rx.exceptions.OnErrorFailedException;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.exceptions.TestException;
+import rx.exceptions.UnsubscribeFailedException;
 import rx.functions.Action0;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaPlugins;
@@ -59,10 +61,12 @@ public void onCompleted() {
             }
         };
         SafeSubscriber<Integer> safe = new SafeSubscriber<Integer>(ts);
-        
+        try {
         safe.onCompleted();
-        
-        assertTrue(safe.isUnsubscribed());
+        Assert.fail();
+        } catch (OnCompletedFailedException e) {
+            assertTrue(safe.isUnsubscribed());
+        }
     }
 
     @Test
@@ -84,7 +88,7 @@ public void onCompleted() {
         assertTrue(safe.isUnsubscribed());
     }
 
-    @Test
+    @Test(expected=OnCompletedFailedException.class)
     public void testPluginException() {
         RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
             @Override
@@ -261,8 +265,12 @@ public void call() {
             }
         }));
 
-        safe.onCompleted();
-        assertEquals(1, (int) calls.get());
-        assertEquals(0, (int) errors.get());
+        try {
+            safe.onCompleted();
+            Assert.fail();
+        } catch(UnsubscribeFailedException e) {
+            assertEquals(1, (int) calls.get());
+            assertEquals(0, (int) errors.get());
+        }
     }
 }