testing-flows-with-turbine

使用 Turbine 测试 Flow——无需 toList 卡住即可断言发射

Turbine 是逐个断言 Flow 发射的标准方式，它内置了取消机制，可避免热流（StateFlow、SharedFlow）导致测试卡住。本技能涵盖 flow.test、ReceiveTurbine API、用于多流测试的 turbineScope，以及在什么情况下直接使用 flow.toList() 确实比 Turbine 更合适。非 Flow 的协程配置请参阅 ../testing-coroutines-with-runtest/SKILL.md。

何时使用此技能

• 被测类公开了 StateFlow<UiState> 或 SharedFlow<Event>，且测试必须断言每次发射。
• 对 SharedFlow 或 StateFlow 使用 flow.toList() 时测试卡住——这两者都不会完成。
• 开发者想断言“一小段时间内没有更多发射”或“下一个事件是错误”，而 toList() 无法表达这些断言。
• 必须并发观察两个流（例如 uiState 和一次性 events 流），并按交错顺序进行断言。
• 构建错误为 TurbineAssertionError: Unconsumed events found 或 Expected an item but found Complete/Error。
• 开发者提到 Turbine API：awaitItem、awaitComplete、awaitError、expectMostRecentItem、skipItems、expectNoEvents、withTurbineTimeout、turbineScope、testIn。

何时不要使用此技能

• 测试涉及 runTest、TestScope、Dispatchers.setMain 或虚拟时间。请先使用 ../testing-coroutines-with-runtest/SKILL.md；Turbine 基于 runTest。
• 流是冷流，会以确定的方式完成，断言内容是“完整列表等于 X”。使用 flow.toList() 更简洁——请参阅下方的决策矩阵。
• “流”实际上是 Compose 快照状态。请通过测试规则读取，而不是使用 Turbine。

前置条件

• Gradle：testImplementation("app.cash.turbine:turbine:1.x")（docs/CORPUS.md §G.1；R5）。
• Turbine 支持多平台——JVM、Android、iOS、JS、Native、wasm。Apache 2.0，Square / Cash App。
• 外围测试必须位于 runTest { … } 函数体中（或任意 suspend 上下文中）。flow.test { } 本身就是 suspend。
• 对于多流测试，需要使用 @OptIn(ExperimentalCoroutinesApi::class)，因为 turbineScope 会与 TestScope.backgroundScope 交互。

flow.test 签名

public suspend fun <T> Flow<T>.test(
    timeout: Duration = 3.seconds,
    name: String? = null,
    validate: suspend ReceiveTurbine<T>.() -> Unit,
)

（docs/CORPUS.md §G；R5。）需要牢记以下三点：

1. test { } 会启动一个新协程，调用 flow.collect { }，并将每次发射送入 ReceiveTurbine<T> 通道。收集器会在运行 validate { } 之前附加，因此可以在代码块内安全地向热流发射数据，不会丢失第一个元素。
2. 默认的 3 秒超时时间按墙上时钟计算——它不由 TestCoroutineScheduler 驱动，也不遵循 runTest 的虚拟时间。awaitItem() 最多等待 3 秒的实际时间。
3. validate { } 返回后，Turbine 会断言没有未消费的事件。如果仍有任何待处理的元素／完成信号／错误，就会抛出 TurbineAssertionError。随后会取消收集协程，因此不会泄漏冷流。

name 参数会附加到错误消息中——当一个测试中运行多个 Turbine 时，这一功能非常有用。

ReceiveTurbine API

public interface ReceiveTurbine<T> {
    // suspending
    public suspend fun awaitItem(): T
    public suspend fun awaitComplete()
    public suspend fun awaitError(): Throwable
    public suspend fun awaitEvent(): Event<T>
    public suspend fun skipItems(count: Int)
    public suspend fun cancel()
    public suspend fun cancelAndIgnoreRemainingEvents()
    public suspend fun cancelAndConsumeRemainingEvents(): List<Event<T>>

    // non-suspending
    public fun expectNoEvents()
    public fun expectMostRecentItem(): T
    public fun ensureAllEventsConsumed()
}

（R5。）

