diff --git a/rxjava-core/src/test/java/rx/CombineLatestTests.java b/rxjava-core/src/test/java/rx/CombineLatestTests.java
new file mode 100644
index 0000000000..78dbd4c4e3
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/CombineLatestTests.java
@@ -0,0 +1,53 @@
+package rx;
+
+import org.junit.Test;
+
+import rx.CovarianceTest.CoolRating;
+import rx.CovarianceTest.ExtendedResult;
+import rx.CovarianceTest.HorrorMovie;
+import rx.CovarianceTest.Media;
+import rx.CovarianceTest.Movie;
+import rx.CovarianceTest.Rating;
+import rx.CovarianceTest.Result;
+import rx.util.functions.Action1;
+import rx.util.functions.Func2;
+
+public class CombineLatestTests {
+    /**
+     * This won't compile if super/extends isn't done correctly on generics
+     */
+    @Test
+    public void testCovarianceOfCombineLatest() {
+        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
+        Observable<CoolRating> ratings = Observable.from(new CoolRating());
+
+        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(extendedAction);
+        Observable.<Media, Rating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
+
+        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine);
+    }
+
+    Func2<Media, Rating, ExtendedResult> combine = new Func2<Media, Rating, ExtendedResult>() {
+        @Override
+        public ExtendedResult call(Media m, Rating r) {
+            return new ExtendedResult();
+        }
+    };
+
+    Action1<Result> action = new Action1<Result>() {
+        @Override
+        public void call(Result t1) {
+            System.out.println("Result: " + t1);
+        }
+    };
+
+    Action1<ExtendedResult> extendedAction = new Action1<ExtendedResult>() {
+        @Override
+        public void call(ExtendedResult t1) {
+            System.out.println("Result: " + t1);
+        }
+    };
+}
diff --git a/rxjava-core/src/test/java/rx/ConcatTests.java b/rxjava-core/src/test/java/rx/ConcatTests.java
index 524b7fd73f..29a51d8dfe 100644
--- a/rxjava-core/src/test/java/rx/ConcatTests.java
+++ b/rxjava-core/src/test/java/rx/ConcatTests.java
@@ -7,6 +7,12 @@
 
 import org.junit.Test;
 
+import rx.CovarianceTest.HorrorMovie;
+import rx.CovarianceTest.Media;
+import rx.CovarianceTest.Movie;
+import rx.Observable.OnSubscribeFunc;
+import rx.subscriptions.Subscriptions;
+
 public class ConcatTests {
 
     @Test
@@ -54,4 +60,62 @@ public void testConcatWithIterableOfObservable() {
         assertEquals("three", values.get(2));
         assertEquals("four", values.get(3));
     }
+    
+    @Test
+    public void testConcatCovariance() {
+        Observable<Media> o1 = Observable.<Media> from(new HorrorMovie(), new Movie());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        Observable<Observable<Media>> os = Observable.from(o1, o2);
+
+        List<Media> values = Observable.concat(os).toList().toBlockingObservable().single();
+    }
+
+    @Test
+    public void testConcatCovariance2() {
+        Observable<Media> o1 = Observable.from(new HorrorMovie(), new Movie(), new Media());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        Observable<Observable<Media>> os = Observable.from(o1, o2);
+
+        List<Media> values = Observable.concat(os).toList().toBlockingObservable().single();
+    }
+    
+    @Test
+    public void testConcatCovariance3() {
+        Observable<Movie> o1 = Observable.from(new HorrorMovie(), new Movie());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        List<Media> values = Observable.concat(o1, o2).toList().toBlockingObservable().single();
+        
+        assertTrue(values.get(0) instanceof HorrorMovie);
+        assertTrue(values.get(1) instanceof Movie);
+        assertTrue(values.get(2) instanceof Media);
+        assertTrue(values.get(3) instanceof HorrorMovie);
+    }
+
+    @Test
+    public void testConcatCovariance4() {
+
+        Observable<Movie> o1 = Observable.create(new OnSubscribeFunc<Movie>() {
+
+            @Override
+            public Subscription onSubscribe(Observer<? super Movie> o) {
+                o.onNext(new HorrorMovie());
+                o.onNext(new Movie());
+                //                o.onNext(new Media()); // correctly doesn't compile
+                o.onCompleted();
+                return Subscriptions.empty();
+            }
+        });
+
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        List<Media> values = Observable.concat(o1, o2).toList().toBlockingObservable().single();
+        
+        assertTrue(values.get(0) instanceof HorrorMovie);
+        assertTrue(values.get(1) instanceof Movie);
+        assertTrue(values.get(2) instanceof Media);
+        assertTrue(values.get(3) instanceof HorrorMovie);
+    }
 }
