单元测试 RxJava

Question

我正在Android中开发一个单元测试来测试与Socket.IO交互的RxJava2代码。当我从MainActivity()运行时，代码运行良好。在单元测试中，我使用以下代码来测试Observable：

ObservableFactory.sharedInstance.connectionStatusObservable.blockingForEach(connectionStatus -> {
    System.out.println(connectionStatus.getStatus());
});

预期输出如下：

Connecting
Connected

但它实际上在显示Connecting状态后NullPointerException陷入EventThread：

Connecting
Jan 29, 2018 6:57:15 PM io.socket.thread.EventThread$2 run
SEVERE: Task threw exception
java.lang.NullPointerException
    at io.socket.engineio.client.HandshakeData.<init>(HandshakeData.java:21)
    at io.socket.engineio.client.HandshakeData.<init>(HandshakeData.java:16)
    at io.socket.engineio.client.Socket.onPacket(Socket.java:539)
    at io.socket.engineio.client.Socket.access$1000(Socket.java:36)
    at io.socket.engineio.client.Socket$5.call(Socket.java:335)
    at io.socket.emitter.Emitter.emit(Emitter.java:117)
    at io.socket.engineio.client.Transport.onPacket(Transport.java:126)
    at io.socket.engineio.client.transports.Polling.access$700(Polling.java:18)
    at io.socket.engineio.client.transports.Polling$2.call(Polling.java:127)
    at io.socket.engineio.parser.Parser.decodePayload(Parser.java:241)
    at io.socket.engineio.client.transports.Polling._onData(Polling.java:135)
    at io.socket.engineio.client.transports.Polling.onData(Polling.java:102)
    at io.socket.engineio.client.transports.PollingXHR$5$1.run(PollingXHR.java:125)
    at io.socket.thread.EventThread$2.run(EventThread.java:80)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Exception in thread "EventThread" java.lang.NullPointerException
    at io.socket.engineio.client.HandshakeData.<init>(HandshakeData.java:21)
    at io.socket.engineio.client.HandshakeData.<init>(HandshakeData.java:16)
    at io.socket.engineio.client.Socket.onPacket(Socket.java:539)
    at io.socket.engineio.client.Socket.access$1000(Socket.java:36)
    at io.socket.engineio.client.Socket$5.call(Socket.java:335)
    at io.socket.emitter.Emitter.emit(Emitter.java:117)
    at io.socket.engineio.client.Transport.onPacket(Transport.java:126)
    at io.socket.engineio.client.transports.Polling.access$700(Polling.java:18)
    at io.socket.engineio.client.transports.Polling$2.call(Polling.java:127)
    at io.socket.engineio.parser.Parser.decodePayload(Parser.java:241)
    at io.socket.engineio.client.transports.Polling._onData(Polling.java:135)
    at io.socket.engineio.client.transports.Polling.onData(Polling.java:102)
    at io.socket.engineio.client.transports.PollingXHR$5$1.run(PollingXHR.java:125)
    at io.socket.thread.EventThread$2.run(EventThread.java:80)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我还尝试在测试开始时使用 TrampolineScheduler 作为规则：

@Rule
public TrampolineSchedulerRule trampolineSchedulerRule = new TrampolineSchedulerRule();

但是我仍然收到上述错误消息。如果我运行以下测试，那么它运行良好：

ConnectionStatus status = ObservableFactory.sharedInstance.connectionStatusObservable.blockingFirst();
assertTrue(status.getStatus() == ConnectionStatus.Status.Connecting);

问题出在Connected状态中，它成功地与Socket.IO建立了连接，但不知何故，即使它在MainActivity()中运行良好，单元测试也会在其上抛出错误。

知道我做错了什么吗？

Answer 1

在对 RxJava 观察器链进行单元测试时，您经常会遇到调度程序的困难。最常见的问题是引入死锁，其中测试正在等待永远不会到达的结果，因为它阻塞了生成结果的同一线程。在测试网络或操作系统服务时，您经常会与服务中嵌入的异步代码发生冲突。在您的情况下，蹦床调度程序导致使用尚未实际设置的变量，因为网络操作尚未完成。

您将需要远离蹦床调度程序(我知道 RxJava2 并不真正支持它(，并使用特定的调度程序，Schedulers.io()或Schedulers.computation()。

下一部分是使用TestSubscriber()来实际检查观察者链是否完成了它的工作。

Observable<ConnectionStatus> obs = 
  ObservableFactory.sharedInstance.connectionStatusObservable
  .subscribeOn( Schedulers.io() );
TestSubscriber ts = obs.test();
ts.awaitTerminalEvent();
ts.assertCompleted();
ts.assertValues( Connecting, Connected );