调用	使用场景
awaitItem(): T	下一个事件必须是项目；如果是 Complete/Error，则失败。
awaitComplete()	下一个事件必须是流完成；如果是项目/错误，则失败。
awaitError(): Throwable	下一个事件必须是错误；返回该异常，以便进行后续断言。
awaitEvent(): Event<T>	密封类入口：Event.Item(value)、Event.Complete、Event.Error(throwable)。
skipItems(n)	丢弃 n 个项目事件；如果下一个事件不是项目，则失败。可用于跳过 StateFlow 的初始值。
expectMostRecentItem()	非挂起；排空缓冲的项目并返回最新值。对于由 StateFlow 驱动且中间值会丢失的 UI，这是用于断言“最终状态”的方法。
expectNoEvents()	非挂起；如果缓冲区中有任何内容，则抛出异常。可与 runCurrent()/advanceTimeBy(...) 搭配，断言没有意外发射。
cancel()	停止收集器；作用域退出时，任何剩余事件都会导致断言失败。
cancelAndIgnoreRemainingEvents()	停止并静默排空。任何热流测试结束时的默认处理方式。
cancelAndConsumeRemainingEvents()	返回 List<Event<T>>，供临时检查使用。

热流与冷流约定

流类型	会自然完成吗？	首次 awaitItem() 返回	需要 cancel() 吗？
flowOf(...) / flow { }（冷流）	是	首次发射	否——awaitComplete() 会终止
MutableSharedFlow<T>（replay = 0）	否	下一个发射（或一直挂起）	始终需要
MutableStateFlow<T>	否	初始值（replay = 1）	始终需要

（docs/CORPUS.md §G；R5。）

重要：SharedFlow 和 StateFlow 永远不会调用 onCompletion。对任一类型调用 awaitComplete()，都会一直挂起，直至 3 秒的墙钟超时。务必以 cancel()（或 cancelAndIgnoreRemainingEvents()）结束。

StateFlow.test 总会先发射初始值，因为 StateFlow 的 replay = 1。如果不关心初始值，可调用 skipItems(1) 跳过。

turbineScope 和 testIn——在一次测试中处理多个流

public suspend fun <R> turbineScope(
    timeout: Duration? = null,
    validate: suspend TurbineTestContext.() -> R,
): R

public fun <T> Flow<T>.testIn(
    scope: CoroutineScope,
    timeout: Duration? = null,
    name: String? = null,
): ReceiveTurbine<T>

当断言需要在两个流之间交替进行时，请使用这些方法：

runTest(mainRule.dispatcher) {
    turbineScope {
        val items = repo.items.testIn(backgroundScope, name = "items")
        val errors = repo.errors.testIn(backgroundScope, name = "errors")
        assertEquals(0, items.awaitItem())
        assertEquals(emptyList<Throwable>(), errors.expectMostRecentItem())
        items.cancel()
        errors.cancel()
    }
}

在 turbineScope 结束前，testIn 必须显式调用 cancel()（或 awaitComplete() / awaitError()）；否则作用域断言会抛出异常。

独立的 Turbine<T>——捕获回调和一次性信号

Turbine 不只是一个 Flow 适配器。独立的 Turbine<T> 是一个发送与接收通道，具有相同的 ReceiveTurbine 断言接口；适用于被测对象不是 Flow 的场景：(T) -> Unit 回调、监听器接口、由你持有的 Channel，以及需要记录调用的手写假对象。

public fun <T> Turbine(timeout: Duration? = null, name: String? = null): Turbine<T>

public interface Turbine<T> : ReceiveTurbine<T> {
    public val isClosed: Boolean
    public fun add(item: T)                       // producer side — push an item
    public fun close(cause: Throwable? = null)    // signal Complete (or Error if cause != null)
    public fun asChannel(): Channel<T>            // hand the producer a plain SendChannel
}

（docs/CORPUS.md §G；R5。）它让“这个回调是否以 X 调用过，而且只调用了这一次？”也能使用相同的 awaitItem() 术语：

