在 Android 上测试 Kotlin 流

测试与 Flow 通信的单元或模块的方式，取决于被测对象是将 Flow 用作输入还是输出。

  • 如果被测对象观察 Flow，你可以在模拟依赖项（fake dependencies）中生成 Flow，并从测试中对其进行控制。
  • 如果单元或模块对外暴露 Flow，你可以在测试中读取并验证 Flow 发出的一个或多个数据项。

创建模拟生产者

当被测对象是 Flow 的消费者时，一种常见的测试方法是用模拟实现（fake implementation）替换生产者。例如，假设某个类在生产环境中观察一个从两个数据源获取数据的存储库：

the subject under test and the data layer
图 1. 被测对象与数据层。

为了使测试具有确定性，你可以用一个始终发出相同模拟数据的模拟存储库及其依赖项来替换它们。

dependencies are replaced with a fake implementation
图 2. 依赖项被替换为模拟实现。

要使 Flow 发出预定义的一系列值，请使用 flow 构建器。

class MyFakeRepository : MyRepository {
    fun observeCount() = flow {
        emit(ITEM_1)
    }
}

在测试中，注入此模拟存储库以替换真实实现。

@Test
fun myTest() {
    // Given a class with fake dependencies:
    val sut = MyUnitUnderTest(MyFakeRepository())
    // Trigger and verify
    ...
}

现在你已经控制了被测对象的输出，可以通过检查其输出来验证它是否正常工作。

在测试中对 Flow 发出的值进行断言

如果被测对象正在暴露 Flow，测试需要对数据流中的元素进行断言。

假设前面示例中的存储库暴露了一个 Flow：

repository with fake dependencies that exposes a flow
图 3. 带有模拟依赖项并暴露 Flow 的存储库（被测对象）。

对于某些测试，你只需要检查第一次发出的值或 Flow 发出的有限数量的项目。

你可以通过调用 first() 来消费 Flow 的第一次发出值。此函数会一直等待直到收到第一个数据项，然后向生产者发送取消信号。

@Test
fun myRepositoryTest() = runTest {
    // Given a repository that combines values from two data sources:
    val repository = MyRepository(fakeSource1, fakeSource2)

    // When the repository emits a value
    val firstItem = repository.counter.first() // Returns the first item in the flow

    // Then check it's the expected item
    assertEquals(ITEM_1, firstItem)
}

如果测试需要检查多个值，调用 toList() 会导致 Flow 等待数据源发出所有值，然后将这些值作为列表返回。这仅适用于有限的数据流。

@Test
fun myRepositoryTest() = runTest {
    // Given a repository with a fake data source that emits ALL_MESSAGES
    val messages = repository.observeChatMessages().toList()

    // When all messages are emitted then they should be ALL_MESSAGES
    assertEquals(ALL_MESSAGES, messages)
}

对于需要更复杂的数据项收集，或者不返回有限数量数据项的数据流，你可以使用 Flow API 来选择和转换数据项。以下是一些示例：

// Take the second item
outputFlow.drop(1).first()

// Take the first 5 items
outputFlow.take(5).toList()

// Takes the first item verifying that the flow is closed after that
outputFlow.single()

// Finite data streams
// Verify that the flow emits exactly N elements (optional predicate)
outputFlow.count()
outputFlow.count(predicate)

测试期间的持续收集

如前一个示例所示，使用 toList() 收集 Flow 在内部使用了 collect()，并且会挂起直到整个结果列表准备就绪可供返回。

为了交替执行导致 Flow 发出值的操作以及对已发出值进行断言，你可以在测试过程中持续从 Flow 中收集值。

例如，以待测试的以下 Repository 类以及伴随的模拟数据源实现为例，该数据源具有一个 emit 方法，可在测试期间动态生成值：

class Repository(private val dataSource: DataSource) {
    fun scores(): Flow<Int> {
        return dataSource.counts().map { it * 10 }
    }
}

class FakeDataSource : DataSource {
    private val flow = MutableSharedFlow<Int>()
    suspend fun emit(value: Int) = flow.emit(value)
    override fun counts(): Flow<Int> = flow
}

在测试中使用此模拟时，你可以创建一个收集协程，该协程将持续从 Repository 接收值。在此示例中，我们将它们收集到一个列表中，然后对其内容进行断言：

@Test
fun continuouslyCollect() = runTest {
    val dataSource = FakeDataSource()
    val repository = Repository(dataSource)

    val values = mutableListOf<Int>()
    backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
        repository.scores().toList(values)
    }

    dataSource.emit(1)
    assertEquals(10, values[0]) // Assert on the list contents

    dataSource.emit(2)
    dataSource.emit(3)
    assertEquals(30, values[2])

    assertEquals(3, values.size) // Assert the number of items collected
}

