Testing Observables with Marble Diagrams

Hello! Welcome back to our journey of building a reactive library.

In the last lesson, we built a crucial piece of our testing infrastructure: the parseMarbles function. It allows us to translate expressive marble strings into the TestNotification arrays our TestScheduler understands. This was a big step towards more readable and declarative tests. However, our test code still involves a manual step of calling the parser and passing the result to our assertion.

Today, we will build the final, elegant API that connects all the pieces. This lesson directly addresses the learning outcome: Create an assertion helper to test an observable's output against an expected marble diagram. By the end of this lesson, you will be able to write a test as concisely as this:

expectObservable(myStream).toBe('--a--b|', { a: 10, b: 20 });

This is the standard for testing reactive streams, and having it in our toolkit will make testing our operators a robust and pleasant experience.

1. The Anatomy of a Marble Assertion

Before we jump into code, let's look at the API we're aiming to build. The industry standard, established by RxJS, is a great model to follow.

This API elegantly splits the assertion process into two logical steps, which our implementation will mirror:

1. Recording (expectObservable): This step tells the test runner which observable's output we want to capture. Our helper will subscribe to this observable within the TestScheduler's virtual time and record every notification (next, error, complete) along with its virtual timestamp into an actual notifications array.

2. Asserting (.toBe): This step defines what the output should look like, using a marble diagram. Our helper will use the parseMarbles function from our previous lesson to convert this string into an expected notifications array. It will then schedule a final assertion to compare the actual and expected arrays after all virtual time has passed.

With this mental model, let's implement it.

2. Implementing the expectObservable Helper

We will modify our TestScheduler.ts file to create the new helper. The expectObservable function needs to be defined inside the run method, because it requires access to the scheduler's context (like the current frame and the assertions queue).

Here is the updated TestScheduler.run method. I've included comments to explain each part of the new expectObservable logic.

// src/testing/TestScheduler.ts

import { Observable } from '../core/Observable';
import { Scheduler } from '../types';
import { Action, Subscription } from '../core/Subscription';
import { parseMarbles, TestNotification } from './marble-parser';

export class TestScheduler implements Scheduler {
  // ... (previous properties: frame, actions, assertions, assertDeepEqual)

  // ... (previous methods: constructor, schedule)
  
  public flush(): void {
    // ... (flush implementation remains the same)
  }

  public run(callback: (helpers: {
    expectObservable: <T>(source: Observable<T>) => {
      toBe: (marbles: string, values?: { [key: string]: T }, error?: any) => void;
    };
  }) => void) {
    // 1. Reset scheduler state for a clean run
    this.frame = 0;
    this.actions = [];
    this.assertions = [];
    
    // 2. Define and provide the `expectObservable` helper
    const helpers = {
      expectObservable: <T>(source: Observable<T>) => {
        // This array will store the actual notifications from the source observable.
        const actual: TestNotification<T>[] = [];
        
        // Return an object with the .toBe method to chain the assertion.
        return {
          toBe: (marbles: string, values?: { [key: string]: T }, errorValue?: any) => {
            // -- RECORDING PHASE SETUP --
            // When .toBe is called, we schedule the subscription to the source.
            // This ensures that subscription happens at the very beginning of virtual time (frame 0).
            this.schedule(() => {
              source.subscribe({
                next: (value: T) => {
                  // When a value is emitted, record it with the current virtual frame.
                  actual.push({ frame: this.frame, kind: 'N', value });
                },
                error: (err: any) => {
                  // Record an error notification.
                  actual.push({ frame: this.frame, kind: 'E', error: err });
                },
                complete: () => {
                  // Record a complete notification.
                  actual.push({ frame: this.frame, kind: 'C' });
                },
              });
            }, 0);

            // -- ASSERTION PHASE SETUP --
            // Use our parser from the last lesson to get the expected notifications.
            const expected = parseMarbles(marbles, values, errorValue);

            // Schedule the final assertion. This function will be called by flush()
            // *after* all other actions have been executed.
            this.assertions.push(() => this.assertDeepEqual(actual, expected));
          },
        };
      },
    };

    // 3. Execute the user's test logic with the provided helpers
    callback(helpers);

    // 4. Flush the scheduler to run all actions and assertions
    this.flush();
  }
}

