协程挂起函数单元测试问题：StateFlow未更新预期值

问题场景 #

实现的Repository代码 #

interface DataSource {
    suspend fun getStatus(id: String): StateFlow<Status>
}
class Repository(
    val dataSource: DataSource,
    val coroutineScope: CoroutineScope
) {
    val map: ConcurrentMap<String, StateFlow<Status>> = ConcurrentHashMap()
    fun getStatus(id: String): StateFlow<Status> {
        return map.getOrPut(id) {
            openChannel(id)
            MutableStateFlow(DEFAULT_VALUE)
        }
    }
    
    private fun openChannel(id: String) {
        coroutineScope.launch { 
            observeDataSource(id)
        }
    }
    
    private suspend fun observeDataSource(id: String) {
       dataSource.getStatus(id)
            .collect {
                map[id]?.update {it}
            }
    }
}

单元测试代码 #

private lateinit var repository: Repository
private val dataSource: DataSource = mockk()

@BeforeEach
fun setup() {
    repository = Repository(
        dataSource = dataSource,
        coroutineScope = CoroutineScope(UnconfinedTestDispatcher()),
    )
}

@Test
fun `test getStatus`() =
    runTest {
        val expectedStatus = createExpectedStatus()
        coEvery { dataSource.getStatus(any()) } returns flowOf(
            expectedStatus
        )
        var result: Status? = null
        backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
            repository.getStatus("id").collect {
                result = it
            }
        }

        assertEquals(
            expectedStatus,
            result
        )
    }

问题描述 #

实际应用中逻辑运行正常，但单元测试时断言result始终为DEFAULT_VALUE，而非预期的expectedStatus，怀疑与UnconfinedTestDispatcher的使用有关。

修复方案 #

1. 推进测试调度器至所有任务完成 #

runTest的测试调度器不会自动执行完所有挂起协程，断言时observeDataSource中的collect逻辑可能还未更新StateFlow。在断言前调用testScheduler.advanceUntilIdle()，确保所有协程任务执行完毕：

@Test
fun `test getStatus`() =
    runTest {
        val expectedStatus = createExpectedStatus()
        coEvery { dataSource.getStatus(any()) } returns flowOf(expectedStatus)
        var result: Status? = null
        backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
            repository.getStatus("id").collect {
                result = it
            }
        }

        // 推进调度器，完成所有待执行的协程任务
        testScheduler.advanceUntilIdle()

        assertEquals(expectedStatus, result)
    }

2. 统一使用测试调度器，避免混合调度器 #

仓库的协程Scope和测试协程应统一使用runTest提供的testScheduler，不要混用UnconfinedTestDispatcher：

• 修改setup中的协程Scope：

@BeforeEach
fun setup() {
    repository = Repository(
        dataSource = dataSource,
        coroutineScope = CoroutineScope(testScheduler), // 使用runTest的测试调度器
    )
}

• 测试中的backgroundScope无需指定额外调度器：

backgroundScope.launch {
    repository.getStatus("id").collect {
        result = it
    }
}

3. 修复Repository中的StateFlow更新逻辑（隐藏bug） #

原代码中map[id]?.update {it}存在参数命名冲突：update的lambda参数是当前StateFlow的旧值，而此处的it是dataSource发射的新值，导致更新逻辑错误。修改为：

private suspend fun observeDataSource(id: String) {
   dataSource.getStatus(id)
        .collect { newStatus ->
            map[id]?.update { _ -> newStatus }
        }
}

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