kotlinx-coroutines-test:1.6.1- Issue with collecting values of StateFlow in tests

[Subuday]

There are collected only the first and the last values. It's an Android application. mainDispatcherRule is used to replace Dispatcher.Main -> UnconfinedTestDispatcher. The idea is to collect all emitted values from StateFlow. But the intermediate values are skipped. I can not grab the problem. I've referenced to this implementation

@ExperimentalCoroutinesApi
class BaseFeatureTest {

    @get:Rule
    val mainDispatcherRule = MainDispatcherRule()

    @Test
    fun `states and news are emitted correctly`() = runTest {
        val sut = createBaseFeature()
        val values = mutableListOf<FakeState>()
        val job = launch(UnconfinedTestDispatcher(testScheduler)) {
            sut.collect {
                values += it
            }
        }

        sut.emit(FakeWish.Increment)

        println(values) // [FakeState(counter=100), FakeState(counter=102)]
        val expectedValues = listOf(FakeState(100), FakeState(101), FakeState(102))
        assertContentEquals(expectedValues, values)

        job.cancel()
        sut.cancel()
    }

    private fun createBaseFeature() = BaseFeature(
        initialState,
        FakeActor(),
        FakeReducer(),
        FakeNewsPublisher()
    )
}

data class FakeState(val counter: Int)
sealed interface FakeWish {
    object Increment : FakeWish
}
sealed interface FakeEffect {
    object TestEffect : FakeEffect
}
sealed interface FakeNews

class FakeActor : Actor<FakeState, FakeWish, FakeEffect> {

    override fun invoke(state: FakeState, wish: FakeWish): Flow<FakeEffect> {
        return when(wish) {
            is FakeWish.Increment -> flowOf(FakeEffect.TestEffect, FakeEffect.TestEffect)
        }
    }
}

class FakeReducer : Reducer<FakeState, FakeEffect> {

    override fun invoke(state: FakeState, effect: FakeEffect): FakeState {
        return when(effect) {
            is FakeEffect.TestEffect -> {
                state.copy(counter = state.counter + 1)
            }
        }
    }
}

class FakeNewsPublisher : NewsPublisher<FakeWish, FakeEffect, FakeState, FakeNews> {

    override fun invoke(wish: FakeWish, effect: FakeEffect, state: FakeState): FakeNews? {
        return null
    }
}

val initialState = FakeState(counter = 100)

open class BaseFeature<Wish : Any, Effect : Any, State : Any, News : Any>(
    initialState: State,
    private val actor: Actor<State, Wish, Effect>,
    private val reducer: Reducer<State, Effect>,
    private val newsPublisher: NewsPublisher<Wish, Effect, State, News>
) : Flow<State>, FlowCollector<Wish>, ViewModel() {

    private val stateFlow = MutableStateFlow(initialState)
    private val wishChannel = Channel<Wish>()

    private val _newsChannel = Channel<News>()
    val news = _newsChannel.receiveAsFlow()

    private val collectJobWish: Job

    init {
        collectJobWish = wishChannel
            .receiveAsFlow()
            .onEach { wish -> processWish(stateFlow.value, wish) }
            .launchIn(viewModelScope)
    }

    override suspend fun collect(collector: FlowCollector<State>) {
        stateFlow.collect(collector)
    }

    override suspend fun emit(value: Wish) {
        wishChannel.send(value)
    }

    fun cancel() {
        viewModelScope.cancel()
    }

    private suspend fun processWish(state: State, wish: Wish) {
        coroutineScope {
            actor(state, wish)
                .onEach { effect -> processEffect(stateFlow.value, wish, effect) }
                .launchIn(this)
        }
    }

    private suspend fun processEffect(state: State, wish: Wish, effect: Effect) {
        val newState = reducer(state, effect)
        stateFlow.value = newState
        publishNews(wish, effect, newState)
    }

    private suspend fun publishNews(wish: Wish, effect: Effect, state: State) {
        newsPublisher.invoke(wish, effect, state)?.let {
            _newsChannel.send(it)
        }
    }
}

[tom2048]

I think I've got the same issue. In my case it looks like a problem with 'runTest' function. Please take a look at the code below:

package com.example.app

import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runBlockingTest
import kotlinx.coroutines.test.runTest
import org.junit.Test

class ExampleFlowTest {

    @Test
    fun correctTest() = runBlockingTest {
        val stateFlow = MutableStateFlow<Int>(0)
        val result = arrayListOf<Int>()

        val observerJob = launch {
            stateFlow.collect {
                result.add(it)
                println("Value sent: $it")
            }
        }

        launch {
            flowOf(1, 2, 3)
                .onEach {
                    stateFlow.value = it
                }
                .collect()
        }
        observerJob.cancel()
        assert(result == arrayListOf(0, 1, 2, 3))
    }

    @Test
    fun incorrectTest() = runTest(UnconfinedTestDispatcher()) {
        val stateFlow = MutableStateFlow<Int>(0)
        val result = arrayListOf<Int>()

        val observerJob = launch {
            stateFlow.collect {
                result.add(it)
                println("Value sent: $it")
            }
        }

        launch {
            flowOf(1, 2, 3)
                .onEach {
                    stateFlow.value = it
                }
                .collect()
        }
        observerJob.cancel()
        assert(result == arrayListOf(0, 1, 2, 3))
    }

}

