Android单元测试：如何验证RxJava retryWhen调用次数及修复测试异常

这种问题我之前在写RxJava重试逻辑测试时也踩过坑！核心问题大概率是你的retryWhen延迟逻辑没有绑定到TestScheduler上，导致调用advanceTimeBy时，真正的延迟还是在真实线程里跑，而TestScheduler的时钟根本没在控制重试的等待时间。调试时因为你手动等待了真实时间，所以重试完成、断言通过，但自动化测试里时间没推进到位，订阅就还处于运行状态。

下面给你一步步排查和解决的方法：

1. 确保retryWhen的延迟逻辑绑定TestScheduler #

首先，你的重试延迟（比如每次重试等待1秒）必须显式指定使用TestScheduler，而不是默认的Scheduler。比如原来的错误写法和修正后的正确写法对比：

错误示例（无法被TestScheduler控制）

// 扩展函数里的retryWhen逻辑
fun <T> Observable<T>.retryWithMaxAttempts(maxRetries: Int): Observable<T> {
    return retryWhen { attempts ->
        attempts.zipWith(Observable.range(1, maxRetries + 1)) { error, count ->
            if (count <= maxRetries) count else throw error
        }.flatMap {
            // 这里用了默认Scheduler，TestScheduler管不到
            Observable.timer(1, TimeUnit.SECONDS)
        }
    }
}

正确示例（绑定TestScheduler）

修改扩展函数，允许传入Scheduler参数，测试时传入TestScheduler：

fun <T> Observable<T>.retryWithMaxAttempts(maxRetries: Int, scheduler: Scheduler): Observable<T> {
    return retryWhen { attempts ->
        attempts.zipWith(Observable.range(1, maxRetries + 1)) { error, count ->
            if (count <= maxRetries) count else throw error
        }.flatMap {
            // 显式指定使用传入的Scheduler
            Observable.timer(1, TimeUnit.SECONDS, scheduler)
        }
    }
}

2. 测试代码的正确写法 #

针对你需要验证的两个场景，编写绑定TestScheduler的测试用例：

场景1：触发最大重试次数后失败

@Test
fun testRetryWhenMaxRetriesReached_Fails() {
    val testScheduler = TestScheduler()
    val attemptCounter = AtomicInteger(0)

    // 模拟每次都失败的API调用
    val failingApi = Observable.defer {
        Observable.error<Response<*>>(IOException("API failed")).doOnSubscribe {
            attemptCounter.incrementAndGet()
        }
    }

    val testObserver = failingApi
        .retryWithMaxAttempts(3, testScheduler) // 最多重试3次，总计4次调用
        .test()

    // 初始状态：第一次调用失败，等待重试
    testObserver.assertNotTerminated()
    assertEquals(1, attemptCounter.get())

    // 推进总时长3秒，触发全部3次重试
    testScheduler.advanceTimeBy(3, TimeUnit.SECONDS)

    // 验证订阅终止，且收到最终错误
    testObserver.assertTerminated()
    testObserver.assertError(IOException::class.java)
    assertEquals(4, attemptCounter.get()) // 1次初始调用 + 3次重试
}

场景2：前n次失败，第n+1次成功

@Test
fun testRetryWhenNthRetrySucceeds() {
    val testScheduler = TestScheduler()
    val attemptCounter = AtomicInteger(0)
    val successResponse = Response(200, null, null)

    // 模拟前2次失败，第3次成功的API调用
    val flakyApi = Observable.defer {
        attemptCounter.incrementAndGet().let { count ->
            if (count <= 2) Observable.error<Response<*>>(IOException("API failed"))
            else Observable.just(successResponse)
        }
    }

    val testObserver = flakyApi
        .retryWithMaxAttempts(3, testScheduler)
        .test()

    // 第一次调用失败
    testObserver.assertNotTerminated()
    assertEquals(1, attemptCounter.get())

    // 推进1秒，触发第一次重试（第二次调用，失败）
    testScheduler.advanceTimeBy(1, TimeUnit.SECONDS)
    assertEquals(2, attemptCounter.get())
    testObserver.assertNotTerminated()

    // 再推进1秒，触发第二次重试（第三次调用，成功）
    testScheduler.advanceTimeBy(1, TimeUnit.SECONDS)
    assertEquals(3, attemptCounter.get())

    // 验证订阅成功终止，收到正确响应
    testObserver.assertTerminated()
    testObserver.assertValue(successResponse)
}

3. 排查其他未绑定TestScheduler的异步操作 #

如果你的call扩展函数内部还有其他异步操作（比如API调用用了subscribeOn(Schedulers.io())），需要统一替换为TestScheduler，或者在测试前后用RxJava插件全局替换默认Scheduler：

@Before
fun setup() {
    val testScheduler = TestScheduler()
    // 替换所有默认Scheduler为TestScheduler
    RxJavaPlugins.setIoSchedulerHandler { testScheduler }
    RxJavaPlugins.setComputationSchedulerHandler { testScheduler }
    RxJavaPlugins.setNewThreadSchedulerHandler { testScheduler }
}

@After
fun teardown() {
    // 测试结束后恢复默认Scheduler
    RxJavaPlugins.reset()
}

为什么逐行调试能通过？ #

调试时你单步执行的过程中，真实时间已经走完了重试的延迟（比如1秒），所以重试逻辑自然完成，订阅终止。但自动化测试里没有真实等待，必须靠TestScheduler主动推进时间，没绑定的话就会一直处于等待状态，导致assertTerminated()失败。

内容的提问来源于stack exchange，提问作者Saehun Sean Oh