Turbine Testing

Testing Flows in your app can be confusing, especially when they’re interconnected with multiple data sources and emissions coming from multiple places. In this second part of our Coroutine testing series we’ll look deeper into how Turbine works with different types of Flows. Then in the next part we’ll cover how testing coroutines changes in KMP and finally I’ll cover how this can be made into a skill for AIs to use when testing your code. This post assumes you have basic knowledge of Turbine.

The recommendations in this post are based on my experience testing production kotlin projects and verified with intentionally flaky tests running hundreds of times (some under heavy cpu load) using Turbine 1.2.0 and kotlinx-coroutines 1.10.1.

We’ll go over how to use the following functions to run tests:

From the coroutine testing library:

• advanceUntilIdle - Runs all other coroutines on the scheduler until there is nothing left in the queue.
• advanceTimeBy - Advances virtual time by the given amount and runs any coroutines scheduled to run before that point in virtual time.
• runCurrent - Runs coroutines that are scheduled at the current virtual time.

From Turbine:

• awaitItem - Asserts a single emitted value
• awaitComplete - Asserts the flow completed
• awaitError - Asserts a Flow Failure

Explanation of Flows

First to best understand how to test let’s have a refresher on the different kinds of flows available to us:

• A basic Flow is a cold asynchronous data stream that emits values over time.
• A SharedFlow is a hot flow that emits values to all consumers that collect from it.
• A StateFlow is a hot state-holder observable flow that emits the current and new state updates to its collectors. It’s commonly used for UI state in ViewModels so it’s an important one to know how to test.
• Bonus: A LiveData is not a flow, but it is a hot state-holder. Its asFlow() function returns a cold Flow view of it. Unlike a StateFlow an empty LiveData has no value to emit.

A cold stream does not start producing values until one starts to collect them. A hot stream on the other hand produces values immediately regardless of whether anyone is collecting.

This is important to understand as this will affect how you want to test your emissions.

State-Holder

The state-holder distinction is important, as it changes how you want to test the flow. For Event streams you can assert each emission using awaitItem and awaitComplete, however for a state-holder flow it’s better to assert the outcome rather than every intermediate state (i.e. assert state.value). This is because those states can become conflated. This is even recommended in the official Android docs:

Note that this stream of values is conflated, which means that if values are set in a StateFlow rapidly, collectors of that StateFlow are not guaranteed to receive all intermediate values, only the most recent one. We generally recommend treating StateFlow as a data holder and asserting on its value property instead. This way, tests validate the current state of the object at a given point in time, and don’t depend on whether or not conflation happens.

With a state-holder flow intermediates can be conflated, identical values are swallowed, and postValue can coalesce states. In other words the flow may never emit that loading or intermediate state, and just skip to the last state. Turbine collects with an unconfined dispatcher so it usually sees every value, even ones set back to back (it saw every value in every run when they were set from the test thread, a Standard producer or a Default producer). Values sometimes disappear upstream, such as when stateIn or flowOn runs on a StandardTestDispatcher, when a producer on an Unconfined dispatcher sets values back to back, or when LiveData coalesces postValue calls. It’s important to keep in mind that most of the time when testing you really only care about the end result anyway. Test what you need, not what you have.

class SearchViewModel(
    private val repository: UserRepository,
    dispatcher: CoroutineDispatcher,
) {
    private val scope = CoroutineScope(SupervisorJob() + dispatcher)

    private val _state = MutableStateFlow<UiState>(UiState.Idle)
    val state: StateFlow<UiState> = _state.asStateFlow()

    fun search(query: String) {
        scope.launch {
            _state.value = UiState.Loading
            runCatching { repository.load(query) }
                .onSuccess { _state.value = UiState.Success(it) }
                .onFailure { _state.value = UiState.Error(it.message ?: "Unknown") }
        }
    }
}

val vm = SearchViewModel(repo, UnconfinedTestDispatcher(testScheduler))
vm.state.test {
    awaitItem()               // Idle
    vm.search("ada")
    awaitItem()               // Success: Loading was never delivered
}