Let's break down the most important part of this implementation: the timing.

• The user calls expectObservable(source).toBe(...) inside the run callback. This happens synchronously before virtual time begins.
• The call to .toBe does two things:
  1. It schedules the subscription to the source observable to happen at frame: 0.
  2. It schedules the final assertion (comparing actual vs expected) to happen after all frames have been processed.
• When run calls this.flush(), the simulation starts. The subscription action is executed at frame 0. As the source observable emits values (perhaps due to scheduled delays), our observer records them into the actual array, capturing the this.frame at the moment of emission.
• After all scheduled actions are complete, flush runs the assertion we pushed, comparing the fully populated actual array with the expected array from the marble diagram.

This ensures that we capture the entire lifecycle of the observable within the virtual timeline and assert against it correctly.

scheduler.run(({ expectObservable }) => {
  const source = of('a', 'b'); // Emits 'a', 'b', then completes synchronously
  const stream = source.pipe(map(char => char.toUpperCase()));
  
  expectObservable(stream).toBe('(AB|)');
});

[
  { "frame": 0, "kind": "N", "value": "A" },
  { "frame": 0, "kind": "N", "value": "B" },
  { "frame": 0, "kind": "C" }
]

3. Finalizing Our Test with the New Helper

Now for the rewarding part. Let's refactor the delay operator test from our previous lesson to use our new, polished assertion helper.

See how the test becomes a clean, declarative statement about the observable's behavior.

// src/operators/delay.test.ts (Final Version)
import { describe, it, expect } from 'vitest';
import { of } from '../creators/of';
import { TestScheduler } from '../testing/TestScheduler';
import { delay } from './delay';

describe('delay', () => {
  it('should delay emissions by the specified duration', () => {
    // 1. Create the scheduler with an assertion function from our test framework
    const scheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });

    // 2. Call run() to start the test in a virtual time context
    scheduler.run(({ expectObservable }) => {
      // 3. Define the source observable and the operator chain to test
      const source = of(1);
      const delayTime = 200;
      const delayedStream = source.pipe(delay(delayTime, scheduler));

      // 4. Declare the expected behavior using a marble diagram
      const expectedMarbles = '200ms (a|)';
      const expectedValues = { a: 1 };

      // 5. Assert that the stream behaves as expected
      expectObservable(delayedStream).toBe(expectedMarbles, expectedValues);
    });
  });
});

This is a professional-grade testing pattern. The test is easy to read, clearly states the input (of(1)), the transformation (delay(200)), and the expected output ('200ms (a|)'). We have successfully abstracted away all the implementation details of notification arrays and manual scheduling.

Conclusion

Excellent work today! You have now implemented the centerpiece of a reactive testing library: a beautiful and ergonomic expectObservable(...).toBe(...) assertion helper. By combining our TestScheduler, parseMarbles function, and this new helper, you have a complete, powerful system for testing complex asynchronous logic deterministically.

Key Takeaways:

• Assertion API Design: You've implemented a fluent API (expectObservable(...).toBe(...)) that mirrors the industry standard set by RxJS, promoting readability.
• Recording and Asserting: The process is split into two phases: scheduling a subscription to record actual events and scheduling an assertion to compare them against expected events from a marble diagram.
• Declarative Testing: Your tests now clearly declare what behavior is expected, rather than imperatively describing how to check for it. This makes tests easier to write, read, and maintain.

In our next lesson, we'll start looking outward. Now that we have a solid foundation for our own library, we will explore interoperability. We will implement adapters to convert between your library's observables and RxJS observables, a crucial step for ensuring your library can be adopted in projects that already use the RxJS ecosystem.