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

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

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

When NOT to Use This Skill