Testing Flows With Turbine

Testing Flows With Turbine — Assert Emissions Without toList Hangs

Turbine is the canonical way to assert Flow emissions one-at-a-time, with built-in cancellation that prevents hot flows (StateFlow, SharedFlow) from hanging the test. This skill covers flow.test, the ReceiveTurbine API, turbineScope for multi-flow tests, and when raw flow.toList() actually beats Turbine. The non-Flow coroutine plumbing is ../testing-coroutines-with-runtest/SKILL.md.

When to use this skill

• The class under test exposes a StateFlow<UiState> or SharedFlow<Event> and the test must assert each emission.
• A test hangs on flow.toList() over a SharedFlow or StateFlow — neither completes.
• The developer wants to assert "no more emissions for a beat" or "the next event is an error", which toList() cannot express.
• Two flows must be observed concurrently (e.g. uiState plus a one-shot events flow) and asserted in interleaved order.
• The build error is TurbineAssertionError: Unconsumed events found or Expected an item but found Complete/Error.
• The developer mentions Turbine APIs: awaitItem, awaitComplete, awaitError, expectMostRecentItem, skipItems, expectNoEvents, withTurbineTimeout, turbineScope, testIn.

When NOT to use this skill

• The test is about runTest, TestScope, Dispatchers.setMain, or virtual time. Use ../testing-coroutines-with-runtest/SKILL.md first; Turbine sits on top of runTest.
• The flow is cold and completes deterministically, and the assertion is "the full list equals X". flow.toList() is shorter — see Decision matrix below.
• The "flow" is actually Compose snapshot state. Read it via the test rule, not Turbine.

Prerequisites

• Gradle: testImplementation("app.cash.turbine:turbine:1.x") (docs/CORPUS.md §G.1; R5).
• Turbine multiplatform — JVM, Android, iOS, JS, Native, wasm. Apache 2.0, Square / Cash App.
• The surrounding test must be a runTest { … } body (or any suspend context). flow.test { } is itself suspend.
• For multi-flow tests, @OptIn(ExperimentalCoroutinesApi::class) because turbineScope interacts with TestScope.backgroundScope.

flow.test signature

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

(docs/CORPUS.md §G; R5.) Three things to internalize:

1. test { } launches a new coroutine that calls flow.collect { } and feeds every emission into a ReceiveTurbine<T> channel. The collector is attached before validate { } runs, so it is safe to emit into a hot flow from inside the block without dropping the first item.
2. The 3-second default timeout is wall clock — it is NOT driven by TestCoroutineScheduler and does NOT honor runTest's virtual time. awaitItem() waits up to 3 real seconds.
3. After validate { } returns, Turbine asserts no unconsumed events remain. If any item / completion / error is pending, it throws TurbineAssertionError. The collecting coroutine is then cancelled, so cold flows do not leak.

The name argument is appended to error messages — invaluable when multiple turbines run in one test.

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.)

Hot vs cold flow contract

(docs/CORPUS.md §G; R5.)

CRITICAL: SharedFlow and StateFlow never call onCompletion. awaitComplete() on either will hang to the 3-second wall-clock timeout. Always end with cancel() (or cancelAndIgnoreRemainingEvents()).

StateFlow.test always emits the seed value first because StateFlow has replay = 1. If you do not care about the seed, skipItems(1) it.

turbineScope and testIn — multiple flows in one test

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>

Use these when assertions are interleaved between two flows:

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()
    }
}

testIn requires explicit cancel() (or awaitComplete() / awaitError()) before the turbineScope ends; otherwise the scope assertion throws.

Standalone Turbine<T> — capturing callbacks and one-shot signals

Turbine is not only a Flow adapter. The standalone Turbine<T> is a send-and-receive channel with the same ReceiveTurbine assertion surface — for when the thing under test is not a Flow: a (T) -> Unit callback, a listener interface, a Channel you own, a hand-rolled fake that needs to record invocations.

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.) It turns "was this callback invoked with X, and only that?" into the same awaitItem() vocabulary:

@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() asserts nothing is left buffered — the standalone-Turbine equivalent of the check flow.test { } runs implicitly on block exit. A standalone Turbine never auto-completes, so awaitComplete() on one hangs to the timeout unless you close() it; treat it like a hot flow.

Synchronous (non-suspend) consumption: takeItem / takeComplete / takeError

When the assertion site cannot suspend — a plain @Test body after the producing coroutine already ran, an assert inside a non-suspend callback — ReceiveTurbine exposes non-suspending takers that require the event to be buffered already and throw if it is not:

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()
}

Prefer suspending awaitItem() inside runTest / flow.test blocks; reach for takeItem() only when the call site genuinely cannot suspend.

Turbine timeout vs runTest virtual time

Turbine's 3-second default is wall clock, never virtual. runTest's 60-second default is also wall clock — but advanceTimeBy(...) inside the test body manipulates TestCoroutineScheduler virtual time, which Turbine does not see.

Implication: a test using delay(10.minutes) finishes in microseconds because runTest skips the delay. But a flow.test { awaitItem() } that waits for an emission whose source is real network I/O still observes real wall-clock time. If the wait is genuinely longer than 3 seconds in real time, override:

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

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