Example using redux-observable + typescript + TestScheduler, with a run helper written

redux-observable-typescript-testing-example.ts

import { map, delay } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';
import { Action } from 'redux';
import { Epic, ofType, ActionsObservable, StateObservable } from 'redux-observable';

const scheduler = new TestScheduler((actual, expected) => {
  if (JSON.stringify(actual) !== JSON.stringify(expected)) {
    throw new Error(`Failing test

actual: ${JSON.stringify(actual, null, 2)}

expected: ${JSON.stringify(expected, null, 2)}
    `);
  }
});

type State = {
  counter: number
};

const PING = 'PING';
const PONG = 'PONG';

interface PingAction {
  type: typeof PING
}

interface PongAction {
  type: typeof PONG,
  payload: number
}

const ping = (): PingAction => ({ type: PING });
const pong = (counter: number): PongAction => ({ type: PONG, payload: counter });

type Actions = PingAction | PongAction;

const fetchEpic: Epic<Actions, PongAction, State> = (action$, state$) =>
  action$.pipe(
    ofType<PingAction>(PING),
    delay(1000),
    map(() => pong(state$.value.counter))
  );

// The callback provided to scheduler.run(callback) uses accepts
// the single arugment "RunHelpers":
//   run<T>(callback: (helpers: RunHelpers) => T): T
//   https://github.com/ReactiveX/rxjs/blob/0d2025540290bd69637247238595302f7ceb5f9b/src/internal/testing/TestScheduler.ts#L13
// but this type is not currently publicly exported, though it prolly
// should be. So we have to do TS voodoo to extract it.
type ArgumentTypes<F extends Function> =
  F extends (...args: infer A) => any ? A : never;
type RunArguments = ArgumentTypes<typeof scheduler.run>;
type RunCallback = RunArguments[0];
type RunCallbackArguments = ArgumentTypes<RunCallback>;
type RunHelpers = RunCallbackArguments[0];
type hotActions = <T extends Action>(marbles: string, values?: { [marble: string]: T }, error?: any) => ActionsObservable<T>;
type hotState = <T = string>(marbles: string, values?: { [marble: string]: T }, error?: any) => StateObservable<T>;
type EpicRunnerRunHelpers = RunHelpers & { hotActions: hotActions, hotState: hotState };

const epicRunner = <T>(scheduler: TestScheduler, callback: (callback: EpicRunnerRunHelpers) => T): T => {
  return scheduler.run(({ hot, ...rest }) => {
    const hotActions: hotActions = <T extends Action>(marbles, values) => {
      const actionInput$ = hot<T>(marbles, values);
      return new ActionsObservable(actionInput$);
    };

    const hotState: hotState = <T>(marbles, values, initialState) => {
      const stateInput$ = hot<T>(marbles, values);
      return new StateObservable<T>(stateInput$, initialState);
    };

    return callback({ hot, hotActions, hotState, ...rest });
  });
};


epicRunner(scheduler, ({ hotActions, hotState, expectObservable }) => {
  const action$ = hotActions('-a', { a: ping() });
  const state$ = hotState('-a', { a: { counter: 1 } }, { counter: 0 });
  const output$ = fetchEpic(action$, state$, null);

  expectObservable(output$).toBe('- 1s a', { a: pong(1) });
});

document.body.innerText = 'success!';