由于此处 Repository 暴露的 Flow 永远不会完成，因此用于收集的 toList 调用永远不会返回。在 TestScope.backgroundScope 中启动收集协程可确保协程在测试结束前被取消。否则，runTest 将一直等待其完成，导致测试停止响应并最终失败。

请注意此处如何将 UnconfinedTestDispatcher 用于收集协程。这可确保收集协程被立即启动，并在 launch 返回后准备好接收值。

使用 Turbine

第三方 Turbine 库为创建收集协程提供了一个便捷的 API，以及用于测试 Flow 的其他便利功能。

@Test
fun usingTurbine() = runTest {
    val dataSource = FakeDataSource()
    val repository = Repository(dataSource)

    repository.scores().test {
        // Make calls that will trigger value changes only within test{}
        dataSource.emit(1)
        assertEquals(10, awaitItem())

        dataSource.emit(2)
        awaitItem() // Ignore items if needed, can also use skip(n)

        dataSource.emit(3)
        assertEquals(30, awaitItem())
    }
}

有关更多详细信息，请参阅该库的文档。

测试 StateFlow

StateFlow 是一个可观察的数据持有者，可以通过收集来作为流观察其随时间持有的值。请注意，这个值流是合并（conflated）的，这意味着如果值在 StateFlow 中快速设置，则不能保证 StateFlow 的收集者能收到所有中间值，只能收到最新值。

在测试中，如果你考虑到合并，可以像收集任何其他 Flow 一样收集 StateFlow 的值（包括使用 Turbine）。在某些测试场景中，尝试收集并断言所有中间值可能是可取的。

但是，我们通常建议将 StateFlow 视为数据持有者，并改对它的 value 属性进行断言。这样，测试验证的是对象在特定时间点的当前状态，而不依赖于是否发生了合并。

例如，以这个从 Repository 收集值并通过 StateFlow 将其暴露给 UI 的 ViewModel 为例：

class MyViewModel(private val myRepository: MyRepository) : ViewModel() {
    private val _score = MutableStateFlow(0)
    val score: StateFlow<Int> = _score.asStateFlow()

    fun initialize() {
        viewModelScope.launch {
            myRepository.scores().collect { score ->
                _score.value = score
            }
        }
    }
}

此 Repository 的模拟实现可能如下所示：

class FakeRepository : MyRepository {
    private val flow = MutableSharedFlow<Int>()
    suspend fun emit(value: Int) = flow.emit(value)
    override fun scores(): Flow<Int> = flow
}

在使用此模拟测试 ViewModel 时，你可以从模拟中发出值以触发 ViewModel 的 StateFlow 更新，然后对更新后的 value 进行断言。

@Test
fun testHotFakeRepository() = runTest {
    val fakeRepository = FakeRepository()
    val viewModel = MyViewModel(fakeRepository)

    assertEquals(0, viewModel.score.value) // Assert on the initial value

    // Start collecting values from the Repository
    viewModel.initialize()

    // Then we can send in values one by one, which the ViewModel will collect
    fakeRepository.emit(1)
    assertEquals(1, viewModel.score.value)

    fakeRepository.emit(2)
    fakeRepository.emit(3)
    assertEquals(3, viewModel.score.value) // Assert on the latest value
}

使用 stateIn 创建的 StateFlow

在前一节中，ViewModel 使用 MutableStateFlow 来存储 Repository 中 Flow 发出的最新值。这是一种常见模式，通常通过使用 stateIn 运算符以更简单的方式实现，该运算符将冷流转换为热 StateFlow。

class MyViewModelWithStateIn(myRepository: MyRepository) : ViewModel() {
    val score: StateFlow<Int> = myRepository.scores()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000L), 0)
}

stateIn 运算符具有一个 SharingStarted 参数，该参数决定了它何时变为活跃状态并开始消费底层 Flow。诸如 SharingStarted.Lazily 和 SharingStarted.WhileSubscribed 之类的选项经常在 ViewModel 中使用。

即使你在测试中对 StateFlow 的 value 进行断言，你也需要创建一个收集者。这可以是一个空的收集者：

@Test
fun testLazilySharingViewModel() = runTest {
    val fakeRepository = HotFakeRepository()
    val viewModel = MyViewModelWithStateIn(fakeRepository)

    // Create an empty collector for the StateFlow
    backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
        viewModel.score.collect {}
    }

    assertEquals(0, viewModel.score.value) // Can assert initial value

    // Trigger-assert like before
    fakeRepository.emit(1)
    assertEquals(1, viewModel.score.value)

    fakeRepository.emit(2)
    fakeRepository.emit(3)
    assertEquals(3, viewModel.score.value)
}

其他资源