Turbine 测试 Kotlin Flow，让异步变同步

「Turbine 测试 Kotlin Flow，让异步变同步」

Flow 的测试一直是个让人头疼的事。Kotlin 协程把异步代码写得像同步一样顺手，可一到单元测试里，这种"顺手"就变成了"烫手"。runTest 确实解决了挂起函数的问题，但面对一个不断发射值的 Flow，你怎么断言中间状态？怎么验证时序？怎么处理永远不会完结的冷流？这些问题在 2021 年之前几乎没有一个让人满意的答案，直到 Cash App 开源了 Turbine。

我第一次认真用 Turbine 是在一个音视频项目的重构里。那个项目的核心是一个 MediaController 类，内部维护着播放状态机，对外暴露 StateFlow<PlaybackState>。状态转换极其复杂：从 Idle 到 Buffering 到 Playing，网络波动时回退到 Buffering，用户拖拽进度条进入 Seeking，错误时跳转到 Error。每个状态转换都有对应的业务逻辑，而旧的测试代码用 runBlocking 加 take(1) 的拼凑方式，覆盖率惨不忍睹。引入 Turbine 之后，测试代码的行数没减多少，但可读性和对时序的精确控制完全是两个世界。这篇文章想分享的，就是 Turbine 到底做了什么、怎么用好它、以及它那些文档里不会明说的坑。

`test` 块：把事件流变成顺序执行

Turbine 最核心的 API 就一个：Flow<T>.test { ... }。这个扩展函数接收一个挂起 lambda，在这个 lambda 内部，你可以按顺序"消费" Flow 发射的每一个值。听起来简单，但它解决了一个根本问题：在协程测试里，生产者和消费者的时间线是纠缠在一起的。

举个例子。假设你有一个这样的 ViewModel：

class TimerViewModel {
    private val _elapsed = MutableStateFlow(0)
    val elapsed: StateFlow<Int> = _elapsed.asStateFlow()
    
    fun start() {
        viewModelScope.launch {
            while (isActive) {
                delay(1000)
                _elapsed.value += 1
            }
        }
    }
}

要测试这个 start() 方法，旧的做法可能是：

@Test
fun testTimer() = runTest {
    val viewModel = TimerViewModel()
    viewModel.start()
    
    delay(2500)
    val value = viewModel.elapsed.value
    assertEquals(2, value)
}

这段代码的问题一眼可见：依赖真实时间（哪怕 runTest 会跳过延迟），断言的是"某个时刻的快照"而非"事件序列"，而且如果状态机更复杂，你根本没法验证中间过程。

Turbine 的做法完全不同：

@Test
fun testTimerWithTurbine() = runTest {
    val viewModel = TimerViewModel()
    viewModel.start()
    
    viewModel.elapsed.test {
        assertEquals(0, awaitItem())  // 初始值
        delay(1100)
        assertEquals(1, awaitItem())  // 第一次递增
        delay(1100)
        assertEquals(2, awaitItem())  // 第二次递增
        cancelAndConsumeRemainingEvents()
    }
}

awaitItem() 是一个挂起函数，它会挂起直到 Flow 发射下一个值。这意味着测试代码和 Flow 的生产完全同步：消费者准备好之后，生产者才继续。这种"拉取"模型让时序控制变得极其精确。你可以在任何两个 awaitItem() 之间插入验证逻辑、模拟副作用、甚至启动另一个协程。

Turbine 的 test 块内部维护了一个无界通道（unbounded channel），Flow 的每个值都会被收集到这个通道里。awaitItem() 从通道头部取出一个值，如果通道空了就挂起。这个实现细节解释了为什么 Turbine 能处理快速发射的场景——它不会丢事件，但也不会背压，内存里会暂存所有未被消费的值。对于测试场景来说，这个 trade-off 完全合理，因为测试数据量通常很小。

`awaitItem()` 家族：精确控制与超时

awaitItem() 并不是唯一的消费函数。Turbine 提供了一整套 awaitXxx 方法，覆盖了各种边界情况：

awaitItem() 等待下一个值，返回该值。如果 Flow 已经完成或出错，会抛出 AssertionError。

awaitError() 等待 Flow 异常终止，返回抛出的异常。如果你预期某个操作会导致 Flow 失败，用这个来断言具体的异常类型和消息。

awaitComplete() 等待 Flow 正常完成。对于有限流（比如 flow { emit(1); emit(2) }），在消费完所有值之后调用它来验证确实结束了。

