diff --git a/src/__tests__/subscribeToMore.ts b/src/__tests__/subscribeToMore.ts
index 419b0b696d..6216eaa85c 100644
--- a/src/__tests__/subscribeToMore.ts
+++ b/src/__tests__/subscribeToMore.ts
@@ -4,7 +4,8 @@ import { DocumentNode, OperationDefinitionNode } from "graphql";
 import { ApolloClient } from "../core";
 import { InMemoryCache } from "../cache";
 import { ApolloLink, Operation } from "../link/core";
-import { itAsync, mockSingleLink, mockObservableLink } from "../testing";
+import { mockSingleLink, mockObservableLink, wait } from "../testing";
+import { ObservableStream, spyOnConsole } from "../testing/internal";
 
 const isSub = (operation: Operation) =>
   (operation.query as DocumentNode).definitions
@@ -57,13 +58,11 @@ describe("subscribeToMore", () => {
     name: string;
   }
 
-  itAsync("triggers new result from subscription data", (resolve, reject) => {
-    let latestResult: any = null;
+  it("triggers new result from subscription data", async () => {
     const wSLink = mockObservableLink();
-    const httpLink = mockSingleLink(req1).setOnError(reject);
+    const httpLink = mockSingleLink(req1);
 
     const link = ApolloLink.split(isSub, wSLink, httpLink);
-    let counter = 0;
 
     const client = new ApolloClient({
       cache: new InMemoryCache({ addTypename: false }),
@@ -71,21 +70,7 @@ describe("subscribeToMore", () => {
     });
 
     const obsHandle = client.watchQuery({ query });
-
-    const sub = obsHandle.subscribe({
-      next(queryResult: any) {
-        latestResult = queryResult;
-        if (++counter === 3) {
-          sub.unsubscribe();
-          expect(latestResult).toEqual({
-            data: { entry: { value: "Amanda Liu" } },
-            loading: false,
-            networkStatus: 7,
-          });
-          resolve();
-        }
-      },
-    });
+    const stream = new ObservableStream(obsHandle);
 
     obsHandle.subscribeToMore<SubscriptionData>({
       document: gql`
@@ -98,26 +83,35 @@ describe("subscribeToMore", () => {
       },
     });
 
-    let i = 0;
-    function simulate() {
-      const result = results[i++];
-      if (result) {
-        wSLink.simulateResult(result);
-        setTimeout(simulate, 10);
-      }
-    }
-    simulate();
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "1" } },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    wSLink.simulateResult(results[0]);
+
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "Dahivat Pandya" } },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    await wait(10);
+    wSLink.simulateResult(results[1]);
+
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "Amanda Liu" } },
+      loading: false,
+      networkStatus: 7,
+    });
   });
 
-  itAsync("calls error callback on error", (resolve, reject) => {
-    let latestResult: any = null;
+  it("calls error callback on error", async () => {
     const wSLink = mockObservableLink();
-    const httpLink = mockSingleLink(req1).setOnError(reject);
-
+    const httpLink = mockSingleLink(req1);
     const link = ApolloLink.split(isSub, wSLink, httpLink);
 
-    let counter = 0;
-
     const client = new ApolloClient({
       link,
       cache: new InMemoryCache({ addTypename: false }),
@@ -126,14 +120,9 @@ describe("subscribeToMore", () => {
     const obsHandle = client.watchQuery<(typeof req1)["result"]["data"]>({
       query,
     });
-    const sub = obsHandle.subscribe({
-      next(queryResult: any) {
-        latestResult = queryResult;
-        counter++;
-      },
-    });
+    const stream = new ObservableStream(obsHandle);
 
-    let errorCount = 0;
+    const onError = jest.fn();
 
     obsHandle.subscribeToMore<SubscriptionData>({
       document: gql`
@@ -144,98 +133,84 @@ describe("subscribeToMore", () => {
       updateQuery: (_, { subscriptionData }) => {
         return { entry: { value: subscriptionData.data.name } };
       },
-      onError: () => {
-        errorCount += 1;
-      },
+      onError,
     });
 
-    setTimeout(() => {
-      sub.unsubscribe();
-      expect(latestResult).toEqual({
-        data: { entry: { value: "Amanda Liu" } },
-        loading: false,
-        networkStatus: 7,
-      });
-      expect(counter).toBe(2);
-      expect(errorCount).toBe(1);
-      resolve();
-    }, 15);
-
-    for (let i = 0; i < 2; i++) {
-      wSLink.simulateResult(results2[i]);
+    for (const result of results2) {
+      wSLink.simulateResult(result);
     }
-  });
 
-  itAsync(
-    "prints unhandled subscription errors to the console",
-    (resolve, reject) => {
-      let latestResult: any = null;
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "1" } },
+      loading: false,
+      networkStatus: 7,
+    });
 
-      const wSLink = mockObservableLink();
-      const httpLink = mockSingleLink(req1).setOnError(reject);
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "Amanda Liu" } },
+      loading: false,
+      networkStatus: 7,
+    });
 
-      const link = ApolloLink.split(isSub, wSLink, httpLink);
+    await wait(15);
 
-      let counter = 0;
+    expect(onError).toHaveBeenCalledTimes(1);
+    expect(onError).toHaveBeenCalledWith(new Error("You cant touch this"));
+  });
 
