### Question:
RxShell { public synchronized Single<Session> open() { if (RXSDebug.isDebug()) Timber.tag(TAG).v("open()"); if (session == null) { session = rxProcess.open() .map(session -> { OutputStreamWriter writer = new OutputStreamWriter(session.input(), StandardCharsets.UTF_8); return new Session(session, writer); }) .subscribeOn(Schedulers.io()) .doOnSuccess(s -> { if (RXSDebug.isDebug()) Timber.tag(TAG).v("open():doOnSuccess %s", s); s.waitFor().subscribe(integer -> { synchronized (RxShell.this) { session = null; } }, e -> Timber.tag(TAG).w(e, "Error resetting session.")); }) .doOnError(t -> { if (RXSDebug.isDebug()) Timber.tag(TAG).v(t, "open():doOnError");}) .cache(); } return session; } RxShell(RxProcess rxProcess); synchronized Single<Session> open(); synchronized Single<Boolean> isAlive(); synchronized Completable cancel(); synchronized Single<Integer> close();  }### Answer:
@Test public void testOpen_error() throws IOException { doReturn(Single.error(new InterruptedException())).when(rxProcess).open(); RxShell rxShell = new RxShell(rxProcess); TestObserver<RxShell.Session> sessionObs = rxShell.open().test(); sessionObs.awaitDone(1, TimeUnit.SECONDS).assertError(InterruptedException.class); sessionObs.assertNoValues(); }
@Test public void testWaitFor() throws InterruptedException { RxShell rxShell = new RxShell(rxProcess); RxShell.Session session = rxShell.open().test().awaitDone(1, TimeUnit.SECONDS).assertNoErrors().values().get(0); assertThat(session.waitFor().test().await(1, TimeUnit.SECONDS), is(false)); waitForEmitter.onSuccess(55); session.waitFor().test().awaitDone(1, TimeUnit.SECONDS).assertNoErrors().assertValue(55); }