如何修复单元测试中的Coroutine Timeout问题

我有一个从Repository获取数据的GetAllGamesUseCase类，并编写了对应的单元测试。原本测试运行正常，在修改Fake Repository加入网络连接状态监听后，测试抛出UncompletedCoroutinesError：等待10秒后测试协程仍未完成。相关代码如下，请问该如何修复这个错误？

原Use Case代码 #

class GetAllGamesUseCase(
    private val repository: GameRepository
) {
    operator fun invoke():Flow<Resource<List<Game>>>{
        return repository.getAllGames()
    }
}

测试代码 #

@ExperimentalCoroutinesApi
class GetAllGamesUseCaseTest{
    @get:Rule
    val instantExecutorRule = InstantTaskExecutorRule()


    private lateinit var game:Game
    private lateinit var repository: FakeGameRepository
    private lateinit var connectivityObserver: FakeNetworkConnectivityObserver


    @Before
    fun setup(){
        connectivityObserver = FakeNetworkConnectivityObserver()
        repository = FakeGameRepository(connectivityObserver)
        val games = mutableListOf<Game>()
        for(i in 0..10){
            game = Game(
                developer = "developer $i",
                freeToGameProfileUrl = "freeUrl $i",
                gameUrl = "gameUrl $i",
                genre = "genre $i",
                id = i,
                platform = "plat $i",
                publisher = "publish $i",
                releaseDate = "release $i",
                shortDescription = "short $i",
                thumbnail = "thumb $i",
                title = "title $i"
            )
            games.add(game)
        }
        repository.insertGame(games)

    }


    @Test
    fun `Get all games`() = runTest {
            connectivityObserver.setStatus(Status.Available)

            val result = repository.getAllGames()

            result.collect{

                assertThat(it.data).contains(game)
            }

    }
}

修改后的Fake Repository代码 #

override fun getAllGames(): Flow<Resource<List<Game>>> {

    return flow {
        connectivityObserver.observe().collect{status ->
            when(status){
                Status.Available ->{
                    emit(Resource.Success(games))
                }
                else ->{
                    games = emptyList<Game>().toMutableList()
                    emit(Resource.Error("No Internet Connection"))
                }
            }
        }
    }
}

FakeNetworkConnectivityObserver代码 #

class FakeNetworkConnectivityObserver:ConnectivityObserver{
    private val statusFlow = MutableStateFlow(Status.Unavailable)

    fun setStatus(status: Status){
        statusFlow.value = status
    }
    override fun observe(): Flow<Status> {
        return statusFlow
    }

    }

错误信息 #

After waiting for 10s, the test coroutine is not completing
kotlinx.coroutines.test.UncompletedCoroutinesError: After waiting for 10s, the test coroutine is not completing
    at app//kotlinx.coroutines.test.TestBuildersKt__TestBuildersKt$runTest$2$1$2$1.invoke(TestBuilders.kt:349)
    at app//kotlinx.coroutines.test.TestBuildersKt__TestBuildersKt$runTest$2$1$2$1.invoke(TestBuilders.kt:333)
    at app//kotlinx.coroutines.InvokeOnCancelling.invoke(JobSupport.kt:1430)
    at app//kotlinx.coroutines.JobSupport.notifyCancelling(JobSupport.kt:1474)
    at app//kotlinx.coroutines.JobSupport.tryMakeCancelling(JobSupport.kt:798)
    at app//kotlinx.coroutines.JobSupport.makeCancelling(JobSupport.kt:758)
    at app//kotlinx.coroutines.JobSupport.cancelImpl$kotlinx_coroutines_core(JobSupport.kt:674)
    at app//kotlinx.coroutines.JobSupport.cancelCoroutine(JobSupport.kt:661)
    at app//kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:159)
    at app//kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:501)
    at app//kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:280)
    at app//kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:108)
    at java.base@17.0.7/java.lang.Thread.run(Thread.java:833)

问题原因 #

修改后的FakeGameRepository.getAllGames()中，外层flow内部调用了connectivityObserver.observe().collect{...}，而connectivityObserver.observe()返回的是MutableStateFlow——这是一个永远不会主动结束的热流。测试里调用result.collect{...}时，这个collect会一直挂起等待新的状态值，永远不会主动结束，最终导致测试超时抛出UncompletedCoroutinesError。

修复方法 #

有两种常见的修复方式：

方式1：测试中只收集一次数据

在测试里使用first()操作符替代collect{...}，拿到第一个发射的值后就结束协程：

@Test
fun `Get all games`() = runTest {
    connectivityObserver.setStatus(Status.Available)
    val result = repository.getAllGames().first()
    assertThat(result.data).contains(game)
}

方式2：修改Fake Repository的flow逻辑

如果业务逻辑中getAllGames()只需要响应一次网络状态并返回结果，不需要持续监听，可以修改Fake Repository的实现，让flow发射一次数据后就结束：

override fun getAllGames(): Flow<Resource<List<Game>>> {
    return connectivityObserver.observe()
        .take(1) // 只取第一个状态值
        .map { status ->
            when(status){
                Status.Available -> Resource.Success(games)
                else -> {
                    games = mutableListOf()
                    Resource.Error("No Internet Connection")
                }
            }
        }
}

或者直接获取单次状态后发射：

override fun getAllGames(): Flow<Resource<List<Game>>> {
    return flow {
        val status = connectivityObserver.observe().first()
        when(status){
            Status.Available -> emit(Resource.Success(games))
            else -> {
                games = mutableListOf()
                emit(Resource.Error("No Internet Connection"))
            }
        }
    }
}

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