-      const client = new ApolloClient({
-        link,
-        cache: new InMemoryCache({ addTypename: false }),
-      });
+  it("prints unhandled subscription errors to the console", async () => {
+    using _ = spyOnConsole("error");
 
-      const obsHandle = client.watchQuery({
-        query,
-      });
-      const sub = obsHandle.subscribe({
-        next(queryResult: any) {
-          latestResult = queryResult;
-          counter++;
-        },
-      });
+    const wSLink = mockObservableLink();
+    const httpLink = mockSingleLink(req1);
+    const link = ApolloLink.split(isSub, wSLink, httpLink);
 
-      let errorCount = 0;
-      const consoleErr = console.error;
-      console.error = (_: Error) => {
-        errorCount += 1;
-      };
+    const client = new ApolloClient({
+      link,
+      cache: new InMemoryCache({ addTypename: false }),
+    });
 
-      obsHandle.subscribeToMore({
-        document: gql`
-          subscription newValues {
-            name
-          }
-        `,
-        updateQuery: () => {
-          throw new Error("should not be called because of initial error");
-        },
-      });
-
-      setTimeout(() => {
-        sub.unsubscribe();
-        expect(latestResult).toEqual({
-          data: { entry: { value: "1" } },
-          loading: false,
-          networkStatus: 7,
-        });
-        expect(counter).toBe(1);
-        expect(errorCount).toBe(1);
-        console.error = consoleErr;
-        resolve();
-      }, 15);
-
-      for (let i = 0; i < 2; i++) {
-        wSLink.simulateResult(results3[i]);
-      }
+    const obsHandle = client.watchQuery({
+      query,
+    });
+    const stream = new ObservableStream(obsHandle);
+
+    obsHandle.subscribeToMore({
+      document: gql`
+        subscription newValues {
+          name
+        }
+      `,
+      updateQuery: () => {
+        throw new Error("should not be called because of initial error");
+      },
+    });
+
+    for (const result of results3) {
+      wSLink.simulateResult(result);
     }
-  );
 
-  itAsync("should not corrupt the cache (#3062)", async (resolve, reject) => {
-    let latestResult: any = null;
-    const wSLink = mockObservableLink();
-    const httpLink = mockSingleLink(req4).setOnError(reject);
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "1" } },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    await wait(15);
+
+    expect(console.error).toHaveBeenCalledTimes(1);
+    expect(console.error).toHaveBeenCalledWith(
+      "Unhandled GraphQL subscription error",
+      new Error("You cant touch this")
+    );
+
+    await expect(stream).not.toEmitAnything();
+  });
 