diff --git a/rxjava-core/src/test/java/rx/CovarianceTest.java b/rxjava-core/src/test/java/rx/CovarianceTest.java
index 4eb31f0da1..69110b6c6a 100644
--- a/rxjava-core/src/test/java/rx/CovarianceTest.java
+++ b/rxjava-core/src/test/java/rx/CovarianceTest.java
@@ -1,17 +1,9 @@
 package rx;
 
-import static org.junit.Assert.*;
-
 import java.util.ArrayList;
-import java.util.List;
 
 import org.junit.Test;
 
-import rx.Observable.OnSubscribeFunc;
-import rx.subscriptions.Subscriptions;
-import rx.util.functions.Action1;
-import rx.util.functions.Func2;
-
 /**
  * Test super/extends of generics.
  * 
@@ -29,187 +21,9 @@ public void testCovarianceOfFrom() {
         // Observable.<HorrorMovie>from(new Movie()); // may not compile
     }
 
-    /**
-     * This won't compile if super/extends isn't done correctly on generics
-     */
-    @Test
-    public void testCovarianceOfMerge() {
-        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
-        Observable<Observable<HorrorMovie>> metaHorrors = Observable.just(horrors);
-        Observable.<Media> merge(metaHorrors);
-    }
-
-    /**
-     * This won't compile if super/extends isn't done correctly on generics
-     */
-    @Test
-    public void testCovarianceOfZip() {
-        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
-        Observable<CoolRating> ratings = Observable.from(new CoolRating());
-
-        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(extendedAction);
-        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
-
-        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);
-    }
-
-    /**
-     * This won't compile if super/extends isn't done correctly on generics
+    /*
+     * Most tests are moved into their applicable classes such as [Operator]Tests.java
      */
-    @Test
-    public void testCovarianceOfCombineLatest() {
-        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
-        Observable<CoolRating> ratings = Observable.from(new CoolRating());
-
-        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(extendedAction);
-        Observable.<Media, Rating, Result> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
-        Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).toBlockingObservable().forEach(action);
-
-        Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine);
-    }
-
-    @Test
-    public void testConcatCovariance() {
-        Observable<Media> o1 = Observable.<Media> from(new HorrorMovie(), new Movie());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        Observable<Observable<Media>> os = Observable.from(o1, o2);
-
-        List<Media> values = Observable.concat(os).toList().toBlockingObservable().single();
-    }
-
-    @Test
-    public void testConcatCovariance2() {
-        Observable<Media> o1 = Observable.from(new HorrorMovie(), new Movie(), new Media());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        Observable<Observable<Media>> os = Observable.from(o1, o2);
-
-        List<Media> values = Observable.concat(os).toList().toBlockingObservable().single();
-    }
-    
-    @Test
-    public void testConcatCovariance3() {
-        Observable<Movie> o1 = Observable.from(new HorrorMovie(), new Movie());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        List<Media> values = Observable.concat(o1, o2).toList().toBlockingObservable().single();
-        
-        assertTrue(values.get(0) instanceof HorrorMovie);
-        assertTrue(values.get(1) instanceof Movie);
-        assertTrue(values.get(2) instanceof Media);
-        assertTrue(values.get(3) instanceof HorrorMovie);
-    }
-
-    @Test
-    public void testConcatCovariance4() {
-
-        Observable<Movie> o1 = Observable.create(new OnSubscribeFunc<Movie>() {
-
-            @Override
-            public Subscription onSubscribe(Observer<? super Movie> o) {
-                o.onNext(new HorrorMovie());
-                o.onNext(new Movie());
-                //                o.onNext(new Media()); // correctly doesn't compile
-                o.onCompleted();
-                return Subscriptions.empty();
-            }
-        });
-
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        List<Media> values = Observable.concat(o1, o2).toList().toBlockingObservable().single();
-        
-        assertTrue(values.get(0) instanceof HorrorMovie);
-        assertTrue(values.get(1) instanceof Movie);
-        assertTrue(values.get(2) instanceof Media);
-        assertTrue(values.get(3) instanceof HorrorMovie);
-    }
-    
-    
-    @Test
-    public void testMergeCovariance() {
-        Observable<Media> o1 = Observable.<Media> from(new HorrorMovie(), new Movie());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        Observable<Observable<Media>> os = Observable.from(o1, o2);
-
-        List<Media> values = Observable.merge(os).toList().toBlockingObservable().single();
-    }
-
-    @Test
-    public void testMergeCovariance2() {
-        Observable<Media> o1 = Observable.from(new HorrorMovie(), new Movie(), new Media());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        Observable<Observable<Media>> os = Observable.from(o1, o2);
-
-        List<Media> values = Observable.merge(os).toList().toBlockingObservable().single();
-    }
-    
-    @Test
-    public void testMergeCovariance3() {
-        Observable<Movie> o1 = Observable.from(new HorrorMovie(), new Movie());
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        List<Media> values = Observable.merge(o1, o2).toList().toBlockingObservable().single();
-        
-        assertTrue(values.get(0) instanceof HorrorMovie);
-        assertTrue(values.get(1) instanceof Movie);
-        assertTrue(values.get(2) instanceof Media);
-        assertTrue(values.get(3) instanceof HorrorMovie);
-    }
-
-    @Test
-    public void testMergeCovariance4() {
-
-        Observable<Movie> o1 = Observable.create(new OnSubscribeFunc<Movie>() {
-
-            @Override
-            public Subscription onSubscribe(Observer<? super Movie> o) {
-                o.onNext(new HorrorMovie());
-                o.onNext(new Movie());
-                //                o.onNext(new Media()); // correctly doesn't compile
-                o.onCompleted();
-                return Subscriptions.empty();
-            }
-        });
-
-        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
-
-        List<Media> values = Observable.merge(o1, o2).toList().toBlockingObservable().single();
-        
-        assertTrue(values.get(0) instanceof HorrorMovie);
-        assertTrue(values.get(1) instanceof Movie);
-        assertTrue(values.get(2) instanceof Media);
-        assertTrue(values.get(3) instanceof HorrorMovie);
-    }
-
-    Func2<Media, Rating, ExtendedResult> combine = new Func2<Media, Rating, ExtendedResult>() {
-        @Override
-        public ExtendedResult call(Media m, Rating r) {
-            return new ExtendedResult();
-        }
-    };
-
-    Action1<Result> action = new Action1<Result>() {
-        @Override
-        public void call(Result t1) {
-            System.out.println("Result: " + t1);
-        }
-    };
-
-    Action1<ExtendedResult> extendedAction = new Action1<ExtendedResult>() {
-        @Override
-        public void call(ExtendedResult t1) {
-            System.out.println("Result: " + t1);
-        }
-    };
 
     static class Media {
     }
