More accurate subscriber for observe helper is added (#65)

It prevents appearance of successful test
with fails in finalize operator

examples/jasmine/observe-spec.ts

@@ -1,6 +1,6 @@
 import { observe } from "rxjs-marbles/jasmine";
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 
 describe("observe", () => {
   it(
@@ -12,4 +12,15 @@ describe("observe", () => {
       );
     })
   );
+
+  it(
+    "should handle assertions in finalize operator",
+    observe(() => {
+      const mock = jasmine.createSpy();
+      return of(1).pipe(
+        tap(() => mock()),
+        finalize(() => expect(mock).toHaveBeenCalled())
+      );
+    })
+  );
 });

examples/jest/observe-spec.ts

@@ -1,6 +1,6 @@
 import { observe } from "rxjs-marbles/jest";
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 
 describe("observe", () => {
   it(
@@ -12,4 +12,15 @@ describe("observe", () => {
       );
     })
   );
+
+  it(
+    "should handle assertions in finalize operator",
+    observe(() => {
+      const mock = jest.fn();
+      return of(1).pipe(
+        tap(() => mock()),
+        finalize(() => expect(mock).toHaveBeenCalled())
+      );
+    })
+  );
 });

examples/mocha/observe-spec.ts

@@ -1,7 +1,7 @@
-import { expect } from "chai";
+import { assert, expect } from "chai";
 import { observe } from "rxjs-marbles/mocha";
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 
 describe("observe", () => {
   it(
@@ -13,4 +13,16 @@ describe("observe", () => {
       );
     })
   );
+
+  it(
+    "should handle assertions in finalize operator",
+    observe(() => {
+      let haveBeenCalled = false;
+      const mock = () => (haveBeenCalled = true);
+      return of(1).pipe(
+        tap(() => mock()),
+        finalize(() => assert.isOk(haveBeenCalled))
+      );
+    })
+  );
 });

fixtures/jasmine/failing-spec.ts

@@ -4,7 +4,7 @@
  */
 
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 import { marbles, observe } from "../../dist/jasmine";
 
 if (process.env.FAILING !== "0") {
@@ -38,5 +38,16 @@ if (process.env.FAILING !== "0") {
         of("fail").pipe(tap(value => expect(value).not.toEqual("fail")))
       )
     );
+
+    it(
+      "should fail on assertions in finalize operator",
+      observe(() => {
+        const mock = jasmine.createSpy();
+        return of("fail").pipe(
+          tap(() => mock()),
+          finalize(() => expect(mock).not.toHaveBeenCalled())
+        );
+      })
+    );
   });
 }

fixtures/jasmine/passing-spec.ts

@@ -5,7 +5,7 @@
 /*tslint:disable:object-literal-sort-keys*/
 
 import { asapScheduler, of, timer } from "rxjs";