@Test fun emitsClickedId() = runTest(mainRule.dispatcher) {
    val clicks = Turbine<Long>()
    val adapter = ItemAdapter(onClick = clicks::add)        // production code calls onClick(id)

    adapter.bind(item)
    adapter.simulateClick()

    assertEquals(42L, clicks.awaitItem())
    clicks.ensureAllEventsConsumed()                        // no extra invocations slipped through
}

ensureAllEventsConsumed() 用于断言缓冲区中没有剩余事件——这相当于独立 Turbine 版本的检查：flow.test { } 会在代码块退出时隐式执行该检查。独立的 Turbine 永远不会自动完成，因此除非调用 close()，否则对它调用 awaitComplete() 会一直挂起，直至超时；请像对待热流一样处理它。

同步（非 suspend）消费：takeItem / takeComplete / takeError

当断言位置无法挂起时——例如产生事件的协程已经运行完毕后的普通 @Test 主体，或非 suspend 回调中的 assert——ReceiveTurbine 提供非挂起的取出方法。这些方法要求事件已经缓冲，否则会抛出异常：

public fun takeItem(): T
public fun takeComplete()
public fun takeError(): Throwable

@Test fun synchronousDrain() = runTest(mainRule.dispatcher) {
    val turbine = vm.uiState.testIn(backgroundScope)
    advanceUntilIdle()                              // let the producer run to quiescence (virtual time)
    assertEquals(Idle, turbine.takeItem())          // no suspension — the value is already buffered
    assertEquals(Loaded, turbine.takeItem())
    turbine.cancel()
}

优先在 runTest / flow.test 代码块中使用挂起的 awaitItem()；只有调用位置确实无法挂起时，才使用 takeItem()。

Turbine 超时与 runTest 虚拟时间

Turbine 的默认 3 秒是墙上时钟时间，绝不是虚拟时间。runTest 默认的 60 秒同样是墙上时钟时间——但在测试代码中调用 advanceTimeBy(...) 会操纵 TestCoroutineScheduler 的虚拟时间，而 Turbine 无法感知到这一点。

这意味着：使用 delay(10.minutes) 的测试会在微秒内完成，因为 runTest 会跳过延迟。但如果 flow.test { awaitItem() } 等待的发射来自真实网络 I/O，它仍会受到真实墙上时钟时间的影响。如果等待在真实时间中确实超过 3 秒，请覆盖默认值：

flow.test(timeout = 10.seconds) { … }            // per-call

withTurbineTimeout(10.seconds) {                 // scoped
    flow.test { … }
}

决策矩阵——Turbine 还是原始 flow.toList()

场景	选择
会完成的冷流；“完整列表等于 X”	flow.toList()（更简洁）
热 StateFlow / SharedFlow	Turbine（可处理取消）
需要断言中间状态，然后触发更多发射	Turbine
需要断言“一段时间内没有更多事件”	Turbine（expectNoEvents）
需要断言错误类型 / 完成	Turbine（awaitError、awaitComplete）
在一个测试中交错处理多个流	Turbine（turbineScope/testIn）
在 JS/Native 上进行多平台测试	两者都可用；Turbine 更简单
不引入额外依赖	flow.toList()

（R5。）

经验法则：对于任何 UI 状态 / ViewModel 测试（几乎总是由 StateFlow 驱动），优先使用 Turbine；对于会确定性完成的冷流仓库 / 转换测试，使用 flow.toList()。

模式

模式：错误——对 SharedFlow 使用 flow.toList()

// WRONG
@Test fun events() = runTest {
    val events: SharedFlow<Event> = vm.events
    val collected = events.toList()                 // never returns
    assertEquals(listOf(LoggedIn), collected)
}
// WRONG because: SharedFlow never completes. toList() suspends until the upstream
// completes, which it never does. The test hangs to the 60s runTest timeout.

// RIGHT
@Test fun events() = runTest(mainRule.dispatcher) {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        cancelAndIgnoreRemainingEvents()             // hot flow — must cancel
    }
}

模式：错误——忘记 StateFlow 的种子值

// WRONG
@Test fun stateFlowEmits() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        vm.refresh()
        assertEquals(Loading, awaitItem())           // FAIL: actually Idle (seed)
        cancel()
    }
}
// WRONG because: StateFlow has replay = 1. The collector immediately receives the
// current value (Idle) before any new emission. The first awaitItem() returns
// the seed, not the post-refresh state.