awaitEvent() 是更底层的 API，返回一个 Event 密封类，包含 Item、Error、Complete 三种子类。当你需要统一处理三种情况时用它。

expectNoEvents() 验证当前没有待处理的事件，并且在未来一段时间内也不会有。它接受一个 duration 参数，默认是 1 秒（在 runTest 里会被自动跳过，所以实际不消耗真实时间）。

这些 API 的组合让测试可以非常精确。比如验证一个搜索框的 debounce 逻辑：

@Test
fun searchDebounce() = runTest {
    val viewModel = SearchViewModel()
    
    viewModel.results.test {
        // 初始空状态
        assertEquals("", awaitItem().query)
        
        viewModel.onQueryChange("k")
        viewModel.onQueryChange("ko")
        viewModel.onQueryChange("kot")
        
        // 300ms debounce 内不应该触发搜索
        expectNoEvents()
        
        // 超过 debounce 时间后，只应该收到最后一个值
        delay(400)
        assertEquals("kot", awaitItem().query)
        
        cancelAndConsumeRemainingEvents()
    }
}

这里的关键是 expectNoEvents()。没有这个断言，你的测试只能验证"最终收到了什么"，而无法验证"中间没收到什么"。很多 bug 恰恰是时序错误导致的重复请求或过早触发。

Turbine 还提供了超时控制。awaitItem() 默认会无限挂起，但你可以用 withTimeout 包裹，或者使用 awaitItem(timeout = 1.seconds)（需要显式传入）。在 runTest 环境下，超时通常意味着测试逻辑有问题——要么漏掉了某个事件，要么 Flow 的行为和预期不符。

`testIn` 与共享流的特殊处理

StateFlow 和 SharedFlow 有个特性：它们会缓存最后一个值（StateFlow 永远缓存，SharedFlow 取决于 replay 参数）。这意味着当你调用 .test { } 时，awaitItem() 会立即收到缓存值，而不是等待下一个新值。

这个行为有时候会让人困惑。比如：

val stateFlow = MutableStateFlow(0)

stateFlow.test {
    assertEquals(0, awaitItem())  // 立即收到缓存的 0
    stateFlow.value = 1
    assertEquals(1, awaitItem())  // 收到新值 1
}

看起来没问题，但如果你的测试需要验证"从某个操作开始之后的新事件"，这个缓存值就成了干扰。Turbine 0.12.0 之后引入了 testIn 来解决这类场景。

testIn(scope) 返回一个 TurbineTest 对象，它不会自动开始消费事件。你可以先执行一些准备操作，然后再开始断言：

@Test
fun testWithTestIn() = runTest {
    val viewModel = ViewModel()
    val turbine = viewModel.state.testIn(backgroundScope)
    
    // 先执行操作
    viewModel.loadData()
    
    // 现在开始消费，跳过缓存值，只关心 loadData 之后的事件
    // 注意：这里需要手动处理缓存值
    assertEquals(Loading, turbine.awaitItem())
    assertEquals(Success(data), turbine.awaitItem())
    
    turbine.cancel()
}

testIn 的另一个重要用途是和 backgroundScope 配合。runTest 的默认测试作用域会在测试结束时取消所有子协程，但 test { } 块会在 lambda 结束时自动取消订阅。如果你需要让 Flow 的收集在测试期间一直运行（比如验证多个独立操作的累积效果），testIn 更灵活。

不过 testIn 有个坑：它不会自动验证 Flow 是否完成或是否有剩余事件。你必须显式调用 cancel() 或 cancelAndConsumeRemainingEvents()，否则测试可能通过但留下未消费的事件，掩盖真正的 bug。

热流测试的陷阱：起始时间点

StateFlow 的测试还有一个更隐蔽的问题：订阅时机。StateFlow 的值更新是同步的，如果你在 test 块外部修改了值，再进入 test 块，你可能收不到那个值——因为它已经被"错过"了。

看这个例子：

val stateFlow = MutableStateFlow(0)

// 危险：在 test 块外部修改
stateFlow.value = 1

stateFlow.test {
    // 这里收到的是 1，但如果上面修改得更快呢？
    assertEquals(1, awaitItem())
}

在单线程的 runTest 里这通常没问题，但一旦涉及真正的并发（比如 Dispatchers.Default 上的计算），订阅和修改的时序就变得不确定。Turbine 的文档建议：对于 StateFlow，总是在 test 块内部或通过 testIn 确保订阅建立后再触发操作。

