Turbine — Flow Testing

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: turbine.

Why Turbine

Testing Flow is awkward without a helper:

• Flow.toList() works for finite flows but never returns for hot/infinite ones
• Manual collect { } in test launches coroutines that race with assertions
• StateFlow/SharedFlow have replay semantics that confuse test setup

Turbine gives a deterministic, suspend-friendly API:

flow.test {
    awaitItem() shouldBe expected1
    awaitItem() shouldBe expected2
    cancelAndIgnoreRemainingEvents()
}

Used by Cash App, Square, JetBrains tooling teams. KMP-compatible.

Setup

// build.gradle.kts (or KMP commonTest)
testImplementation("app.cash.turbine:turbine:1.2.0")
testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.10.0")

// KMP
sourceSets.commonTest.dependencies {
    implementation("app.cash.turbine:turbine:1.2.0")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.10.0")
}

Basic Usage

import app.cash.turbine.test
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class FlowTest {
    @Test fun `emits 3 items`() = runTest {
        flowOf(1, 2, 3).test {
            assertEquals(1, awaitItem())
            assertEquals(2, awaitItem())
            assertEquals(3, awaitItem())
            awaitComplete()
        }
    }
}

test { } collects the flow within an isolated coroutine and exposes a DSL with assertion helpers.

API Reference

Method	Purpose
awaitItem(): T	Wait for next emission, return it
awaitItem(timeout: Duration)	With custom timeout
awaitComplete()	Assert flow completes (no error)
awaitError(): Throwable	Wait for terminal error and return it
expectNoEvents()	Assert no pending emissions / completion / error
expectMostRecentItem()	Drain all pending emissions, return latest
cancel()	Cancel collection and verify no untested events
cancelAndIgnoreRemainingEvents()	Cancel and discard remaining emissions
cancelAndConsumeRemainingEvents(): List<Event<T>>	Cancel and return remaining as list
skipItems(count: Int)	Discard N items

StateFlow Patterns

StateFlow always replays its current value to new collectors → first awaitItem() is the initial value.

