test("Informs about disconnected kernels, allows reconnection", async () => {
   const state$ = {
     value: {
       core: stateModule.makeStateRecord({
         kernelRef: "fake",
         entities: stateModule.makeEntitiesRecord({
           contents: stateModule.makeContentsRecord({
             byRef: Immutable.Map({
               fakeContent: stateModule.makeNotebookContentRecord()
             })
           }),
           kernels: stateModule.makeKernelsRecord({
             byRef: Immutable.Map({
               fake: stateModule.makeRemoteKernelRecord({
                 channels: null,
                 status: "not connected"
               })
             })
           })
         })
       }),
       app: {
         notificationSystem: { addNotification: jest.fn() }
       }
     }
   };
   const action$ = ActionsObservable.of(
     actions.executeCell({ id: "first", contentRef: "fakeContentRef" })
   );
   const responses = await executeCellEpic(action$, state$)
     .pipe(toArray())
     .toPromise();
   expect(responses).toEqual([]);
 });

describe('add$ epic', () => {
  const action$ = ActionsObservable.of({
    type: TodoAction.Constants.ADD_ITEM,
    payload: { label: 'new' },
  });

  it('dispatches the correct action when it is successful', async () => {
    const mockResponse = [{ label: 'new', id: '789', completed: false }, ...mockData];
    const mockService: Todo.Adder = {
      addItem: () => Promise.resolve(mockResponse),
    };

    const expectedOutputActions = {
      type: TodoAction.Constants.SET_TODOS,
      payload: mockResponse,
    };

    const obs = add$(mockService)(action$);
    await expect(obs.toPromise()).resolves.toEqual(expectedOutputActions);
  });

  it('dispatches an error when the call fails', async () => {
    const mockService: Todo.Adder = {
      addItem: () => Promise.reject('error'),
    };

    const expectedOutputActions = {
      type: TodoAction.Constants.REQUEST_FAILED,
      payload: 'Add Item Failed',
    };

    const obs = add$(mockService)(action$);
    await expect(obs.toPromise()).resolves.toEqual(expectedOutputActions);
  });
});

export function executeAllCellsEpic(
  action$: ActionsObservable<ExecuteAllCells | ExecuteAllCellsBelow>,
  state$: StateObservable<AppState>
) {
  return action$.pipe(
    ofType(actions.EXECUTE_ALL_CELLS, actions.EXECUTE_ALL_CELLS_BELOW),
    concatMap((action: ExecuteAllCells | ExecuteAllCellsBelow) => {
      const state = state$.value;
      const contentRef = action.payload.contentRef;

      const model = selectors.model(state, { contentRef });
      // If it's not a notebook, we shouldn't be here
      if (!model || model.type !== "notebook") {
        return empty();
      }

      let codeCellIds = Immutable.List();

      if (action.type === actions.EXECUTE_ALL_CELLS) {
        codeCellIds = selectors.notebook.codeCellIds(model);
      } else if (action.type === actions.EXECUTE_ALL_CELLS_BELOW) {
        codeCellIds = selectors.notebook.codeCellIdsBelow(model);
      }
      return of(
        ...codeCellIds.map((id: CellId) =>
          actions.executeCell({ id, contentRef: action.payload.contentRef })
        )
      );
    })
  );
}

export const updateDisplayEpic = (
  action$: ActionsObservable<NewKernelAction>
) =>
  // Global message watcher so we need to set up a feed for each new kernel
  action$.pipe(
    ofType(actions.LAUNCH_KERNEL_SUCCESSFUL),
    switchMap((action: NewKernelAction) =>
      action.payload.kernel.channels.pipe(
        ofMessageType("update_display_data"),
        map((msg: JupyterMessage) =>
          actions.updateDisplay({
            content: msg.content,
            contentRef: action.payload.contentRef
          })
        ),
        catchError(error =>
          of(
            actions.updateDisplayFailed({
              error,
              contentRef: action.payload.contentRef
            })
          )
        )
      )
    )
  );

 test("returns an Observable with an initial state of idle", done => {
   const action$ = ActionsObservable.of({
     type: actionsModule.LAUNCH_KERNEL_SUCCESSFUL,
     payload: {
       kernel: {
         channels: of({
           header: { msg_type: "status" },
           content: { execution_state: "idle" }
         }) as Subject<any>,
         cwd: "/home/tester",
         type: "websocket"
       },
       kernelRef: "fakeKernelRef",
       contentRef: "fakeContentRef",
       selectNextKernel: false
     }
   });
   const obs = watchExecutionStateEpic(action$);
   obs.pipe(toArray()).subscribe(
     // Every action that goes through should get stuck on an array
     actions => {
       const types = actions.map(({ type }) => type);
       expect(types).toEqual([actionsModule.SET_EXECUTION_STATE]);
     },
     err => done.fail(err), // It should not error in the stream
     () => done()
   );
 });

    test("and allows continuing", async () => {
      let registeredCallback;
      ipc.once = (event, callback) => {
        if (event === "show-message-box-response") {
          registeredCallback = callback;
        }
      };
      ipc.send = (event, data) => {
        expect(event).toBe("show-message-box");
        expect(data.message).toEqual(
          "Unsaved data will be lost. Are you sure you want to quit?"
        );

        // "Yes"
        registeredCallback("dummy-event", 0);
      };

      const state = buildState(true);
      const responses = await closeNotebookEpic(
        ActionsObservable.of(
          actions.closeNotebook({ contentRef: "contentRef1" })
        ),
        { value: state }
      )
        .pipe(toArray())
        .toPromise();

      expect(responses).toEqual([
        coreActions.killKernel({ kernelRef: "kernelRef1", restarting: false }),
        actions.closeNotebookProgress({
          newState: DESKTOP_NOTEBOOK_CLOSING_READY_TO_CLOSE
        })
      ]);
    });

export const interruptKernelEpic = (
  action$: ActionsObservable<actions.InterruptKernel>,
  state$: StateObservable<AppState>
) =>
  action$.pipe(
    ofType(actions.INTERRUPT_KERNEL),
    // This epic can only interrupt kernels on jupyter websockets
    filter(() => selectors.isCurrentHostJupyter(state$.value)),
    // If the user fires off _more_ interrupts, we shouldn't interrupt the in-flight
    // interrupt, instead doing it after the last one happens
    concatMap((action: actions.InterruptKernel) => {
      const state = state$.value;

      const host = selectors.currentHost(state);
      if (host.type !== "jupyter") {
        // Dismiss any usage that isn't targeting a jupyter server
        return empty();
      }
      const serverConfig: ServerConfig = selectors.serverConfig(host);

      const kernel = selectors.currentKernel(state);
      if (!kernel) {
        return of(
          actions.interruptKernelFailed({
            error: new Error("Can't interrupt a kernel we don't have"),
            kernelRef: action.payload.kernelRef
          })
        );
      }

      if (kernel.type !== "websocket" || !kernel.id) {
        return of(
          actions.interruptKernelFailed({
            error: new Error("Invalid kernel type for interrupting"),
            kernelRef: action.payload.kernelRef
          })
        );
      }

      const id = kernel.id;

      return kernels.interrupt(serverConfig, id).pipe(
        map(() =>
          actions.interruptKernelSuccessful({
            kernelRef: action.payload.kernelRef
          })
        ),
        catchError(err =>
          of(
            actions.interruptKernelFailed({
              error: err,
              kernelRef: action.payload.kernelRef
            })
          )
        )
      );
    })
  );