我个人踩过的一个坑是在测试 ViewModel 的 init 块。很多 ViewModel 在 init { } 里就开始加载数据，如果 test 订阅建立得慢，可能错过 Loading 状态，直接收到 Success。解决方案是用 testIn 提前建立订阅，或者重构 ViewModel 暴露一个显式的 initialize() 方法。

与 `runTest` 的协作：虚拟时间不是万能药

Kotlin 协程测试库提供的 runTest 和 Turbine 是天作之合，但它们的协作方式需要理解清楚。

runTest 会劫持 delay 调用，让虚拟时间立即前进。这意味着 delay(1000) 在测试里不会真的等一秒。Turbine 的 expectNoEvents() 也利用了这一点——它的默认超时在 runTest 里会被压缩到几乎瞬间完成。

但这带来一个限制：如果你的 Flow 内部使用了非挂起的阻塞操作（比如 Thread.sleep），runTest 无能为力，Turbine 也会卡住。更常见的问题是 Dispatchers.Default 或 Dispatchers.IO 上的操作。runTest 默认使用 StandardTestDispatcher，它的调度器是单线程的，不会并行执行。如果你的 Flow 在 flowOn(Dispatchers.Default) 上运行，而测试代码在 StandardTestDispatcher 上等待，死锁很容易发生。

解决方案是注入调度器。不要硬编码 Dispatchers.IO，而是用构造函数参数或依赖注入提供。测试时传入 StandardTestDispatcher 或 UnconfinedTestDispatcher：

class Repository(
    private val dispatcher: CoroutineDispatcher = Dispatchers.IO
) {
    fun fetchData(): Flow<Data> = flow {
        // 模拟网络请求
        emit(api.getData())
    }.flowOn(dispatcher)
}

@Test
fun testRepository() = runTest {
    val repository = Repository(StandardTestDispatcher(testScheduler))
    
    repository.fetchData().test {
        assertEquals(mockData, awaitItem())
        awaitComplete()
    }
}

Turbine 本身不处理调度器问题，但它假设你能控制 Flow 的执行上下文。如果你的测试时不时挂死，先检查调度器配置，而不是怀疑 Turbine。

版本演进与 API 变迁

Turbine 的 API 经历过一次比较重大的重构。0.x 版本时期，test 块返回的是 FlowTurbine，消费方法直接挂在外层。1.0.0 之后改成了接收器模式，test 块的 receiver 就是 ReceiveTurbine，所有 awaitXxx 方法都在这个 receiver 上调用。

这个改动打破了二进制兼容，但提升了类型安全。旧代码长这样：

// Turbine 0.x
val turbine = flow.test()
assertEquals(1, turbine.awaitItem())
turbine.cancel()

新代码：

// Turbine 1.x
flow.test {
    assertEquals(1, awaitItem())
    // 自动取消
}

我升级的时候花了不少时间改测试。最烦人的是 cancelAndIgnoreRemainingEvents() 被移除了，取而代之的是 cancelAndConsumeRemainingEvents()，后者会返回剩余事件列表，让你可以断言"确实没有遗漏重要事件"。

当前最新版本是 1.1.0（Maven Central 可查），支持 Kotlin 1.9+ 和 coroutines 1.7+。如果你还在用 0.x，升级的收益除了 API 更清爽，还有对 kotlinx-coroutines-test 新版本的更好兼容。Cash App 的维护节奏不算快，但关键 bug 修得及时。GitHub 仓库是 cashapp/turbine，issue 区有很多实战讨论。

局限：它不能测什么

Turbine 很强，但不是万能药。有几个场景它要么不支持，要么不是最佳选择。

第一是性能测试。Turbine 内部用通道缓存所有事件，这引入了额外的内存分配和调度开销。测量 Flow 的吞吐量或延迟时，Turbine 会扭曲结果。应该用原生的 measureTimeMillis 或 JMH。

第二是与其他异步框架的集成。Turbine 只支持 Kotlin Flow。如果你在用 RxJava 的 Observable，需要先用 asFlow() 转换，但转换本身可能改变行为（比如背压策略）。Cash App 自己也有 Rx 版本的测试工具，但那是另一个库了。

第三是跨平台的限制。Turbine 1.x 支持 JVM 和 Android，但 Kotlin/Native 和 Kotlin/JS 的支持在 1.0 之后才逐步完善。如果你在做 KMP 项目，需要确认目标平台是否包含在支持的列表里。官方文档的 README 里有明确的平台矩阵。

