带timeout操作符的RxJava单元测试未触发subscribe订阅问题

• 调度器异步问题：生产代码里用到的Schedulers.io()、AndroidSchedulers.mainThread()以及timeout指定的调度器默认都是异步调度，单元测试运行在主线程，测试执行完成时异步任务还没触发，所以看不到回调执行。
• 订阅时机错误：你在调用viewModel.loadData()之后才调用viewModel.onShowError().test()订阅，doOnSubscribe里设置的onShowError.value = false已经发射过了，订阅时无法收到历史值，就算是粘性的可观察对象，异步调度也会导致值还没发射就开始等待，最终超时无结果。
• awaitNextValue()会阻塞等待下一个值，上游没有触发发射的话就会一直卡住，无法拿到结果。

1. 给ViewModel注入调度器，不要硬编码调度器实现
   将调度器作为ViewModel构造参数传入，生产环境用默认的正式调度器，测试环境传入同步测试调度器：

class MyViewModel(
    private val loadDataUseCase: LoadDataUseCase,
    private val ioScheduler: Scheduler = Schedulers.io(),
    private val mainScheduler: Scheduler = AndroidSchedulers.mainThread(),
    private val timeoutScheduler: Scheduler = Schedulers.computation() // 对应你原代码timeout方法传入的schedulers
) : ViewModel()

修改loadData方法调用注入的调度器：

fun loadData() {
    loadDataUseCase.loadData(true)
        .subscribeOn(ioScheduler)
        .timeout(30L, TimeUnit.SECONDS, timeoutScheduler)
        .observeOn(mainScheduler)
        // 剩余原有逻辑保持不变
        .doOnSubscribe {
            onShowLoading.value = true
            onShowError.value = false
            onShowContent.value = false
        }.subscribe(
            {
                onConnected.value = true
                onShowContent.value = true
                onShowError.value = false
                onShowLoading.value = false
            },
            {
                onShowError.value = true
                onShowLoading.value = false
                onShowContent.value = false
            }
        )
        .addTo(compositeDisposable)
}

2. 调整测试代码逻辑
   测试时用同步的TrampolineScheduler让所有任务都在当前线程立即执行，同时调整订阅时机，先订阅再触发业务逻辑：

// 如果你的onShowError是LiveData，需要添加这个Rule让LiveData同步执行
@get:Rule
val instantTaskExecutorRule = InstantTaskExecutorRule()

@Test
fun `Load data is success`() {
    // given 构造ViewModel时传入同步调度器
    val testViewModel = MyViewModel(
        loadDataUseCase = loadDataUseCase,
        ioScheduler = Schedulers.trampoline(),
        mainScheduler = Schedulers.trampoline(),
        timeoutScheduler = Schedulers.trampoline()
    )
    whenever(loadDataUseCase.loadData(true)).thenReturn(Observable.just(true))
    // 先订阅事件，再执行业务逻辑
    val errorTestObserver = testViewModel.onShowError().test()

    // when
    testViewModel.loadData()

    // then 按顺序断言两次false的值：第一次是doOnSubscribe设置的，第二次是成功回调设置的
    errorTestObserver.assertValues(false, false)
}

如果需要测试超时场景，可以用TestScheduler手动推进时间：

val testScheduler = TestScheduler()
// 构造ViewModel时所有调度器都传入testScheduler
testViewModel.loadData()
// 手动推进时间到30秒，触发timeout逻辑
testScheduler.advanceTimeBy(30, TimeUnit.SECONDS)
// 断言超时后的错误状态

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