// RIGHT — assert the seed
@Test fun stateFlowEmits() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        assertEquals(Idle, awaitItem())              // seed
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

// RIGHT — skip the seed if you do not care
@Test fun stateFlowEmits_skipSeed() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

模式：错误——对热流调用 awaitComplete()

// WRONG
@Test fun events() = runTest {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        awaitComplete()                              // hangs 3s, then TurbineAssertionError
    }
}
// WRONG because: SharedFlow never completes. awaitComplete() waits 3 wall-clock
// seconds (Turbine default) and then throws. The test is slow AND fails.

// RIGHT
@Test fun events() = runTest(mainRule.dispatcher) {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

模式：StateFlow 合并值——使用 expectMostRecentItem

@Test fun finalState() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)                                 // seed
        vm.flipFast()                                // posts 5 transitions in <1ms
        advanceUntilIdle()                           // virtual; let coroutines settle
        // StateFlow conflates — only the latest value is kept.
        // awaitItem() may see anything from the burst; expectMostRecentItem is deterministic.
        assertEquals(Final, expectMostRecentItem())
        cancelAndIgnoreRemainingEvents()
    }
}

模式：断言“没有更多事件”

@Test fun noSpuriousEmissions() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        advanceTimeBy(5.seconds)                     // virtual — flush any timers
        runCurrent()
        expectNoEvents()                             // confirms quiescence
        cancelAndIgnoreRemainingEvents()
    }
}

模式：会完成的冷流——Turbine 与 toList

// FINE — toList() is shorter for terminal cold flows
@Test fun coldList() = runTest {
    val collected = flowOf(1, 2, 3).toList()
    assertEquals(listOf(1, 2, 3), collected)
}

// ALSO FINE — Turbine if you need intermediate-state assertions
@Test fun coldTurbine() = runTest {
    flowOf(1, 2, 3).test {
        assertEquals(1, awaitItem())
        assertEquals(2, awaitItem())
        assertEquals(3, awaitItem())
        awaitComplete()                              // cold flow — completes naturally
    }
}

模式：使用 turbineScope 交错处理两个流

@Test fun interleaved() = runTest(mainRule.dispatcher) {
    turbineScope {
        val state = vm.uiState.testIn(backgroundScope, name = "state")
        val events = vm.events.testIn(backgroundScope, name = "events")
        assertEquals(Idle, state.awaitItem())
        vm.login()
        assertEquals(Loading, state.awaitItem())
        assertEquals(LoggedIn, events.awaitItem())
        assertEquals(Success, state.awaitItem())
        state.cancel(); events.cancel()
    }
}

强制规则

• 必须将任何 StateFlow / SharedFlow 断言包在 flow.test { … cancelAndIgnoreRemainingEvents() } 中。不得对热流调用 flow.toList()——测试会一直挂起。
• 必须在每个针对热流的 test { } 块末尾调用 cancel() / cancelAndIgnoreRemainingEvents()。对热流使用 awaitComplete() 是错误的——它们永远不会完成。
• 必须考虑 StateFlow 的初始值：第一个 awaitItem() 会返回当前值（重放数 = 1）。要么断言该值，要么调用 skipItems(1)。
• 必须仅通过 testImplementation 依赖 Turbine。不得将 app.cash.turbine 放在 implementation 中。
• 必须在等待的发射确实需要超过 3 秒实际时间时，覆盖 withTurbineTimeout(...)（或按调用设置 timeout =）。默认的 3 秒按实际时间计算，且独立于 runTest 的虚拟时间。
• 必须在 testIn 块中、turbineScope 结束前调用 cancel()（而非 awaitComplete()）；否则作用域断言会抛出异常。
• 不得将 Turbine 超时与 runTest 的虚拟时钟语义混为一谈。advanceTimeBy(10.minutes) 使用虚拟时间；Turbine 的 3 秒等待使用实际时间。
• **优先使用：**在针对合流 StateFlow 的确定性“最终状态”断言中使用 expectMostRecentItem()，因为中间值可能会被折叠。
• **优先使用：**对于会确定性完成的冷流，直接使用 flow.toList()；这种情况下使用 Turbine 只会增加开销。
• **优先使用：**当测试中同时运行多个 Turbine 时，为它们命名（flow.test(name = "uiState")、flow.testIn(scope, name = "events")）——错误消息会引用名称。
• **优先使用：**验证 (T) -> Unit 回调或监听器是否被调用时，使用独立的 Turbine<T>()（搭配 add / close / ensureAllEventsConsumed），而不是手动累积到 MutableList 中——它能提供相同的 awaitItem() 诊断信息，并检查是否有未消费事件。
• **优先使用：**在 runTest / flow.test 块中使用挂起式 awaitItem()，而不是非挂起式 takeItem() / takeComplete() / takeError()；仅当调用位置确实无法挂起时，才使用这些取值方法。

