23、Flow与测试进阶：Flow的虚拟时间测试、Flow的断言库、集成测试

说实话，很多Android开发者写Flow的时候挺爽的，一到测试就头疼。我早期带团队的时候，就见过不少人因为测试写不好，干脆不测了，结果线上出了几次数据竞争的问题。嗯，今天我们就来把这层窗户纸捅破。

Flow的测试，说白了就三块：虚拟时间控制、专门的断言工具、以及集成测试的套路。我们一个一个来拆。

虚拟时间测试：让时间为你加速

你想想看，如果你的Flow里有个delay(5秒)，测试真要等5秒？那肯定不行。Kotlin协程提供了TestCoroutineScheduler和StandardTestDispatcher，可以让我们手动控制时间。

我个人习惯在测试基类里这样配置：

class FlowTestBase {

    protected val testScheduler = TestCoroutineScheduler()
    protected val testDispatcher = StandardTestDispatcher(testScheduler)

    protected fun advanceTimeBy(ms: Long) {
        testScheduler.advanceTimeBy(ms)
    }

    protected fun advanceUntilIdle() {
        testScheduler.advanceUntilIdle()
    }
}

然后测试的时候，你可以像导演喊“卡”一样，让时间快进：

class MyViewModelTest : FlowTestBase() {

    @Test
    fun `debounce 应该延迟发射`() = runTest(testDispatcher) {
        val flow = flow {
            emit(1)
            delay(100)
            emit(2)
        }.debounce(50)

        val results = mutableListOf<Int>()
        launch {
            flow.collect { results.add(it) }
        }

        advanceTimeBy(60)  // 跳过第一个debounce窗口
        assertEquals(listOf(1), results)

        advanceTimeBy(60)  // 跳过第二个debounce窗口
        assertEquals(listOf(1, 2), results)
    }
}

我曾经在项目里遇到过一个坑：用runTest但不传testDispatcher，结果虚拟时间根本不起作用。记住，runTest默认用的是UnconfinedTestDispatcher，它不会自动推进虚拟时间。你需要显式传入StandardTestDispatcher。

Flow的断言库：Turbine 真香

手动收集Flow然后断言，写多了真的很累。而且容易漏掉一些边界情况，比如Flow有没有正常结束，有没有抛异常。这时候就需要专门的断言库了。

我个人最常用的是 Turbine，它让Flow测试变得像写诗一样优雅。

先加依赖：

testImplementation("app.cash.turbine:turbine:1.0.0")

然后你就可以这样写测试：

class UserRepositoryTest {

    @Test
    fun `用户数据流应该按顺序发射`() = runTest {
        val repository = UserRepository()

        repository.userFlow.test {
            // 等待并断言第一个值
            assertEquals(User.Loading, awaitItem())

            // 等待并断言第二个值
            val user = awaitItem()
            assertTrue(user is User.Success)
            assertEquals("Alice", (user as User.Success).name)

            // 确认Flow正常结束
            awaitComplete()
        }
    }
}

你看，awaitItem()会阻塞直到下一个值发射，awaitComplete()等待Flow结束。如果Flow应该抛异常，可以用awaitError()。

为什么说Turbine香？因为它帮你处理了背压和取消的问题。你自己写collect的时候，很容易忘记在测试结束后取消协程，导致泄漏。Turbine在test块结束后会自动取消。

集成测试：把Flow放进真实环境

单元测试只能测逻辑，集成测试才能暴露真实问题。比如Room的Flow、Retrofit的Flow、DataStore的Flow，这些都需要在集成测试中验证。

我建议用 Hilt 的测试支持 或者 Koin 的测试模块 来替换真实依赖。这里以Room为例：

@HiltAndroidTest
class UserDaoIntegrationTest {

    @Inject
    lateinit var userDao: UserDao

    @Inject
    lateinit var database: AppDatabase

    @Before
    fun setup() {
        // 使用内存数据库，每次测试后自动清理
        // Hilt 会注入测试用的数据库实例
    }

    @Test
    fun `插入用户后Flow应该发射新数据`() = runTest {
        val user = User(id = 1, name = "Bob")

        // 启动收集
        val job = launch {
            userDao.getAllUsers().test {
                // 初始状态应该是空列表
                assertEquals(emptyList(), awaitItem())

                // 插入数据
                userDao.insert(user)

                // 应该收到包含新用户的数据
                val users = awaitItem()
                assertEquals(1, users.size)
                assertEquals("Bob", users[0].name)

                awaitComplete()
            }
        }

        // 等待测试完成
        job.join()
    }

    @After
    fun teardown() {
        database.close()
    }
}

这里有个关键点：Room的Flow是冷流，每次有新的collector都会重新查询。所以在集成测试中，你要确保collect在insert之前就已经启动了，否则可能会漏掉初始值。

我曾经在集成测试中踩过一个坑：用runBlockingTest（已废弃）来测试Room Flow，结果因为调度器不匹配，Flow根本不发射。后来全部换成runTest + StandardTestDispatcher才解决。

核心知识体系

下面这张图总结了Flow测试的三个层次，以及它们之间的关系：

常见问题与避坑指南

问题	原因	解决方案
虚拟时间不生效	使用了 UnconfinedTestDispatcher 或未传入调度器	显式使用 StandardTestDispatcher
Turbine 测试超时	Flow 有延迟但未设置超时时间	使用 test(timeout = ...) 调整
集成测试中 Flow 不发射	collect 在 insert 之后才启动	先启动 collect，再执行数据操作
测试结束后协程泄漏	未取消 collect 协程	使用 Turbine 的 test 块自动管理

嗯，Flow 测试其实没那么玄乎。掌握了虚拟时间控制、用好 Turbine、再加上集成测试的套路，你就能写出又快又稳的测试代码。我个人觉得，测试写得好不好，直接决定了你重构的时候敢不敢下手。所以，别偷懒，把测试写起来。