Kotlin Flow Combine单元测试无法收集全部数据问题排查

问题原因 #

你创建job2时使用了独立的CoroutineScope(testDispatcher)，这个协程不属于runBlockingTest的作用域，导致advanceUntilIdle()不会等待该协程的任务完成。当第二次调用flow2.emit(true)后，combine的收集逻辑被放到了独立作用域的调度队列中未被执行，因此flow没有发射第二个值，最终collectedValues仅收集到一个元素。

另外，runBlockingTest在Kotlin 1.6+已被标记为废弃，推荐使用更可靠的runTest API。

解决方案 #

方案1：在runBlockingTest作用域下启动协程

修改job2的创建方式，直接使用当前runBlockingTest的作用域启动协程，确保advanceUntilIdle()能等待其任务完成：

@Test
fun testEmission2() = runBlockingTest {
    val flow = MutableSharedFlow<Boolean>(replay = 1)
    val flow2 = MutableSharedFlow<Boolean>(replay = 1)
    val flow3 = MutableSharedFlow<Boolean>(replay = 1)
    val collectedValues = mutableListOf<Boolean>()
    
    val job: Job = launch {
        flow.collect {
            collectedValues.add(it)
        }
    }

    // 改为在当前runBlockingTest作用域下启动协程
    val job2 = launch(testDispatcher) {
        combine(
            flow2,
            flow3,
            transform = { a, b ->
                Pair(a, b)
            }
        ).collect {
            val (a, b) = it
            flow.emit(a)
        }
    }

    flow3.emit(false)
    flow2.emit(false)
    flow2.emit(true)

    advanceUntilIdle() // 现在会等待job2的任务完成
    job.cancel()
    job2.cancel()

    val expectedValues = listOf(false, true)
    assertEquals(expectedValues, collectedValues)
}

方案2：迁移到runTest API（推荐）

runTest是Kotlin 1.6+官方推荐的测试API，自带TestScope管理协程，无需手动维护TestCoroutineDispatcher：

首先修改父类BaseCoroutineTest，移除过时的调度器配置：

open class BaseCoroutineTest {

    @get:Rule
    val liveDataThreadTestRule = InstantTaskExecutorRule()

    @After
    fun clearMocks() {
        Mockito.framework().clearInlineMocks()
    }
}

然后修改测试函数：

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun testEmission2() = runTest {
    val flow = MutableSharedFlow<Boolean>(replay = 1)
    val flow2 = MutableSharedFlow<Boolean>(replay = 1)
    val flow3 = MutableSharedFlow<Boolean>(replay = 1)
    val collectedValues = mutableListOf<Boolean>()
    
    val job: Job = launch {
        flow.collect {
            collectedValues.add(it)
        }
    }

    // 直接在runTest的TestScope下启动协程
    val job2 = launch {
        combine(
            flow2,
            flow3,
            transform = { a, b ->
                Pair(a, b)
            }
        ).collect {
            val (a, b) = it
            flow.emit(a)
        }
    }

    flow3.emit(false)
    flow2.emit(false)
    flow2.emit(true)

    advanceUntilIdle()
    job.cancel()
    job2.cancel()

    val expectedValues = listOf(false, true)
    assertEquals(expectedValues, collectedValues)
}

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