diff --git a/rxjava-core/src/test/java/rx/MergeTests.java b/rxjava-core/src/test/java/rx/MergeTests.java
new file mode 100644
index 0000000000..11f30c7908
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/MergeTests.java
@@ -0,0 +1,86 @@
+package rx;
+
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import rx.CovarianceTest.HorrorMovie;
+import rx.CovarianceTest.Media;
+import rx.CovarianceTest.Movie;
+import rx.Observable.OnSubscribeFunc;
+import rx.subscriptions.Subscriptions;
+
+public class MergeTests {
+
+    /**
+     * This won't compile if super/extends isn't done correctly on generics
+     */
+    @Test
+    public void testCovarianceOfMerge() {
+        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
+        Observable<Observable<HorrorMovie>> metaHorrors = Observable.just(horrors);
+        Observable.<Media> merge(metaHorrors);
+    }
+
+    @Test
+    public void testMergeCovariance() {
+        Observable<Media> o1 = Observable.<Media> from(new HorrorMovie(), new Movie());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        Observable<Observable<Media>> os = Observable.from(o1, o2);
+
+        List<Media> values = Observable.merge(os).toList().toBlockingObservable().single();
+    }
+
+    @Test
+    public void testMergeCovariance2() {
+        Observable<Media> o1 = Observable.from(new HorrorMovie(), new Movie(), new Media());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        Observable<Observable<Media>> os = Observable.from(o1, o2);
+
+        List<Media> values = Observable.merge(os).toList().toBlockingObservable().single();
+    }
+    
+    @Test
+    public void testMergeCovariance3() {
+        Observable<Movie> o1 = Observable.from(new HorrorMovie(), new Movie());
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        List<Media> values = Observable.merge(o1, o2).toList().toBlockingObservable().single();
+        
+        assertTrue(values.get(0) instanceof HorrorMovie);
+        assertTrue(values.get(1) instanceof Movie);
+        assertTrue(values.get(2) instanceof Media);
+        assertTrue(values.get(3) instanceof HorrorMovie);
+    }
+
+    @Test
+    public void testMergeCovariance4() {
+
+        Observable<Movie> o1 = Observable.create(new OnSubscribeFunc<Movie>() {
+
+            @Override
+            public Subscription onSubscribe(Observer<? super Movie> o) {
+                o.onNext(new HorrorMovie());
+                o.onNext(new Movie());
+                //                o.onNext(new Media()); // correctly doesn't compile
+                o.onCompleted();
+                return Subscriptions.empty();
+            }
+        });
+
+        Observable<Media> o2 = Observable.from(new Media(), new HorrorMovie());
+
+        List<Media> values = Observable.merge(o1, o2).toList().toBlockingObservable().single();
+        
+        assertTrue(values.get(0) instanceof HorrorMovie);
+        assertTrue(values.get(1) instanceof Movie);
+        assertTrue(values.get(2) instanceof Media);
+        assertTrue(values.get(3) instanceof HorrorMovie);
+    }
+    
+    
+}
diff --git a/rxjava-core/src/test/java/rx/ReduceTests.java b/rxjava-core/src/test/java/rx/ReduceTests.java
new file mode 100644
index 0000000000..822001e615
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/ReduceTests.java
@@ -0,0 +1,69 @@
+package rx;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import rx.CovarianceTest.HorrorMovie;
+import rx.CovarianceTest.Movie;
+import rx.operators.OperationScan;
+import rx.util.functions.Func2;
+
+public class ReduceTests {
+
+    @Test
+    public void reduceInts() {
+        Observable<Integer> o = Observable.from(1, 2, 3);
+        int value = o.reduce(new Func2<Integer, Integer, Integer>() {
+
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + t2;
+            }
+        }).toBlockingObservable().single();
+
+        assertEquals(6, value);
+    }
+
+    @Test
+    public void reduceWithObjects() {
+        Observable<HorrorMovie> horrorMovies = Observable.from(new HorrorMovie());
+
+        Func2<Movie, Movie, Movie> chooseSecondMovie =
+                new Func2<Movie, Movie, Movie>() {
+                    public Movie call(Movie t1, Movie t2) {
+                        return t2;
+                    }
+                };
+
+        Observable<Movie> reduceResult = Observable.create(OperationScan.scan(horrorMovies, chooseSecondMovie)).takeLast(1);
+
+        //TODO this isn't compiling
+        //        Observable<Movie> reduceResult2 = horrorMovies.reduce(chooseSecondMovie);
+    }
+
+    @Test
+    public void reduceCovariance() {
+        Observable<HorrorMovie> horrorMovies = Observable.from(new HorrorMovie());
+
+        // do something with horrorMovies, relying on the fact that all are HorrorMovies
+        // and not just any Movies...
+
+        // pass it to library (works because it takes Observable<? extends Movie>)
+        libraryFunctionActingOnMovieObservables(horrorMovies);
+    }
+
+    public void libraryFunctionActingOnMovieObservables(Observable<? extends Movie> obs) {
+        Func2<Movie, Movie, Movie> chooseSecondMovie =
+                new Func2<Movie, Movie, Movie>() {
+                    public Movie call(Movie t1, Movie t2) {
+                        return t2;
+                    }
+                };
+
+        //TODO this isn't compiling
+        //        Observable<Movie> reduceResult = obs.reduce((Func2<? super Movie, ? super Movie, ? extends Movie>) chooseSecondMovie); 
+        // do something with reduceResult...
+    }
+
+}
diff --git a/rxjava-core/src/test/java/rx/ZipTests.java b/rxjava-core/src/test/java/rx/ZipTests.java
new file mode 100644
index 0000000000..ee25bb6ade
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/ZipTests.java
@@ -0,0 +1,54 @@
+package rx;
+
+import org.junit.Test;
+
+import rx.CovarianceTest.CoolRating;
+import rx.CovarianceTest.ExtendedResult;
+import rx.CovarianceTest.HorrorMovie;
+import rx.CovarianceTest.Media;
+import rx.CovarianceTest.Movie;
+import rx.CovarianceTest.Rating;
+import rx.CovarianceTest.Result;
+import rx.util.functions.Action1;
+import rx.util.functions.Func2;
+
+public class ZipTests {
+
+    /**
+     * This won't compile if super/extends isn't done correctly on generics
+     */
+    @Test
+    public void testCovarianceOfZip() {
+        Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());
+        Observable<CoolRating> ratings = Observable.from(new CoolRating());
+
+        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(extendedAction);
+        Observable.<Media, Rating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
+        Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);
+
+        Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);
+    }
+
+    Func2<Media, Rating, ExtendedResult> combine = new Func2<Media, Rating, ExtendedResult>() {
+        @Override
+        public ExtendedResult call(Media m, Rating r) {
+            return new ExtendedResult();
+        }
+    };
+
+    Action1<Result> action = new Action1<Result>() {
+        @Override
+        public void call(Result t1) {
+            System.out.println("Result: " + t1);
+        }
+    };
+
+    Action1<ExtendedResult> extendedAction = new Action1<ExtendedResult>() {
+        @Override
+        public void call(ExtendedResult t1) {
+            System.out.println("Result: " + t1);
+        }
+    };
+}