In the example above, the ViewModel on an UnconfinedTestDispatcher skips the Loading state entirely, because the producer runs straight through to Success before Turbine can see the in-between state. The Kotlinx docs tell us that, like Dispatchers.Unconfined, UnconfinedTestDispatcher doesn’t guarantee execution order when several coroutines are queued, so what Turbine sees can differ from what you expect.

The same updates made from the test body, or from a coroutine on a StandardTestDispatcher or Dispatchers.Default, are all delivered.

If you do need to wait for a specific state, such as after an asynchronous call, you keep awaiting until you see it:

var state = awaitItem()
while (state !is Loaded) {
    state = awaitItem()
}

There is a PR up for adding awaitUntil in the Turbine repo which would make this kind of await official, but it’s still open.

The loop should check for the state you want, not an intermediate one like Loading that may never arrive.

If you really do need to test the intermediate states, such as a loading UI state, then you can assert it but first make the timing deterministic by using a StandardTestDispatcher or having a sync point before continuing. A rule of thumb is to ask: “is the sequence part of the contract?”. If yes, assert it and control the timing. If no, assert the outcome.

If your flow has operators that drop values on purpose (such as debounce, sample, or mapLatest), then you should assert exactly what the operator is meant to let through.

Here’s a table for reference:

Kind of flow	What to assert
State holders: StateFlow, LiveData, anything derived with stateIn	The outcome (.value). Intermediates can be conflated, identical values are swallowed, and postValue coalesces.
Event streams: cold Flow, SharedFlow used for one-time events (i.e. navigation, toasts), Channel, callbackFlow, Rx’s PublishSubject	Each emission, in order (awaitItem(), …, awaitComplete()).
Operators that drop values on purpose (debounce, sample, mapLatest)	Exactly what the operator is meant to let through.
The intermediate is the behavior (i.e. a Loading the user must see, an event that must not fire twice)	Assert it, but first make the timing deterministic. Use things like StandardTestDispatcher, a repository that parks until released, runCurrent().

In general most flows behave well with Turbine. Turbine can see every value from a Channel, callbackFlow, RxJava subjects converted with asFlow(), and operators like conflate() or distinctUntilChanged() (none of them dropped a value in my tests). Keep in mind a cold flow re-runs every time it’s collected, so collecting it twice repeats its side effects.

Sync Points

Sometimes it’s important to have explicit sync points, so that you can make sure that the work you expect to be done is done. Some Turbine calls don’t wait, so without a sync point they can run before your async work has finished. A sync point forces the queued test-dispatcher work to run first. After that, the buffer is in a known state and the check means something. In my tests expectNoEvents() passed every time even though the event arrived right afterwards, and expectMostRecentItem() returned the stale value in nearly every. With a runCurrent() sync point the result was identical in all the runs.

We can use a combination of Coroutine and Turbine calls to ensure that our work is synced. See the example below:

sealed interface FeedState {
    data object Idle : FeedState
    data object Loading : FeedState
    data object Loaded : FeedState
}

class FeedViewModel(
    private val repo: FeedRepository,
    private val dispatcher: CoroutineDispatcher,
) : ViewModel() {

    private val _state = MutableStateFlow<FeedState>(FeedState.Idle)
    val state: StateFlow<FeedState> = _state.asStateFlow()

    fun refresh() {
        viewModelScope.launch(dispatcher) {
            _state.value = FeedState.Loading
            repo.fetchFeed()
            _state.value = FeedState.Loaded
        }
    }
}

runTest {
    val vm = FeedViewModel(repo, StandardTestDispatcher(testScheduler))

    vm.state.test {
        awaitItem()                    // initial value
        vm.refresh()                   // launches a coroutine; nothing has run yet

        expectNoEvents()               // true: the launch is still queued
        runCurrent()                   // sync point: the refresh coroutine runs now
        val expectedItem = expectMostRecentItem()
        assertEquals(Loaded, expectedItem)
        cancelAndIgnoreRemainingEvents()
    }
}

Let’s go over what happens:

1. refresh() launches a coroutine, but within the test nothing has run yet
2. expectNoEvents() - Asserts that there are no unconsumed events which have already been received. This is because the launch of the coroutine is still queued, and since we have a virtual time from the testScheduler we haven’t elapsed any time yet.
3. runCurrent() - This runs the scheduled coroutines in the virtual time
4. expectMostRecentItem() - Returns the most recent buffered item immediately, skipping any older ones (It’s meant for state-like flows, as it will throw on a fresh SharedFlow or an empty LiveData). Use awaitItem() if you want to wait for a specific next item.