-import { delay, map, tap } from "rxjs/operators";
+import { delay, finalize, map, tap } from "rxjs/operators";
 import {
   cases,
   DoneFunction,
@@ -128,6 +128,17 @@ describe("observe", () => {
     "should support observe",
     observe(() => of("pass").pipe(tap(value => expect(value).toEqual("pass"))))
   );
+
+  it(
+    "should handle assertions in finalize operator",
+    observe(() => {
+      const mock = jasmine.createSpy();
+      return of("pass").pipe(
+        tap(() => mock()),
+        finalize(() => expect(mock).toHaveBeenCalled())
+      );
+    })
+  );
 });
 
 describe("fakeSchedulers", () => {

fixtures/jest/failing-spec.ts

@@ -4,37 +4,53 @@
  */
 
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 import { marbles, observe } from "../../dist/jest";
 
 if (process.env.FAILING !== "0") {
-  test(
-    "it should fail with marbles",
-    marbles(m => {
-      const values = {
-        a: 1,
-        b: 2,
-        c: 3,
-        d: 4
-      };
+  describe("marbles", () => {
+    test(
+      "it should fail",
+      marbles(m => {
+        const values = {
+          a: 1,
+          b: 2,
+          c: 3,
+          d: 4
+        };
 
-      const source = m.hot("  --^-a-b-c-|", values);
-      const subs = "            ^-------!";
-      const expected = m.cold(" --a-a-a-|", values);
+        const source = m.hot("  --^-a-b-c-|", values);
+        const subs = "            ^-------!";
+        const expected = m.cold(" --a-a-a-|", values);
 
-      const destination = source.pipe(map(value => value + 1));
+        const destination = source.pipe(map(value => value + 1));
 
-      m.expect(destination).toBeObservable(expected);
-      m.expect(source).toHaveSubscriptions(subs);
-    })
-  );
+        m.expect(destination).toBeObservable(expected);
+        m.expect(source).toHaveSubscriptions(subs);
+      })
+    );
+  });
 
-  test(
-    "it should fail with observe",
-    observe(() =>
-      of("fail").pipe(tap(value => expect(value).not.toEqual("fail")))
-    )
-  );
+  describe("observe", () => {
+    test(
+      "it should fail",
+      observe(() =>
+        of("fail").pipe(tap(value => expect(value).not.toEqual("fail")))
+      )
+    );
+
+    test(
+      "it should fail on assertions in finalize operator",
+      observe(() => {
+        expect.assertions(1);
+        const mock = jest.fn();
+        return of("fail").pipe(
+          tap(() => mock()),
+          finalize(() => expect(mock).not.toHaveBeenCalled())
+        );
+      })
+    );
+  });
 } else {
   test("it should pass", () => {});
 }

fixtures/jest/passing-spec.ts

@@ -5,7 +5,7 @@
 /*tslint:disable:object-literal-sort-keys*/
 
 import { asapScheduler, of, timer } from "rxjs";
-import { delay, map, tap } from "rxjs/operators";
+import { delay, finalize, map, tap } from "rxjs/operators";
 import {
   cases,
   DoneFunction,
@@ -124,6 +124,18 @@ describe("observe", () => {
     "it should support observe",
     observe(() => of("pass").pipe(tap(value => expect(value).toEqual("pass"))))
   );
+
+  test(
+    "it should handle assertions in finalize operator",
+    observe(() => {
+      expect.assertions(1);
+      const mock = jest.fn();
+      return of("pass").pipe(
+        tap(() => mock()),
+        finalize(() => expect(mock).toHaveBeenCalled())
+      );
+    })
+  );
 });
 
 describe("fakeSchedulers", () => {

fixtures/mocha/failing-spec.ts

@@ -3,9 +3,9 @@
  * can be found in the LICENSE file at https://github.com/cartant/rxjs-marbles
  */
 
-import { expect } from "chai";
+import { assert, expect } from "chai";
 import { of } from "rxjs";
-import { map, tap } from "rxjs/operators";
+import { finalize, map, tap } from "rxjs/operators";
 import { marbles, observe } from "../../dist/mocha";
 
 if (process.env.FAILING !== "0") {
@@ -39,5 +39,17 @@ if (process.env.FAILING !== "0") {
         of("fail").pipe(tap(value => expect(value).to.not.equal("fail")))
       )
     );
+
+    it(
+      "should fail on assertions in finalize operator",
+      observe(() => {
+        let haveBeenCalled = false;
+        const mock = () => (haveBeenCalled = true);
+        return of("fail").pipe(
+          tap(() => mock()),
+          finalize(() => assert.isNotOk(haveBeenCalled))
+        );
+      })
+    );
   });
 }

fixtures/mocha/passing-spec.ts

@@ -4,9 +4,9 @@
  */
 /*tslint:disable:no-unused-expression object-literal-sort-keys*/
 
-import { expect } from "chai";
+import { assert, expect } from "chai";
 import { asapScheduler, of, timer } from "rxjs";
-import { delay, map, tap } from "rxjs/operators";
+import { delay, finalize, map, tap } from "rxjs/operators";
 import * as sinon from "sinon";
 import {
   configure,
@@ -647,6 +647,18 @@ describe("observe", () => {
     "should support observe",
     observe(() => of("pass").pipe(tap(value => expect(value).to.equal("pass"))))
   );
+
+  it(
+    "should handle assertions in finalize operator",
+    observe(() => {
+      let haveBeenCalled = false;
+      const mock = () => (haveBeenCalled = true);
+      return of("pass").pipe(
+        tap(() => mock()),
+        finalize(() => assert.isOk(haveBeenCalled))
+      );
+    })
+  );
 });
 
 describe("fakeSchedulers", () => {

source/jasmine/observe.ts

@@ -4,6 +4,7 @@
  */
 
 import { Observable } from "rxjs";
+import { VerboseSubscriber } from "../verbose-subscriber";
 
 export interface DoneFunction {
   (): void;
@@ -13,6 +14,8 @@ export interface DoneFunction {
 export function observe<T>(
   observableTest: () => Observable<T>
 ): (done: DoneFunction) => void {
-  return (done: DoneFunction) =>
-    observableTest().subscribe(undefined, done.fail, done);
+  return (done: DoneFunction) => {
+    const subscriber = new VerboseSubscriber(done.fail, done);
+    observableTest().subscribe(subscriber);
+  };
 }

source/jest/observe.ts

@@ -4,6 +4,7 @@
  */
 
 import { Observable } from "rxjs";
+import { VerboseSubscriber } from "../verbose-subscriber";
 
 export interface DoneFunction {
   (): void;
@@ -13,6 +14,8 @@ export interface DoneFunction {
 export function observe<T>(
   observableTest: () => Observable<T>
 ): (done: DoneFunction) => void {
-  return (done: DoneFunction) =>
-    observableTest().subscribe(undefined, done.fail, done);
+  return (done: DoneFunction) => {
+    const subscriber = new VerboseSubscriber(done.fail, done);
+    observableTest().subscribe(subscriber);
+  };
 }

source/mocha/observe.ts

@@ -4,12 +4,15 @@
  */
 
 import { Observable } from "rxjs";
+import { VerboseSubscriber } from "../verbose-subscriber";
 
 export type DoneFunction = (error?: Error) => void;
 
 export function observe<T>(
   observableTest: () => Observable<T>
 ): (done: DoneFunction) => void {
-  return (done: DoneFunction) =>
-    observableTest().subscribe(undefined, done, done);
+  return (done: DoneFunction) => {
+    const subscriber = new VerboseSubscriber(done, done);
+    observableTest().subscribe(subscriber);
+  };
 }

source/verbose-subscriber.ts

@@ -0,0 +1,15 @@
+import { Subscriber } from "rxjs";
+
+export class VerboseSubscriber<T> extends Subscriber<T> {
+  constructor(private onError: (error: any) => void, onComplete: () => void) {
+    super(undefined, onError, onComplete);
+  }
+
+  unsubscribe(): void {
+    try {
+      super.unsubscribe();
+    } catch (error) {
+      this.onError(error);
+    }
+  }
+}