[Unit testing RxJS] Force completion for long running observables with marbles

describe("Marble testing in Rxjs", () => {
  let testScheduler;

  beforeEach(() => {
    testScheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it("should let you test snapshots of streams that do not complete", () => {
    testScheduler.run((helpers) => {
      const { expectObservable } = helpers;
      const source$ = interval(1000).pipe(map((val) => `${val + 1}sec`));
      const expected = "1s a 999ms b 999ms c";
      const unsubscribe = "3999ms !";

      expectObservable(source$, unsubscribe).toBe(expected, {
        a: "1sec",
        b: "2sec",
        c: "3sec",
      });
    });
  });

原文地址：http://www.cnblogs.com/Answer1215/p/16792887.html