This way we are explicitly defining the status of the flow before and after the function is called. It’s important to note that you want to use runCurrent() here instead of advanceUntilIdle(), because we only want to run the task that is queued now (i.e. the refresh). This is assuming the repository doesn’t delay. If that’s the case you’ll need advanceUntilIdle() to reach Loaded. Here’s a breakdown.

Function	What it does	Use it when
runCurrent()	Runs tasks due now (including backgroundScope work) and stops. The virtual clock does not move.	You need a sync point without moving time: let a launched coroutine start and settle, keep a Loading state visible, settle stateIn on backgroundScope, or before expectNoEvents() / expectMostRecentItem().
advanceUntilIdle()	Runs everything until nothing is left, jumping the clock past every delay. Skips backgroundScope work.	You only care about the final state and any delay (debounce, timeout, retry) is allowed to fire: “run it all to the end”.
advanceTimeBy(n)	Moves the clock forward by n and runs tasks scheduled before that point, not at exactly n.	You are testing a time boundary. Follow it with runCurrent() to run tasks due exactly at n.

Rules of thumb:

• Prefer runCurrent() when you want to look at an in-between state, advanceUntilIdle() cannot show you one because it skips past it.
• Use advanceUntilIdle() when the end state is all you assert. Be aware that it also fires debounces and timeouts.
• Use advanceTimeBy + runCurrent() for exact time boundaries.
• Don’t use advanceUntilIdle() for backgroundScope work (it skips it) or for a periodic loop that never ends (it would never return).

None of these three calls wait for a real dispatcher.

Best Practices

Beware how you change state

Sometimes new states can be skipped if they’re not handled properly. If you emit an identical state twice there will be no emission on the second state (my test saw 0 events for the second identical result), and if you’re awaiting that second state it can cause hangs or unexpected failures. Similarly, if you modify the object that a MutableStateFlow holds in place (i.e. adding to a list and reassigning the same instance) there will be no emission, because it’s the same object. Create a new instance instead, and prefer asserting .value so the test doesn’t depend on whether an emission happens.

Understand shareIn, stateIn and replays

Sometimes you need to use shareIn when several collectors need to share one upstream, such as when one expensive stream like a WebSocket connection needs to be collected by multiple observers. In these cases shareIn is the right tool for the job, but be careful with replay = 0 on a StandardTestDispatcher. The sharing coroutine is queued and hasn’t started collecting yet, so anything emitted from a hot upstream in that gap is dropped and Turbine sees nothing. Either start the collection first using runCurrent(), or run the sharing on an Unconfined dispatcher so collection starts immediately.

stateIn on a StandardTestDispatcher has the same issue. Only the last value makes it through (in every variant I tried: Eagerly, Lazily and WhileSubscribed), so awaiting Loading hangs:

val derived = source.map { it }
    .stateIn(
        CoroutineScope(StandardTestDispatcher(testScheduler)),
        SharingStarted.Eagerly,
        UiState.Idle
    )

derived.test {
    awaitItem()                         // Idle
    source.value = UiState.Loading
    source.value = UiState.Success("Ada")
    advanceUntilIdle()
    // only Success arrives
}

This only affects a hot upstream. A cold flow doesn’t emit anything until collection starts. (With replay = 0 a subscriber that attaches after an emission still misses it.)

Use Turbine Scope Properly

Turbine includes testIn for testing multiple flows at once (called inside a turbineScope). It can also be used to collect on the same flow multiple times if you want to test different parts of the same flow. This should be used instead of nesting, as nesting can cause ambiguity and confusion between the two Turbines (It’s not that nesting test {} blocks is racy, it’s just hard to read). When one action updates two streams start both collectors first, then act.

Do

runTest { // backgroundScope comes from runTest
    turbineScope {
        val turbineA = flowA.testIn(backgroundScope)
        val turbineB = flowB.testIn(backgroundScope)
        viewModel.doThing()
        assertEquals(..., turbineA.awaitItem())
        assertEquals(..., turbineB.awaitItem())
        turbineA.cancelAndIgnoreRemainingEvents()
        turbineB.cancelAndIgnoreRemainingEvents()
    }
}

