Junit observer fix

src/test/java/org/numenta/nupic/network/AlgorithmDeterminacyTest.java

@@ -60,7 +60,7 @@
  * 3 modes:</p><p>
  * <ol>
  *  <li>Straight through the TM algorithm class</li>
- *  <li>Using a Layer in with synchronous calls</li>
+ *  <li>Using a Layer with synchronous calls</li>
  *  <li>Using the full NAPI and starting the Layer's thread</li>
  * </ol>
  * <p>

src/test/java/org/numenta/nupic/network/NetworkTest.java

@@ -59,9 +59,10 @@
 
 import rx.Observer;
 import rx.Subscriber;
+import rx.observers.TestObserver;
 
 
-public class NetworkTest {
+public class NetworkTest extends ObservableTestBase {
     private int[][] dayMap = new int[][] { 
         new int[] { 1, 1, 0, 0, 0, 0, 0, 1 }, // Sunday
         new int[] { 1, 1, 1, 0, 0, 0, 0, 0 }, // Monday
@@ -730,9 +731,9 @@ public void testNetworkComputeWithNoSensor() {
         }
 
         // Test that we get proper output after prediction stabilization
-        r1.observe().subscribe(new Subscriber<Inference>() {
+        TestObserver<Inference> tester;
+        r1.observe().subscribe(tester = new TestObserver<Inference>() {
             @Override public void onCompleted() {}
-            @Override public void onError(Throwable e) { e.printStackTrace(); }
             @Override public void onNext(Inference i) {
                 int nextDay = ((int)Math.rint(((Number)i.getClassification("dayOfWeek").getMostProbableValue(1)).doubleValue()));
                 assertEquals(6, nextDay);
@@ -741,6 +742,9 @@ public void testNetworkComputeWithNoSensor() {
 
         multiInput.put("dayOfWeek", 5.0);
         n.compute(multiInput);
+
+        // Check for exception during the TestObserver's onNext() execution.
+        checkObserver(tester);
     }
 
     @Test
@@ -961,13 +965,17 @@ public void testObservableWithCoordinateEncoder_NEGATIVE() {
                     .add(new SpatialPooler())
                     .add(htmSensor)));
 
-        network.observe().subscribe(new Observer<Inference>() {
+        TestObserver<Inference> tester;
+        network.observe().subscribe(tester = new TestObserver<Inference>() {
             @Override public void onCompleted() {
                 //Should never happen here.
                 assertEquals(0, anomaly, 0);
                 completed = true;
+
+                super.onCompleted();
             }
             @Override public void onError(Throwable e) { 
+                super.onError(e);
                 errorMessage = e.getMessage();
                 network.halt();
             }
@@ -995,6 +1003,8 @@ public void testObservableWithCoordinateEncoder_NEGATIVE() {
         assertFalse(completed);
         assertEquals("Cannot autoclassify with raw array input or  " +
             "Coordinate based encoders... Remove auto classify setting.", errorMessage);
+
+        assertTrue(hasErrors(tester));
     }
 
     ///////////////////////////////////////////////////////////////////////////////////

src/test/java/org/numenta/nupic/network/ObservableTestBase.java

@@ -0,0 +1,28 @@
+package org.numenta.nupic.network;
+
+import rx.observers.TestObserver;
+
+/**
+ * Base class for tests which use the Observable framework to test
+ * their methods. Contains useful methods to 
+ * @author cogmission
+ *
+ */
+public class ObservableTestBase {
+
+    protected <T> void checkObserver(TestObserver<T> obs) {
+        if(obs.getOnErrorEvents().size() > 0) {
+            Throwable e = (Throwable) obs.getOnErrorEvents().get(0);
+            e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+    }
+
+    protected <T> boolean hasErrors(TestObserver<T> obs) {
+        return !obs.getOnErrorEvents().isEmpty();
+    }
+
+    protected <T> boolean hasCompletions(TestObserver<T> obs) {
+        return !obs.getOnCompletedEvents().isEmpty();
+    }
+}

src/test/java/org/numenta/nupic/network/ObservableTestBaseTest.java

@@ -0,0 +1,194 @@
+package org.numenta.nupic.network;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Subscriber;
+import rx.observers.TestObserver;
+import rx.subjects.PublishSubject;
+
+/**
+ * Indicates the difference in {@link AssertionFailure} handling between
+ * {@link Observable}s which are directly subscribed to, and {@code Observable}s
+ * which are indirectly subscribed to.
+ * 
+ * HTM.Java's Network API uses an "indirect" method of emitting events allowing
+ * more flexibility in management of event dispatch however it causes a problem
+ * within JUnit tests where assertion failures don't actually get dispatched to JUnit
+ * so that the test will reliably fail if the conditions of the test aren't met.
+ * 
+ * This test class illustrates the differences and provides an example of a solution
+ * which allows exceptions to be gathered at the site of assertion failure and later
+ * propagated to JUnit so that it can fail the test if the test actually has a 
+ * failure condition.
+ * 
+ * @author cogmission
+ * @see ObservableTestBase
+ */
+public class ObservableTestBaseTest extends ObservableTestBase {
+
+    /**
+     * This test shows that failures arising in {@link Observer#onNext(Object)} can 
+     * indeed be caught by JUnit - providing the propagating infrastructure reports
+     * the failure directly to the subscribed {@link Observable}. 
+     * 
+     * HTM.Java's {@link Layer} class however, has an internal {@link PublishSubject}
+     * which dispatches all Layer computations and is a level of indirection between
+     * the Observable originally subscribed to, and the Observable (PublishSubject)
+     * used to dispatch (emit) the Layer events.
+     */
+    @Test
+    public void testDirectObservableSubscriberCanCatchFailedAssertions() {
+        Observable<Inference> observable = Observable.create(new Observable.OnSubscribe<Inference>() {
+            @Override public void call(Subscriber<? super Inference> subscriber) {
+                ManualInput inf = new ManualInput();
+                inf.anomalyScore(1.0);
+
+                subscriber.onNext(inf);
+            }
+        });
+
+        Observer<Inference> observer = new Observer<Inference>() {
+            @Override public void onCompleted() {}
+
+            // Here fail works and forces JUnit to pick up the failure
+            @Override public void onError(Throwable e) { fail(); }  // <--- PROPAGATE FAILURE TO JUnit
+
+            @Override public void onNext(Inference inf) {
+                // The inference's anomaly score = 1.0 so this should force failure.
+                assertTrue(inf.getAnomalyScore() == 0.0); 
+            }
+        };
+
+        try {
+            observable.subscribe(observer);
+
+            // This test would fail if this point was reached
+            // (but it isn't reached because the above failed assertion is correctly
+            // caught by JUnit - as it should be). Instead the above throws an 
+            // exception.
+            fail();   
+
+        }catch(Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * Demonstrates that the indirection of the internal PublishSubject within
+     * the Network's Layer class is what causes failed assertions to not be
+     * recognized by JUnit (i.e. they get swallowed). 
+     * 
+     * This test passes even though there are errors.
+     */
+    @Test
+    public void testCheckObservable_Incorrectly_Passes() {
+        FauxNetwork network = new FauxNetwork();
+
+        TestObserver<Inference> observer = new TestObserver<Inference>() {
+            @Override
+            public void onNext(Inference i) {
+                assertTrue(i.getAnomalyScore() == 0.0);
+            }
+        };
+
+        network.subscribe(observer);
+
+        ManualInput inf = new ManualInput();
+        inf.anomalyScore(1.0);
+
+        network.compute(inf);
+
+        // Test that there are errors even though the test passes.
+        assertTrue(hasErrors(observer));
+    }
+
+    /**
+     * Demonstrates that the indirection of the internal PublishSubject within
+     * the Network's Layer class is what causes failed assertions to not be
+     * recognized by JUnit (i.e. they get swallowed). 
+     * 
+     * This test passes even though there are errors.
+     */
+    @Test
+    public void testCheckObservable_Correctly_Fails() {
+        FauxNetwork network = new FauxNetwork();
+
+        TestObserver<Inference> observer = new TestObserver<Inference>() {
+            @Override
+            public void onNext(Inference i) {
+                assertTrue(i.getAnomalyScore() == 0.0);
+            }
+        };
+
+        network.subscribe(observer);
+
+        ManualInput inf = new ManualInput();
+        inf.anomalyScore(1.0);
+
+        network.compute(inf);
+
+        // Test that there are errors even thought the test passes.
+        assertTrue(hasErrors(observer));
+
+        try {
+            checkObserver(observer);
+            // Should not reach this point. We want the checkObserver()
+            // method above to throw an exception which will indicate the
+            // test failure - though the test failing is what makes the 
+            // actual (negative) test pass.
+            fail(); 
+        }catch(Exception e) {
+            // Test passes when this point is reached.
+            assertTrue(e.getCause() instanceof AssertionError);
+        }
+    }
+
+
+    /**
+     * Mimics the internal dispatching of the Layer class of the Network package
+     * which uses a PublishSubject internally for central management of subscriptions
+     * and whose indirection causes the asserts within {@link Observer#onNext(Object)}
+     * to pass the tests when they actually shouldn't.
+     */
+    class FauxNetwork {
+        List<Observer<Inference>> observers = new ArrayList<>();
+
+        PublishSubject<Inference> internalDispatch = PublishSubject.create();
+        Observable<Inference> clientObservable;
+
+        public FauxNetwork() {
+            internalDispatch.subscribe(new Observer<Inference>() {
+                @Override public void onCompleted() {}
+                @Override public void onError(Throwable e) { e.printStackTrace(); }
+                @Override public void onNext(Inference i) {
+                    for(Observer<Inference> o : observers) {
+                        o.onNext(i);
+                    }
+                }
+            });
+
+            clientObservable = Observable.create(new Observable.OnSubscribe<Inference>() {
+                @SuppressWarnings("unchecked")
+                @Override public void call(Subscriber<? super Inference> t) {
+                    observers.add((Observer<Inference>)t);
+                }
+            });
+        }
+
+        public void subscribe(Observer<Inference> subscriber) {
+            clientObservable.subscribe(subscriber);
+        }
+
+        public void compute(Inference inference) {
+            internalDispatch.onNext(inference);
+        }
+    }
+}