第四，也是我最想强调的：Turbine 测试的是 Flow 的"外部行为"，不是内部实现。你可以验证 emit(1) 之后 awaitItem() 收到 1，但无法直接断言"这个值是在哪个协程上下文发射的"。有时候这很重要——比如验证 flowOn 是否生效——但 Turbine 不提供这种内省能力。需要配合 TestCoroutineScheduler 的调试日志或自定义拦截器。

还有一个实际使用中的痛点：错误信息不够友好。当 awaitItem() 超时或收到意外值时，异常堆栈经常指向 Turbine 内部，而不是你的测试代码。1.0 之后有所改善，但相比 AssertJ 或 Kotest 的错误提示，仍然有差距。我通常会包装一个辅助函数，在断言失败时打印当前已收到的事件历史，方便定位问题。

一个完整的实战模式

把上面的内容串起来，我想分享一个我在项目中反复使用的测试模式。场景是：一个 Repository 暴露 Flow<Result<Data>>，ViewModel 把它转换成 UiState，UI 层消费。

测试 ViewModel 时，我需要验证：

1. 初始状态正确

2. 调用 load 后进入 Loading

3. Repository 返回成功数据后进入 Success

4. 过程中没有多余的状态跳转

用 Turbine 写出来：

@Test
fun loadData_success() = runTest {
    val fakeRepository = FakeRepository()
    val viewModel = MyViewModel(fakeRepository)
    
    viewModel.uiState.test {
        // 验证初始状态
        assertEquals(UiState.Idle, awaitItem())
        
        // 触发加载
        viewModel.load()
        
        // 验证 Loading
        assertEquals(UiState.Loading, awaitItem())
        
        // 让 Repository 发射成功数据
        fakeRepository.emit(Result.Success(mockData))
        
        // 验证 Success
        val success = awaitItem() as UiState.Success
        assertEquals(mockData, success.data)
        
        // 验证没有更多事件
        expectNoEvents()
        
        cancelAndConsumeRemainingEvents()
    }
}

这里的关键是 expectNoEvents() 的位置。放在最后，确认状态机没有意外跳转。FakeRepository 是一个可控的测试替身，内部用 MutableSharedFlow 让测试代码能精确控制发射时机。

如果 Repository 的实现有并发（比如同时从网络和数据库取数据），我会用 testIn 替代 test，在 backgroundScope 中收集，然后在主测试逻辑中操作 FakeRepository 并断言。这种模式覆盖了项目中 90% 的 Flow 测试场景。

与其他测试工具的对比

最后想提一下 Turbine 在生态中的位置，不用表格，纯文字描述。

kotlinx-coroutines-test 提供的 runTest 是基础设施，解决的是"协程代码怎么在测试里跑起来"。它提供了 TestDispatcher、TestScope 等概念，但没有针对 Flow 的专用断言工具。你可以用 toList() 收集 Flow 的所有值，但那是终端操作，会挂起直到 Flow 完成，对无限流或需要中间断言的场景无能为力。

Kotest 的 kotest-assertions-coroutines 模块提供了一些 Flow 的断言扩展，比如 flow.shouldEmit(1, 2, 3)。它的风格更函数式，链式调用写起来很爽，但时序控制能力不如 Turbine。特别是需要验证"某个事件不应该发生"时，Kotest 的 API 比较别扭。

AppCash 自己还有一个 molecule 库，专门测试 Compose 的 StateFlow 衍生状态。它和 Turbine 有重叠，但定位更窄。如果你在写 Compose 的 MVI 架构，可以看看 molecule 的 testMolecule 函数，内部其实用了 Turbine。

我的建议是：runTest 打底，Turbine 处理所有 Flow 断言，Kotest 或 AssertJ 处理普通断言。三者不冲突，各司其职。

写在最后

Turbine 把 Flow 测试从"异步黑箱"变成了"同步白盒"，这个转变的价值怎么强调都不为过。它不是一个复杂的库，核心代码不过几千行，但设计非常精准——只解决 Flow 测试的痛点，不越界。

它的坑主要集中在热流的缓存行为、testIn 的手动生命周期管理、以及与调度器配置的交互。这些不是设计缺陷，而是异步编程固有的复杂性。理解之后，写出来的测试既可靠又易读。

如果你还没试过，可以从一个现有的 runBlocking 测试开始迁移。把 toList() 或 first() 换成 test { awaitItem() }，感受下时序控制带来的确定性。这个投入绝对值得。

Turbine 在 Maven Central 的坐标是 app.cash.turbine:turbine，当前版本 1.1.0，Apache 2.0 协议，免费使用。GitHub 仓库 cashapp/turbine 有完整文档和示例。