Don’t

flowA.test {
    flowB.test {
        viewModel.doThing()
        assertEquals(..., awaitItem()) // which Turbine?
    }
    assertEquals(..., awaitItem())
}

Start Turbine before acting

To ensure you don’t miss any emissions, make sure you call .test before performing anything that changes the state. This matters mostly for a SharedFlow, which drops events emitted before Turbine starts collecting. A StateFlow doesn’t need it, but there’s no harm in doing it anyway. In my test the event emitted before Turbine started was lost every time.

// Don't
viewModel.send("Navigated")
viewModel.events.test { 
    expectNoEvents(); 
    cancelAndIgnoreRemainingEvents()
}

// Do
viewModel.events.test {
    viewModel.send("Navigated")
    assertEquals("Navigated", awaitItem())
    cancelAndIgnoreRemainingEvents()
}

This goes for multiple flows when using turbineScope as well.

Do

turbineScope {
    val turbineA = flowA.testIn(backgroundScope)
    val turbineB = flowB.testIn(backgroundScope)
    viewModel.doThing()
    turbineA.awaitItem()
    viewModel.doAnotherThing()
    turbineB.awaitItem()
    turbineA.cancelAndIgnoreRemainingEvents()
    turbineB.cancelAndIgnoreRemainingEvents()
}

Don’t

flowA.test { 
    viewModel.doThing()
    awaitItem()
}
viewModel.doAnotherThing()
flowB.test {
    awaitItem()
}

Know the first item

A StateFlow always has an initial value when it’s created either from stateIn or MutableStateFlow. This means the first awaitItem() is that initial value when it’s created, but also keep in mind that it’s a hot flow so depending on when the Turbine tests start the value might look different. For example if you are updating that state in the init of a ViewModel then by the time you start Turbine the state might’ve changed. The same goes for stateIn with WhileSubscribed: a new collector’s first item can be the stale cached value, since the upstream only restarts right away with a stop timeout of 0.

If a SharedFlow or LiveData has not emitted any items yet when expectMostRecentItem() is called the test will throw an AssertionError. It won’t throw on a StateFlow, since its initial value is already buffered.

You need to end the stream

Be aware of how you end the test. If a flow ends up emitting more items after your test that you don’t care about, be sure to call cancelAndIgnoreRemainingEvents (or cancelAndConsumeRemainingEvents if you want to log the remaining emits). In my tests an item that was still buffered when test {} ended failed the test every time, while a late event failed only occasionally (3 out of 300 runs in one setup), which is the flaky kind. Ending with cancelAndIgnoreRemainingEvents() had 0 failures when testing.

If you want to confirm nothing else appears use expectNoEvents, but remember it only checks what’s already buffered so put a sync point (runCurrent()) before it. If it’s a single use flow then call awaitComplete or awaitError so you confirm it’s done.

Stream type	Typical end
StateFlow / LiveData.asFlow()	cancelAndIgnoreRemainingEvents()
SharedFlow	cancelAndIgnoreRemainingEvents()
Finite cold Flow	awaitComplete() or awaitError()
Continuous cold / Room-style Flow	cancelAndIgnoreRemainingEvents()

Utilize Timeouts

As mentioned in the last post, when you are testing using real threads be sure to utilize timeouts in Turbine. If you don’t care how long tests run use a larger timeout value, especially if a run is in CI. Turbine’s timeout is wall-clock: in my tests a virtual delay(10.seconds) passed a 500 ms timeout, while a real 800 ms delay failed it. The timeout defaults to 3 seconds, but you can pass in a custom timeout value when calling test.

flow { withContext(Dispatchers.Default) { delay(800) }; emit(1) }
    .test(timeout = 10.seconds) {          // virtual delays don't need this; real ones do
        assertEquals(1, awaitItem())
        awaitComplete()
    }

Conclusion

In this post we covered tips on how to effectively use Turbine to avoid flakiness. Hopefully this has given some insight into how Turbine should be used with various flows, and some of the best practices when writing Turbine tests. In the next post we’ll create AI Skills to write tests and audit existing tests using best practices.