RxJava单元测试：时间推进修改interval响应测试pollForState3

核心逻辑是替换所有异步调度器为RxJava提供的TestScheduler——这个调度器不会真的异步执行任务，完全支持手动推进时间、触发排队任务，配合Mock接口动态返回响应，就能精确验证每一步轮询逻辑。

前置准备 #

• 测试框架选JUnit即可，Mock层用Mockito或者MockK都可以，核心依赖是RxJava自带的测试工具包，里面包含TestScheduler和TestObserver两个核心类。
• 你代码里自定义的SchedulerProvider是天然的测试替换入口，不需要修改任何业务代码。

测试步骤 #

• 初始化TestScheduler实例，实现测试用的SchedulerProvider，让所有io()（以及你用到的其他调度器）都返回这个TestScheduler实例，注入到待测试的Repository中。
• MockApi接口的getStatus()方法，准备按顺序返回的响应队列：前N次返回对应State1、State2的响应，最后一次返回对应State3的响应，模拟接口状态随时间变化的场景。
• 订阅pollForState3()，调用RxJava流的test()方法拿到TestObserver实例，这个对象提供了完整的事件断言能力，可以检查收到的值、流是否完成、是否报错、接口调用次数等。
• 按轮询的时间节奏手动推进TestScheduler的时间，每推进一段就做一次断言，不要一次性跳到结束时间，逐段验证才能定位轮询间隔、请求时机的问题。
• 验证终止逻辑：拿到State3之后，再多推进几秒时间，确认不会再发起新的接口请求，验证takeUntil确实终止了上游轮询。

参考测试代码 #

import io.reactivex.rxjava3.schedulers.TestScheduler
import io.reactivex.rxjava3.kotlin.test
import org.junit.Test
import java.util.concurrent.TimeUnit
import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.mockk

class RepositoryTest {
    // 测试专用可控调度器
    private val testScheduler = TestScheduler()
    // 测试用调度器提供者，所有调度逻辑都走testScheduler
    private val testSchedulerProvider = object : SchedulerProvider {
        override fun io() = testScheduler
        // 如果你的SchedulerProvider还有ui、computation等方法，全部返回testScheduler即可
    }
    // Mock接口实例
    private val mockApi: Api = mockk()
    // 初始化待测试的Repository
    private val repository = Repository(mockApi, testSchedulerProvider)

    @Test
    fun `verify pollForState3 works correctly`() {
        // 准备接口返回队列：依次返回State1、State2、State3对应的响应
        val responseQueue = ArrayDeque<Response>().apply {
            add(Response(code = 1)) // 业务map逻辑转换后为State1
            add(Response(code = 2)) // 业务map逻辑转换后为State2
            add(Response(code = 3)) // 业务map逻辑转换后为State3
        }
        // 打桩：每次调用接口从队列头取一个响应返回
        coEvery { mockApi.getStatus() } coAnswers { responseQueue.removeFirst() }

        // 订阅目标轮询流
        val testObserver = repository.pollForState3().test()

        // 初始状态：还没触发调度动作，断言没有收到任何值，流未完成
        testObserver.assertNoValues()
        testObserver.assertNotComplete()

        // 触发startWith(0)对应的初始请求（订阅后立刻排队的0延迟任务）
        testScheduler.triggerActions()
        // 断言第一次返回State1，流未终止
        testObserver.assertValueAt(0) { it is State1 }
        testObserver.assertNotComplete()

        // 推进1秒时间，触发interval的第一次定时任务
        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS)
        // 断言第二次返回State2，流未终止
        testObserver.assertValueAt(1) { it is State2 }
        testObserver.assertNotComplete()

        // 再推进1秒时间，触发interval的第二次定时任务
        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS)
        // 断言第三次返回State3，流正常终止，无错误
        testObserver.assertValueAt(2) { it is State3 }
        testObserver.assertComplete()
        testObserver.assertNoErrors()

        // 多推进5秒，验证takeUntil生效后轮询停止
        testScheduler.advanceTimeBy(5, TimeUnit.SECONDS)
        // 断言总共只收到3个值，没有多余事件
        testObserver.assertValueCount(3)
        // 断言接口总共只调用了3次，没有多余请求
        coVerify(exactly = 3) { mockApi.getStatus() }
    }
}

注意避坑 #

• 不要在测试中使用真实的IO、computation调度器，否则测试会依赖真实时间等待，执行慢还容易出现时序波动导致的偶发失败。
• 初始的startWith(0)是订阅后立刻在调度器上排队的任务，第一次验证前要先调用triggerActions()触发当前排队的所有任务，不要直接上来就推进时间，否则会漏掉第一次请求的断言。
• 必须验证轮询终止后的行为，很多轮询实现会出现takeUntil触发后上游没正确取消、还在后台发请求的内存泄漏/无用请求问题，多推进时间加接口调用次数断言就能覆盖这个场景。

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