如何正确对只读SharedFlow进行单元测试？

问题描述 #

测试仓库类中的SharedFlow时遇到异常：实际场景是单例类订阅多个Flow，经处理后通过shareIn()作为应用内单一数据源，但测试时无法获取第一个发射值——要么出现协程未完成的问题，要么toList()返回空列表。同时希望保持Flow的只读特性，不直接从测试中发射值，仅验证其自身逻辑（如合并Flow、API调用等）。

简化后的代码如下：

仓库类代码 #

import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.*

class Repo(dispatcher: CoroutineDispatcher) {

    val flow1 = MutableStateFlow(0)
    val flow2 =  MutableStateFlow(1)

    val sharedFlow = flow1.flatMapLatest { zero ->
        flow2.map { one -> zero to one }
    }.transformLatest { (zero, one) ->
        emit(zero + one)
    }.shareIn(CoroutineScope(dispatcher), SharingStarted.Eagerly, 1)
}

测试类代码 #

import junit.framework.Assert.assertEquals
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runTest
import org.junit.Test

class TestClass {

    @Test
    fun `checking the right state for the shared flow in repo`() = runTest {
        val repo = Repo(testDispatcher)

        launch {
            val state = repo.sharedFlow.firstOrNull()
            assertEquals(state, 1) // 0 + 1
        }.join()
    }
}

错误原因与解决办法 #

1. shareIn的CoroutineScope生命周期不匹配 #

你在shareIn中传入的是CoroutineScope(dispatcher)，这个独立Scope没有和测试的runTest作用域绑定。测试代码执行时，该Scope可能还没来得及启动收集上游Flow，测试就已结束，导致无法收集到值。

解决办法：
修改Repo类，让其依赖外部传入的CoroutineScope而非CoroutineDispatcher，确保SharedFlow的收集生命周期和测试作用域一致：

class Repo(private val scope: CoroutineScope) {

    val flow1 = MutableStateFlow(0)
    val flow2 =  MutableStateFlow(1)

    val sharedFlow = flow1.flatMapLatest { zero ->
        flow2.map { one -> zero to one }
    }.transformLatest { (zero, one) ->
        emit(zero + one)
    }.shareIn(scope, SharingStarted.Eagerly, 1)
}

2. 测试代码冗余的协程调用 #

测试中额外launch协程再join的操作完全多余，直接在runTest的作用域内调用first()即可，避免协程调度带来的不确定性。

最终修正后的测试代码 #

import junit.framework.Assert.assertEquals
import kotlinx.coroutines.test.runTest
import org.junit.Test

class TestClass {

    @Test
    fun `checking the right state for the shared flow in repo`() = runTest {
        val repo = Repo(this) // 传入runTest的CoroutineScope
        val state = repo.sharedFlow.first()
        assertEquals(state, 1)
    }
}

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