### Question:
RxCmdShell { public synchronized Single<Session> open() { if (RXSDebug.isDebug()) Timber.tag(TAG).v("open()"); if (session == null) { session = Single .create((SingleOnSubscribe<Session>) emitter -> rxShell.open().subscribe(new SingleObserver<RxShell.Session>() { @Override public void onSubscribe(Disposable d) { } @Override public void onSuccess(RxShell.Session shellSession) { try { final Iterator<Map.Entry<String, String>> envIterator = environment.entrySet().iterator(); while (envIterator.hasNext()) { final Map.Entry<String, String> entry = envIterator.next(); shellSession.writeLine(entry.getKey() + "=" + entry.getValue(), !envIterator.hasNext()); } } catch (IOException e) { emitter.tryOnError(e); return; } CmdProcessor cmdProcessor = processorFactory.create(); cmdProcessor.attach(shellSession); final Session cmdShellSession = new Session(shellSession, cmdProcessor); emitter.onSuccess(cmdShellSession); } @Override public void onError(Throwable e) { Timber.tag(TAG).w("Failed to open RxShell session!"); synchronized (RxCmdShell.this) { session = null; } emitter.tryOnError(e); } })) .subscribeOn(Schedulers.io()) .doOnSuccess(s -> { if (RXSDebug.isDebug()) Timber.tag(TAG).v("open():doOnSuccess %s", s); s.waitFor().subscribe(integer -> { synchronized (RxCmdShell.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; } @SuppressWarnings("unused") private  RxCmdShell();   RxCmdShell(Builder builder, RxShell rxShell); synchronized Single<Session> open(); Single<Boolean> isAlive(); synchronized Completable cancel(); synchronized Single<Integer> close(); static Builder builder();  }### Answer:
@Test public void testOpen() { RxCmdShell rxCmdShell = new RxCmdShell(builder, rxShell); rxCmdShell.open().test().awaitCount(1).assertNoErrors(); verify(rxShell).open(); }
@Test public void testOpen_exception() { doReturn(Single.error(new IOException())).when(rxShell).open(); RxCmdShell rxCmdShell = new RxCmdShell(builder, rxShell); rxCmdShell.open().test().awaitDone(1, TimeUnit.SECONDS).assertError(IOException.class); verify(rxShell).open(); verify(rxShellSession, never()).outputLines(); verify(rxShellSession, never()).errorLines(); }
@Test public void testReinit_exception() { RxCmdShell shell = new RxCmdShell(builder, rxShell); doReturn(Single.error(new IOException())).when(rxShell).open(); shell.open().test().awaitDone(1, TimeUnit.SECONDS).assertError(IOException.class); doReturn(Single.just(rxShellSession)).when(rxShell).open(); shell.open().test().awaitDone(1, TimeUnit.SECONDS).assertNoErrors().assertValueCount(1); }
@Test public void testEnvironmentSetting() throws IOException { final HashMap<String, String> envMap = new HashMap<>(); envMap.put("key", "value"); when(builder.getEnvironment()).thenReturn(envMap); RxCmdShell shell = new RxCmdShell(builder, rxShell); shell.open().test().awaitCount(1).assertNoErrors(); verify(rxShellSession).writeLine("key=value", true); }
@Test public void testWaitFor() throws InterruptedException { RxCmdShell shell = new RxCmdShell(builder, rxShell); RxCmdShell.Session session = shell.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); }