@Test fun `state updates correctly`() = runTest {
    val state = MutableStateFlow(0)

    state.test {
        assertEquals(0, awaitItem())            // initial
        state.value = 1
        assertEquals(1, awaitItem())
        state.value = 2
        assertEquals(2, awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

For ViewModel state flows:

@Test fun `loads wallets and updates state`() = runTest {
    val viewModel = WalletViewModel(repo = fakeRepo)

    viewModel.state.test {
        // Initial state
        val initial = awaitItem()
        initial.isLoading shouldBe true
        initial.wallets shouldBe emptyList()

        // After load completes
        val loaded = awaitItem()
        loaded.isLoading shouldBe false
        loaded.wallets shouldHaveSize 3

        cancelAndIgnoreRemainingEvents()
    }
}

SharedFlow / Events Pattern

@Test fun `emits navigation event on save`() = runTest {
    val viewModel = WalletViewModel()

    viewModel.events.test {
        viewModel.save()
        val event = awaitItem()
        event.shouldBeInstanceOf<NavEvent.PopBack>()
        cancelAndIgnoreRemainingEvents()
    }
}

For replay-cache SharedFlow, you can use expectMostRecentItem():

@Test fun `most recent value`() = runTest {
    val flow = MutableSharedFlow<Int>(replay = 1)
    flow.emit(1)
    flow.emit(2)
    flow.emit(3)

    flow.test {
        // 3 was the last emission; replay buffer of 1 holds 3
        assertEquals(3, expectMostRecentItem())
        cancelAndIgnoreRemainingEvents()
    }
}

Combine / Map / Filter

@Test fun `combine emits when either source changes`() = runTest {
    val a = MutableStateFlow("foo")
    val b = MutableStateFlow(1)

    combine(a, b) { x, y -> "$x:$y" }.test {
        assertEquals("foo:1", awaitItem())

        a.value = "bar"
        assertEquals("bar:1", awaitItem())

        b.value = 2
        assertEquals("bar:2", awaitItem())

        cancelAndIgnoreRemainingEvents()
    }
}

Debounce / FlatMapLatest (Virtual Time)

runTest provides virtual time. Use advanceTimeBy and advanceUntilIdle:

@Test fun `debounce filters fast input`() = runTest {
    val source = MutableSharedFlow<String>()

    source.debounce(300).test {
        source.emit("a"); advanceTimeBy(100)
        source.emit("ab"); advanceTimeBy(100)
        source.emit("abc"); advanceTimeBy(400)

        assertEquals("abc", awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

@Test fun `flatMapLatest cancels previous`() = runTest {
    val query = MutableStateFlow("")

    query
        .debounce(300)
        .flatMapLatest { q -> searchApi(q) }
        .test {
            query.value = "btc"
            advanceTimeBy(300)
            val first = awaitItem()
            first shouldContain "btc"

            query.value = "eth"
            advanceTimeBy(300)
            val second = awaitItem()
            second shouldContain "eth"

            cancelAndIgnoreRemainingEvents()
        }
}

Errors

@Test fun `propagates upstream error`() = runTest {
    flow<Int> {
        emit(1)
        throw IllegalStateException("boom")
    }.test {
        assertEquals(1, awaitItem())
        val error = awaitError()
        error.shouldBeInstanceOf<IllegalStateException>()
        error.message shouldBe "boom"
    }
}

@Test fun `handles error and continues`() = runTest {
    val source = flow {
        emit(1)
        throw IOException()
    }

    source.catch { emit(-1) }.test {
        assertEquals(1, awaitItem())
        assertEquals(-1, awaitItem())
        awaitComplete()
    }
}

Multiple Flows in One Test

@Test fun `multiple flow assertions`() = runTest {
    val state = MutableStateFlow(0)
    val events = MutableSharedFlow<String>()

    turbineScope {
        val stateTurbine = state.testIn(backgroundScope)
        val eventsTurbine = events.testIn(backgroundScope)

        assertEquals(0, stateTurbine.awaitItem())

        state.value = 1
        events.emit("changed")

        assertEquals(1, stateTurbine.awaitItem())
        assertEquals("changed", eventsTurbine.awaitItem())

        stateTurbine.cancelAndIgnoreRemainingEvents()
        eventsTurbine.cancelAndIgnoreRemainingEvents()
    }
}

Custom Timeouts

Default timeout: 1 second per awaitItem(). For slow operations:

flow.test(timeout = 10.seconds) {
    awaitItem() shouldBe expected
}

// Per-call
awaitItem(timeout = 5.seconds)

ExpectNoEvents / VerifyEmpty

Useful to confirm a flow does NOT emit during a window:

@Test fun `does not emit when input below threshold`() = runTest {
    val flow = MutableStateFlow(0)

    flow.filter { it > 10 }.test {
        flow.value = 5
        flow.value = 7
        expectNoEvents()                       // confirms no emissions

        flow.value = 11
        assertEquals(11, awaitItem())

        cancelAndIgnoreRemainingEvents()
    }
}

Patterns for ViewModel Testing

MVI / Single State

class HomeViewModelTest {
    private val testDispatcher = StandardTestDispatcher()

    @BeforeTest fun setup() {
        Dispatchers.setMain(testDispatcher)
    }

    @AfterTest fun tearDown() {
        Dispatchers.resetMain()
    }

    @Test fun `load updates state correctly`() = runTest {
        val repo = FakeRepo(wallets = listOf(testWallet()))
        val viewModel = HomeViewModel(repo)

        viewModel.state.test {
            // Initial
            awaitItem().wallets shouldBe emptyList()

            viewModel.load()

            // Loading
            val loading = awaitItem()
            loading.isLoading shouldBe true

            // Settled
            val loaded = awaitItem()
            loaded.isLoading shouldBe false
            loaded.wallets shouldHaveSize 1

            cancelAndIgnoreRemainingEvents()
        }
    }
}

Side-effect events

@Test fun `emits navigation on click`() = runTest {
    val viewModel = HomeViewModel(repo)

    viewModel.events.test {
        viewModel.openWallet("abc")
        val event = awaitItem()
        event shouldBe NavEvent.Open(WalletDetailRoute("abc"))
        cancelAndIgnoreRemainingEvents()
    }
}

Anti-Patterns

Anti-pattern	Why it's bad	Correct approach
Using flow.toList() on hot flow	Never returns	Use flow.test { }
Forgetting cancelAndIgnoreRemainingEvents()	Test fails with "untested events"	Always cancel at end
Asserting via outer-scope variables instead of awaitItem()	Race conditions	Use awaitItem inside test block
runBlocking instead of runTest	No virtual time	runTest for coroutines
Forgetting Dispatchers.setMain for ViewModelScope	Coroutines never run	Setup test dispatcher
Asserting state.value directly without test {}	Misses transitional states	Use Turbine to capture all emissions
Missing advanceTimeBy for time-based operators	Stuck waiting	Drive virtual time forward
Putting delay(real_ms) in test	Slow + flaky	Use virtual time
awaitItem() after awaitComplete()	Throws	Order matters — items first, then complete
Multiple flows without turbineScope	One blocks the other	Use testIn(scope) for parallel testing

KMP / Multiplatform

Turbine works in commonTest:

// shared/src/commonTest/kotlin/WalletViewModelTest.kt
import app.cash.turbine.test
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class WalletViewModelTest {
    @Test fun `loads wallets`() = runTest {
        val viewModel = WalletViewModel(FakeRepo())

        viewModel.state.test {
            assertEquals(UiState.Loading, awaitItem())

            viewModel.load()
            advanceUntilIdle()

            val loaded = awaitItem() as UiState.Success
            loaded.wallets shouldHaveSize 1

            cancelAndIgnoreRemainingEvents()
        }
    }
}

Runs on JVM, iOS, Desktop targets:

./gradlew :shared:jvmTest
./gradlew :shared:iosSimulatorArm64Test

Troubleshooting

Symptom	Cause	Fix
Expected event.. but got nothing within 1000ms	Slow operation, no virtual time	runTest + advanceTimeBy / longer timeout
Unconsumed events found	Flow emitted more than asserted	cancelAndIgnoreRemainingEvents()
Test passes locally, fails in CI	Real Dispatchers.Main	Dispatchers.setMain(testDispatcher) in setup
awaitItem() returns initial state instead of update	StateFlow replays current value	Either await initial then update, or use SharedFlow
Flow never completes	Hot flow (StateFlow/SharedFlow)	They never awaitComplete — always cancelAndIgnoreRemainingEvents()
combine emits twice on setup	Both sources have replay	Filter initial duplicate or use distinctUntilChanged
flatMapLatest doesn't cancel previous in test	Missing time advance between emissions	advanceTimeBy to trigger debounce + new branch

When NOT to Use This Skill

Scenario	Use Instead
Mobile E2E	testing/maestro
Compose snapshot	testing/compose-snapshot
Generic Kotlin tests	testing/kotest
Pure suspend (no Flow)	kotlinx-coroutines-test runTest
Java reactive (RxJava)	RxJava test schedulers