Android单元测试中如何测试Observable与Subscriber模式代码

核心解决两个问题：异步调度器的隔离、Observable执行结果的捕获与验证，可按以下步骤实现：

1. 先替换测试环境的调度器 #

你代码中使用了schedulers.main()主线程调度器，单元测试环境无Android主线程，需要统一替换为测试可控的调度器：

• 测试类初始化前用RxJava钩子替换全局调度器：

@Before
fun setupSchedulers() {
    // 替换IO、计算、主线程所有调度器为同一个TestScheduler，方便统一控制
    val testScheduler = TestScheduler()
    RxJavaPlugins.setIoSchedulerHandler { testScheduler }
    RxJavaPlugins.setComputationSchedulerHandler { testScheduler }
    RxJavaPlugins.setMainThreadSchedulerHandler { testScheduler }
}

@After
fun resetSchedulers() {
    // 测试结束后重置钩子，避免影响其他测试用例
    RxJavaPlugins.reset()
}

2. 模拟Api接口返回 #

用Mockito/Mockk模拟api.loadData的返回值，按需构造正常返回/异常返回的Observable：

// 模拟正常返回场景
every { api.loadData(any()) } returns Observable.just(testData)
// 模拟异常返回场景
// every { api.loadData(any()) } returns Observable.error(RuntimeException("test error"))

3. 两种测试方式可选 #

方案1：使用TestObserver直接订阅验证（推荐） #

官方提供的TestObserver就是封装好的模拟Subscriber实现，无需自定义即可完成结果验证：

val testObserver = api.loadData(request)
    .observeOn(schedulers.main())
    .doOnNext(this::addDataCheck)
    .test()

// 触发所有待执行的异步任务
testScheduler.triggerActions()

// 验证结果
testObserver.assertComplete()
testObserver.assertValue(testData)
testObserver.assertNoErrors()

方案2：验证回调触发场景 #

如果需要验证onDataLodaded、onAdFailedToLoad回调是否正常触发，可以用CountDownLatch等待回调执行，再验证回调参数：

// 模拟listener
Listener mockListener = Mockito.mock(Listener.class);
CountDownLatch latch = new CountDownLatch(1);

// 执行原业务代码
api.loadData(request)
    .observeOn(schedulers.main())
    .doOnNext(this::addDataCheck)
    .subscribe(
        data -> {
            onDataLodaded(mockListener, data);
            latch.countDown();
        },
        e -> {
            onAdFailedToLoad(mockListener, Api.error(request), e);
            latch.countDown();
        }
    )
    .addTo(disposables);

// 触发异步任务执行
testScheduler.triggerActions();
// 最多等待1秒，避免测试卡死
latch.await(1, TimeUnit.SECONDS);

// 验证回调是否被正确调用
Mockito.verify(mockListener).onSuccess(any());

内容的提问来源于stack exchange，提问作者amodkanthe