+  it("should not corrupt the cache (#3062)", async () => {
+    const wSLink = mockObservableLink();
+    const httpLink = mockSingleLink(req4);
     const link = ApolloLink.split(isSub, wSLink, httpLink);
-    let counter = 0;
 
     const client = new ApolloClient({
       cache: new InMemoryCache({ addTypename: false }).restore({
@@ -256,13 +231,7 @@ describe("subscribeToMore", () => {
     const obsHandle = client.watchQuery<(typeof req4)["result"]["data"]>({
       query,
     });
-
-    const sub = obsHandle.subscribe({
-      next(queryResult: any) {
-        latestResult = queryResult;
-        counter++;
-      },
-    });
+    const stream = new ObservableStream(obsHandle);
 
     let nextMutation: { value: string };
     obsHandle.subscribeToMore<SubscriptionData>({
@@ -279,9 +248,6 @@ describe("subscribeToMore", () => {
       },
     });
 
-    const wait = (dur: any) =>
-      new Promise((resolve) => setTimeout(resolve, dur));
-
     for (let i = 0; i < 2; i++) {
       // init optimistic mutation
       let data = client.cache.readQuery<(typeof req4)["result"]["data"]>(
@@ -302,105 +268,104 @@ describe("subscribeToMore", () => {
       client.cache.removeOptimistic(i.toString());
       // note: we don't complete mutation with performTransaction because a real example would detect duplicates
     }
-    sub.unsubscribe();
-    expect(counter).toBe(3);
-    expect(latestResult).toEqual({
+
+    await expect(stream).toEmitValue({
+      data: { entry: [{ value: 1 }, { value: 2 }] },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    await expect(stream).toEmitValue({
+      data: {
+        entry: [{ value: 1 }, { value: 2 }, { value: "Dahivat Pandya" }],
+      },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    await expect(stream).toEmitValue({
       data: {
         entry: [
-          {
-            value: 1,
-          },
-          {
-            value: 2,
-          },
-          {
-            value: "Dahivat Pandya",
-          },
-          {
-            value: "Amanda Liu",
-          },
+          { value: 1 },
+          { value: 2 },
+          { value: "Dahivat Pandya" },
+          { value: "Amanda Liu" },
         ],
       },
       loading: false,
       networkStatus: 7,
     });
-    resolve();
+
+    await expect(stream).not.toEmitAnything();
   });
   // TODO add a test that checks that subscriptions are cancelled when obs is unsubscribed from.
 
-  itAsync(
-    "allows specification of custom types for variables and payload (#4246)",
-    (resolve, reject) => {
-      interface TypedOperation extends Operation {
-        variables: {
-          someNumber: number;
-        };
-      }
-      const typedReq = {
-        request: { query, variables: { someNumber: 1 } } as TypedOperation,
-        result,
+  it("allows specification of custom types for variables and payload (#4246)", async () => {
+    interface TypedOperation extends Operation {
+      variables: {
+        someNumber: number;
       };
-      interface TypedSubscriptionVariables {
-        someString: string;
-      }
+    }
+    const typedReq = {
+      request: { query, variables: { someNumber: 1 } } as TypedOperation,
+      result,
+    };
+    interface TypedSubscriptionVariables {
+      someString: string;
+    }
 
-      let latestResult: any = null;
-      const wSLink = mockObservableLink();
-      const httpLink = mockSingleLink(typedReq).setOnError(reject);
-
-      const link = ApolloLink.split(isSub, wSLink, httpLink);
-      let counter = 0;
-
-      const client = new ApolloClient({
-        cache: new InMemoryCache({ addTypename: false }),
-        link,
-      });
-
-      type TData = (typeof typedReq)["result"]["data"];
-      type TVars = (typeof typedReq)["request"]["variables"];
-      const obsHandle = client.watchQuery<TData, TVars>({
-        query,
-        variables: { someNumber: 1 },
-      });
-
-      const sub = obsHandle.subscribe({
-        next(queryResult: any) {
-          latestResult = queryResult;
-          if (++counter === 3) {
-            sub.unsubscribe();
-            expect(latestResult).toEqual({
-              data: { entry: { value: "Amanda Liu" } },
-              loading: false,
-              networkStatus: 7,
-            });
-            resolve();
-          }
-        },
-      });
-
-      obsHandle.subscribeToMore<SubscriptionData, TypedSubscriptionVariables>({
-        document: gql`
-          subscription newValues {
-            name
-          }
-        `,
-        variables: {
-          someString: "foo",
-        },
-        updateQuery: (_, { subscriptionData }) => {
-          return { entry: { value: subscriptionData.data.name } };
-        },
-      });
-
-      let i = 0;
-      function simulate() {
-        const result = results[i++];
-        if (result) {
-          wSLink.simulateResult(result);
-          setTimeout(simulate, 10);
+    const wSLink = mockObservableLink();
+    const httpLink = mockSingleLink(typedReq);
+    const link = ApolloLink.split(isSub, wSLink, httpLink);
+
+    const client = new ApolloClient({
+      cache: new InMemoryCache({ addTypename: false }),
+      link,
+    });
+
+    type TData = (typeof typedReq)["result"]["data"];
+    type TVars = (typeof typedReq)["request"]["variables"];
+    const obsHandle = client.watchQuery<TData, TVars>({
+      query,
+      variables: { someNumber: 1 },
+    });
+    const stream = new ObservableStream(obsHandle);
+
+    obsHandle.subscribeToMore<SubscriptionData, TypedSubscriptionVariables>({
+      document: gql`
+        subscription newValues {
+          name
         }
-      }
-      simulate();
-    }
-  );
+      `,
+      variables: {
+        someString: "foo",
+      },
+      updateQuery: (_, { subscriptionData }) => {
+        return { entry: { value: subscriptionData.data.name } };
+      },
+    });
+
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "1" } },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    wSLink.simulateResult(results[0]);
+
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "Dahivat Pandya" } },
+      loading: false,
+      networkStatus: 7,
+    });
+
+    await wait(10);
+    wSLink.simulateResult(results[1]);
+
+    await expect(stream).toEmitValue({
+      data: { entry: { value: "Amanda Liu" } },
+      loading: false,
+      networkStatus: 7,
+    });
+  });
 });