RxAndroid,Retrofit 2单元测试Schedulers.io

Question

我刚学会了RxAndroid,但不幸的是我研究过的这本书没有涵盖任何单元测试.我在谷歌上搜索了很多,但没有找到任何简单的教程,以精确的方式涵盖RxAndroid单元测试.

我基本上使用RxAndroid和Retrofit 2编写了一个小的REST API.这是ApiManager类:

public class MyAPIManager {
    private final MyService myService;

    public MyAPIManager() {
        HttpLoggingInterceptor logging = new HttpLoggingInterceptor();
        // set your desired log level
        logging.setLevel(HttpLoggingInterceptor.Level.BODY);

        OkHttpClient.Builder b = new OkHttpClient.Builder();
        b.readTimeout(35000, TimeUnit.MILLISECONDS);
        b.connectTimeout(35000, TimeUnit.MILLISECONDS);
        b.addInterceptor(logging);
        OkHttpClient client = b.build();

        Retrofit retrofit = new Retrofit.Builder()
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .baseUrl("http://192.168.1.7:8000")
                .client(client)
                .build();

        myService = retrofit.create(MyService.class);
    }

    public Observable<Token> getToken(String username, String password) {
        return myService.getToken(username, password)
                .subscribeOn(Schedulers.io());
                .observeOn(AndroidSchedulers.mainThread());
    }
}

我正在尝试创建一个单元测试getToken.这是我的样本测试:

public class MyAPIManagerTest {
    private MyAPIManager myAPIManager;
    @Test
    public void getToken() throws Exception {
        myAPIManager = new MyAPIManager();

        Observable<Token> o = myAPIManager.getToken("hello", "mytoken");
        o.test().assertSubscribed();
        o.test().assertValueCount(1);
    }

}

由于subscribeOn(Schedulers.io)上面的测试没有在主线程上运行,因为它返回0值.如果我删除subscribeOn(Schedulers.io)从MyAPIManager那么它运行良好,并返回值1.有没有办法测试Schedulers.io？

Answer 1

很好的问题,当然还有一个在社区中缺乏大量报道的话题.我想分享一些我个人使用的解决方案并且非常精彩.这些被认为是RxJava 2,但它们可以在不同的名称下使用RxJava 1.如果需要,您肯定会找到它.

1. RxPlugins和RxAndroidPlugins(这是我最喜欢的)

所以实际上的Rx提供了一种机制,以改变由内而外的静态方法提供的调度Schedulers和AndroidSchedulers.这些是例如:

RxJavaPlugins.setComputationSchedulerHandler
RxJavaPlugins.setIoSchedulerHandler
RxJavaPlugins.setNewThreadSchedulerHandler
RxJavaPlugins.setSingleSchedulerHandler
RxAndroidPlugins.setInitMainThreadSchedulerHandler

这些做的很简单.它们确保在您调用时,即Schedulers.io()返回的调度程序是您在处理程序集中提供的调度程序setIoSchedulerHandler.您想使用哪个调度程序？好吧你想要的Schedulers.trampoline().这意味着代码将在以前的相同线程上运行.如果所有调度程序都在trampoline调度程序中,那么所有调度程序都将在该JUnit线程上运行.运行测试后,您可以通过调用以下方法清除整个测试:

RxJavaPlugins.reset()
RxAndroidPlugins.reset()

我认为最好的方法是使用JUnit规则.这是一个可能的(抱歉kotlin语法):

class TrampolineSchedulerRule : TestRule {
  private val scheduler by lazy { Schedulers.trampoline() }

  override fun apply(base: Statement?, description: Description?): Statement =
        object : Statement() {
            override fun evaluate() {
                try {
                    RxJavaPlugins.setComputationSchedulerHandler { scheduler }
                    RxJavaPlugins.setIoSchedulerHandler { scheduler }
                    RxJavaPlugins.setNewThreadSchedulerHandler { scheduler }
                    RxJavaPlugins.setSingleSchedulerHandler { scheduler }
                    RxAndroidPlugins.setInitMainThreadSchedulerHandler { scheduler }
                    base?.evaluate()
                } finally {
                    RxJavaPlugins.reset()
                    RxAndroidPlugins.reset()
                }
            }
        }
}

在单元测试的顶部,您只需要声明一个@Rule使用此类注释并使用此类实例化的公共属性:

@Rule
public TrampolineSchedulerRule rule = new TrampolineSchedulerRule()

在kotlin

@get:Rule
val rule = TrampolineSchedulerRule()

2. 注入调度程序(又称依赖注入)

另一种可能性是在您的类中注入调度程序,因此在测试时您可以再次注入,Schedulers.trampoline()并且在您的应用程序中,您可以注入正常的调度程序.这可能会有一段时间,但是当你需要为一个简单的类注入大量的调度程序时,它很快就会变得很麻烦.这是一种做到这一点的方法

public class MyAPIManager {
  private final MyService myService;
  private final Scheduler io;
  private final Scheduler mainThread;

  public MyAPIManager(Scheduler io, Scheduler mainThread) {
    // initialise everything
    this.io = io;
    this.mainThread = mainThread;
  }

  public Observable<Token> getToken(String username, String password) {
    return myService.getToken(username, password)
            .subscribeOn(io);
            .observeOn(mainThread);
  }
}

如您所见,我们现在可以告诉班级实际的调度程序.在你的测试中,你会做类似的事情:

public class MyAPIManagerTest {
  private MyAPIManager myAPIManager;
  @Test
  public void getToken() throws Exception {
    myAPIManager = new MyAPIManager(
          Schedulers.trampoline(),
          Schedulers.trampoline());

    Observable<Token> o = myAPIManager.getToken("hello", "mytoken");
    o.test().assertSubscribed();
    o.test().assertValueCount(1);
  }
}

关键点是:

• 您希望它在Schedulers.trampoline()调度程序上确保所有内容都在JUnit线程上运行

• 您需要能够在测试时修改调度程序.

就这样.希望能帮助到你.

================================================== =======

这是我在以上Kotlin示例之后使用的Java版本:

public class TrampolineSchedulerRule implements TestRule {
    @Override
    public Statement apply(Statement base, Description description) {
        return new MyStatement(base);
    }

    public class MyStatement extends Statement {
        private final Statement base;

        @Override
        public void evaluate() throws Throwable {
            try {
                RxJavaPlugins.setComputationSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setIoSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setNewThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxJavaPlugins.setSingleSchedulerHandler(scheduler -> Schedulers.trampoline());
                RxAndroidPlugins.setInitMainThreadSchedulerHandler(scheduler -> Schedulers.trampoline());
                base.evaluate();
            } finally {
                RxJavaPlugins.reset();
                RxAndroidPlugins.reset();
            }
        }

        public MyStatement(Statement base) {
            this.base = base;
        }
    }
}