The conflation problem of testing StateFlows

StateFlow has a bit of a wave–particle duality. On one hand, it's a data holder which holds a current value. On the other hand, it's also a Flow, emitting the values it holds over time to its collectors. Importantly, as the type's documentation states:

Updates to the value are always conflated. So a slow collector skips fast updates, but always collects the most recently emitted value.

This conflation means that depending on how "fast" the code setting values in the StateFlow and the code collecting values from it are relative to each other, the collector may or may not receive intermediate values when the StateFlow's value is rapidly updated multiple times.

In production code, this generally shouldn't cause any issues. The collector should not be affected by conflation happening or not happening. The end result - for example, what's displayed on the UI - should remain the same. Testing, however, is a different story.

Testing StateFlows

There are two approaches for testing a StateFlow: you can either make assertions on its value property, or collect it as a Flow and assert on the values collected.

This article assumes that you're familiar with coroutine testing using the kotlinx-coroutines-test library.

Generally speaking, it's simpler to treat StateFlow as a state holder and make assertions on its value after performing actions in the test, as described in the Android documentation about testing StateFlows.

However, there might be cases where this is not feasible to do, and you need to collect from a StateFlow in a test. For example, a single action on a class under test may trigger multiple value writes on the StateFlow, and you may want to assert on all intermediate values written (instead of just asserting on value once, after the last write).

A typical case of this is testing a ViewModel function that first triggers a loading state and then replaces that with actual data (oversimplified pseudocode!):

class MyViewModel(private val repo: Repo) : ViewModel() {
    val state: MutableStateFlow<MyState> = MutableStateFlow(Initial)

    fun initialize() {
        state.value = Loading
        state.value = Content(repo.getData())
    }
}

In this specific example, you might be able to write a test which only reads the value property of the StateFlow, by injecting a fake repository implementation that lets us control when getData returns, giving us a chance to read state.value while it's still in the loading state. However, it's often not possible or too cumbersome to control the object under test this way.

Experiencing conflation

If you choose to test a StateFlow by collecting it, you'll need to take conflation into account. When writing a test asserting on values collected from a StateFlow, you have to decide whether you expect to see all values without conflation, or to experience conflation and only see the most recent value after rapid updates.

Let's see what happens if you start two coroutines in a test using runTest, one collecting from the StateFlow and another setting values in it. In this case, they'll be equally "fast", as they'll both inherit the StandardTestDispatcher from runTest.

We'll call the first coroutine the collecting coroutine, and the second coroutine the producing coroutine.

@Test
fun useStandardTestDispatcherForCollection() = runTest {
    val stateFlow = MutableStateFlow(0)

    // Collecting coroutine
    launch {
        val values = stateFlow.take(2).toList()
        assertEquals(listOf(0, 3), values) // Conflation happened
    }

    // Producing coroutine
    launch {
        stateFlow.value = 1
        stateFlow.value = 2
        stateFlow.value = 3
    }
}

As they are launched on a StandardTestDispatcher, both coroutines here will first be queued up on the test scheduler. When runTest reaches the end of the lambda passed to it, it will start executing the tasks queued up on the scheduler.

This starts the collecting coroutine first, which begins collecting the StateFlow. The StateFlow immediately emits its initial value, which is collected into the list. The collecting coroutine then suspends as toList waits for new values.

This lets the second coroutine start, which sets the value of the StateFlow several times. Setting the value property of StateFlow is a regular property assignment, which is not a suspending call, however its implementation does resume any collecting coroutines.

Though the collector is resumed, it's on a StandardTestDispatcher and still needs a chance to dispatch in order to run its code, which it doesn't get until the test thread is yielded. That only happens on the completion of the producing coroutine, after all three value writes are done.

With the completion of the producing coroutine, the collector gets to execute, and receives only the latest value, which it places in the list. Conflation happened.

Avoiding conflation

Let's see what you can do if you want to avoid conflation in our tests.

Note that conflation is not inherently bad. However, it's often undesirable in testing scenarios that want to verify all the behaviour of an object.

Yielding manually

For a simple approach, if you let go of the test thread after each value assignment - for example, using a simple yield() call - the collecting coroutine gets a chance to dispatch, and receives all four values, eliminating conflation:

@Test
fun yieldingExample() = runTest {
    val stateFlow = MutableStateFlow(0)

    launch {
        val values = stateFlow.take(4).toList()
        assertEquals(listOf(0, 1, 2, 3), values) // No conflation
    }

    launch {
        stateFlow.value = 1
        yield()
        stateFlow.value = 2
        yield()
        stateFlow.value = 3
    }
}

This doesn't scale very well, as it requires explicitly yielding the thread every time a new value is set in the StateFlow.

To make things worse, in a real test this likely happens somewhere in code that's called from the test, and not within the test itself. Modifying the scheduling behaviour of production code to make testing more convenient isn't great.

Collecting faster

Alternatively, you can change how "fast" the collecting coroutine is, so that it can better keep up with new values being produced. In the next example, the collecting coroutine is created using an eager UnconfinedTestDispatcher while the producing coroutine keeps using the lazier StandardTestDispatcher.

For an explanation of the scheduling of these two dispatchers, see the Android docs on TestDispatchers.

@Test
fun useUnconfinedTestDispatcherForCollection() = runTest {
    val stateFlow = MutableStateFlow(0)

    launch(UnconfinedTestDispatcher(testScheduler)) {
        val values = stateFlow.take(4).toList()
        assertEquals(listOf(0, 1, 2, 3), values) // No conflation
    }

    launch {
        stateFlow.value = 1
        stateFlow.value = 2
        stateFlow.value = 3
    }
}

In this test, collection happens immediately whenever the assignment of value resumes the collecting coroutine, thanks to UnconfinedTestDispatcher which doesn't require a dispatch before resuming execution. Conflation is gone!

Using UnconfinedTestDispatcher for the collecting coroutine also has the added benefit that the collecting coroutine is launched eagerly. This means that by the time the first launch call of the test returns the first coroutine has already started executing, collected the initial value of the StateFlow, and is suspended waiting for new values to be produced.

Using the Turbine library instead of collecting from the Flow yourself behaves the exact same way, as Turbine's test function also uses UnconfinedTestDispatcher under the hood if you're using it with runTest.

@Test
fun useTurbineForCollection() = runTest {
    val stateFlow = MutableStateFlow(0)

    launch {
        stateFlow.test { // No conflation
            assertEquals(0, awaitItem())
            assertEquals(1, awaitItem())
            assertEquals(2, awaitItem())
            assertEquals(3, awaitItem())
        }
    }

    launch {
        stateFlow.value = 1
        stateFlow.value = 2
        stateFlow.value = 3
    }
}

Note that your collecting coroutine can only go faster if the producing coroutine is on a "slow" StandardTestDispatcher. If the producing coroutine runs on an UnconfinedTestDispatcher and does not yield the thread at any point, conflation makes a comeback, and you have no chance of seeing those intermediate values.

Here's an example demonstrating just that:

@Test
fun useUnconfinedTestDispatcherForCollectionAndProduction() = runTest {
    val stateFlow = MutableStateFlow(0)

    launch(UnconfinedTestDispatcher(testScheduler)) {
        val values = stateFlow.take(2).toList()
        assertEquals(listOf(0, 3), values) // Conflation happened
    }

    launch(UnconfinedTestDispatcher(testScheduler)) {
        stateFlow.value = 1
        stateFlow.value = 2
        stateFlow.value = 3
    }
}

Conclusion

When testing a StateFlow, you have two options: reading its value at various points during the test or collecting it as a Flow. The former is the easier path if your circumstances allow it. Otherwise, you'll have to keep in mind that collectors of a StateFlow are affected by conflation.

It's up to you to decide whether you want values to be conflated or not during a test, and set up assertions accordingly. You can control whether conflation can happen by choosing the dispatchers your code executes on during tests. Using injected dispatchers that you can replace with appropriate TestDispatcher instances during tests is crucial for this.

Finally, the only way your collecting coroutine can be "faster" than the producing coroutine and avoid conflation is if the collector is on an UnconfinedTestDispatcher while the producer is on a StandardTestDispatcher. Other cases—where the coroutines are equally fast or the collector is slower—will result in conflation.