Both functions are basically the same and should send four emissions to the state flow (emissions: 0, 1, 2, 3). The only difference is, that in the first case I use deprecated function 'runBlockingTest' (it works correctly), and in the second case I use new 'runTest' function (it fails, it sends only emissions: 0, 3).

It would be greatly appreciated if this could be improved in 'runTest'. @Subuday, maybe for now you could use 'runBlockingTest' as a workaround.

[mhernand40]

@tom2048 what if after stateFlow.value = it, you either call yield() or runCurrent()? Does the test now pass? I suspect this is very specific to MutableStateFlow's behavior in that it may be conflating the emissions during the test run.

[tom2048]

@mhernand40, adding yield() after stateFlow.value = it makes emissions work correctly, on the other hand runCurrent() does not.
However, similar code which changes stateFlow value is a part of my application code, not the test code. It's responsible for changing viewModel's state and the test validates it. That's why I'm not convinced to put yield() after every value change. It would be best if the behavior of the test function could be improved.

[mhernand40]

@tom2048 Understood. You definitely do not want to put yield() calls in your application code simply to get the tests to "pass".

If the tests need to assert all the values emitted by the MutableStateFlow rather than just asserting on the latest state value, may I suggest using https://github.com/cashapp/turbine. Since each emission to be asserted against is captured by the API's suspending awaitItem() call, your tests would be updated to assert as each value is emitted rather than trying to collect them all in a List and then asserting afterward.

[dkhalanskyjb]

@tom2048 please see #3339. In short, runTest is working exactly as intended because MutableStateFlow does conflate successive emissions, which you may observe in production as well. It's runBlockingTest that's incorrect because it hides this behavior.

[tom2048]

@mhernand40, @dkhalanskyjb thanks for your answers. Indeed you're right about conflating successive emissions. However there are some cases to be tested which are not processed as successive while running an application but become successive during unit test. Please see the following code:

package com.exampleapp

import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.launch

class PreloaderExampleViewModel(private val useCase: () -> Flow<Unit>) : ViewModel() {

    private val _uiState = MutableStateFlow(UiState(false))
    val uiState: StateFlow<UiState> = _uiState

    fun loadData() {
        viewModelScope.launch {
            useCase.invoke()
                .onStart { _uiState.value = _uiState.value.copy(preloader = true) }
                .catch { _uiState.value = _uiState.value.copy(preloader = false) }
                .collect {
                    _uiState.value = _uiState.value.copy(preloader = false)
                }
        }
    }

    data class UiState(
        val preloader: Boolean
        // other state properties
        // ...
    )
}

The ViewModel above is an example of preloader visibility logic while loading data on the screen. While running the application state changes are processed after the delay and therefore are treated as separate emissions.
On the other hand in unit test useCase can be mocked as simple flowOf(Unit) and then the emissions are conflated so the behavior is different. Would you suggest proper approach to handle such cases?

[ifuatgucluer]

Thanks 3 Mar 2023 Cum 20:36 tarihinde tom2048 ***@***.***> şunu yazdı:

…

@mhernand40 <https://github.com/mhernand40>, @dkhalanskyjb <https://github.com/dkhalanskyjb> thanks for your answers. Indeed you're right about conflating successive emissions. However there are some cases to be tested which are not processed as successive while running an application but become successive during unit test. Please see the following code: package com.exampleapp import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.onStart import kotlinx.coroutines.launch class PreloaderExampleViewModel(private val useCase: () -> Flow<Unit>) : ViewModel() { private val _uiState = MutableStateFlow(UiState(false)) val uiState: StateFlow<UiState> = _uiState fun loadData() { viewModelScope.launch { useCase.invoke() .onStart { _uiState.value = _uiState.value.copy(preloader = true) } .catch { _uiState.value = _uiState.value.copy(preloader = false) } .collect { _uiState.value = _uiState.value.copy(preloader = false) } } } data class UiState( val preloader: Boolean // other state properties // ... ) } The ViewModel above is an example of preloader visibility logic while loading data on the screen. While running the application state changes are processed after the delay and therefore are treated as separate emissions. On the other hand in unit test useCase can be mocked as simple flowOf(Unit) and then the emissions are conflated so the behavior is different. Would you suggest proper approach to handle such cases? — Reply to this email directly, view it on GitHub <#3367 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/A3WKLSCFHC5BIQ2DEBANSLLW2IT3VANCNFSM53THNFVA> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[tom2048]

The solution with mocking delay(1000) suggested in #3339 didn't work in case when I wanted to test the exception thrown from the use case in the example above. However I managed to succesfully test the viewmodel when I mocked the use case this way:

coEvery { useCase.invoke() } returns flow {
    yield()
    throw TestException()
}

I guess yield() solved the problem. Thanks for your help.

[ifuatgucluer]

good job

[dkhalanskyjb]

The master issue for this topic is now #3939, closing this one in its favor.