验证

• 没有对任何 StateFlow 或 SharedFlow 调用 flow.toList()（使用 grep -rn 'toList()' src/test 搜索，并与流类型交叉核对）。
• 每个针对热流的 flow.test { … } 都以 cancel() 或 cancelAndIgnoreRemainingEvents() 结束。
• 没有对 StateFlow / SharedFlow 调用 awaitComplete()。
• StateFlow 测试会通过 awaitItem() 断言初始值，或在后续断言前调用 skipItems(1)。
• 可能合理地等待超过 3 秒的 Turbine 使用 withTurbineTimeout(...) 或按调用设置 timeout = N.seconds。
• 多流测试使用 turbineScope { … testIn(backgroundScope) … }，而不是嵌套的 test { } 块。
• 回调 / 监听器断言使用独立的 Turbine<T>()（并调用 ensureAllEventsConsumed()），而不是临时创建的 MutableList。
• CI 运行连续完成 50 次迭代，且没有出现 TurbineAssertionError 波动性失败。

参考资料

• Cash App Turbine — README 和 API：https://github.com/cashapp/turbine
• Cash App Turbine — 发行版本：https://github.com/cashapp/turbine/releases
• Android Developers — 在 Android 上测试 Kotlin Flow：https://developer.android.com/kotlin/flow/test
• Android Developers — StateFlow 和 SharedFlow：https://developer.android.com/kotlin/flow/stateflow-and-sharedflow
• JetBrains — 异步 Flow：https://kotlinlang.org/docs/flow.html
• 研究：tasks/research/R5-coroutines-test-turbine.md — Turbine API 范围、热流/冷流契约，以及与 toList() 的比较。
• docs/CORPUS.md §G.1 — Gradle 坐标和版本固定。
• 同级：../testing-coroutines-with-runtest/SKILL.md — runTest、MainDispatcherRule，以及 Turbine 所依赖的虚拟时间基础。
• 同级：../../mocking/mocking-with-mockito/SKILL.md — 对产生流的挂起依赖进行模拟。
• 同级：../../mocking/mocking-with-mockk/SKILL.md — coEvery { repo.items() } returns flowOf(...)。
• 同级：../../runner/configuring-junit4-on-android/SKILL.md — JUnit4 配置。
• 同级：../../robolectric/using-robolectric-correctly/SKILL.md — 当被测系统（SUT）需要在 JVM 上使用 Android 框架时。
• 跨集合：../../../fundamentals/strategies/applying-testing-strategies/SKILL.md — Flow 测试的小型/中型/大型规模划分。
• 跨集合：../../../fundamentals/strategies/organizing-test-source-sets/SKILL.md — Flow 测试的归属位置（src/test/）。
• 跨集合：../../../compose/synchronization/controlling-the-test-clock/SKILL.md — Compose 的 MainTestClock 与 Turbine 的墙上时钟互不相关。
• 跨集合：../../../instrumentation/runner/running-instrumented-tests-with-androidjunit4/SKILL.md — Turbine 也可在设备上使用，API 相同。
• 跨集合：../../../kotlin/kotlin-test/writing-tests-with-kotlin-test/SKILL.md — 将 awaitItem() 与 kotlin.test 的 assertEquals / assertIs 配对使用；即 flow.test { } 中的断言用法。