Hi guys, I'm trying to add unit tests to the State...

# coroutines

[Michal Klimczak]

Hi guys, I'm trying to add unit tests to the StateFlow and it's awfully unpredictable (to me). Can you explain to me based on this example (the test passes like this):

@Test
    fun `stateIn self-contained example`() = runBlockingTest {

        suspend fun makeHeavyRequest(): String {
            return "heavy result"
        }

        val flow = flow<Unit> {} //in production this is a channel meant to refresh the Flow
            .onStart { emit(Unit) }
            .map { makeHeavyRequest() } //using mapLatest breaks 
            .flowOn(testDispatcher)
//            .onEach {  } //uncommenting this line also breaks
            .stateIn(GlobalScope, SharingStarted.WhileSubscribed(), "init state")

        val results = mutableListOf<String>()
        val job = launch {
            flow.collect { results.add(it) }
        }

        assertEquals("heavy result", results[0])

        job.cancel()

    }

1. Why the test breaks (I get init state instead of heavy result) when I use mapLatest instead of map?
2. Why the test breaks (I get init state instead of heavy result) when I uncomment the onEach below the flowOn?
3. Bonus question: why don't I get both, the init state AND the heavy result?

[Asagald]

I am not an expert, but I think GlobalScope you use in stateIn is the problem

[Shalom Halbert]

An alternative way to test can be achieved by using Square’s Turbine. One issue that is faced with StateFlow and SharedFlow that you do not face with other Flow s is that they never complete. So, you need a system for closing the flow after your assertion is completed.

[Michal Klimczak]

Thanks guys. What this is meant to do in production is to hold some global state as a singleton with lifecycle tied to app's lifecycle. Think SQL database in Android app. GlobalScope seems to be the only way to achieve that, I think?

[Michal Klimczak]

@elizarov you seemed to agree with the GlobalScope comment. Can you tell me if there's a better way to achieve that? Or should I rather go to stackoverflow with that question?

[Michal Klimczak]

@Shalom Halbert I really use com.github.ologe:flow-test-observer for testing that infinite flow, but it behaves exactly the same as this original code and I wanted to rule out some unexpected library behavior, hence I used this simplistic approrach. But Turibne seems liek a nice tool, thanks

[Shalom Halbert]

It’s likely GlobalScope is an issue because it means that Flow is not being run within the same CoroutineScope as runBlockingTest which probably created a race-condition.

Assuming runBlockingTest makes the asynchronous work run synchronously, which is the conventional implementation: Launching the Flow from the `runBlockingTest`’s CoroutineScope should help with running a cold Flow by ensuring the Flow runs to completion before the test ends. However, with hot `Flow`s like StateFlow that doesn’t work because the Flow never completes. That’s where Turbine became convenient for me. You can call expectItem() and cancelAndIgnoreRemainingEvents().

[Michal Klimczak]

Thank you, I will give it a spin (pun intended ;))

[Michal Klimczak]

Unfortunately, there is no difference wheter I cancel the job manually or with turbine

@Test
    fun `stateFlow (turbine)`() = runBlockingTest {

        suspend fun makeHeavyRequest(): String {
            return "heavy result"
        }

        val flow = flow<Unit> {}
            .onStart { emit(Unit) }
            .map { makeHeavyRequest() }
            .onEach { logThread("before flowOn") }
            .flowOn(testDispatcher)
            .onEach { logThread("after flowOn") } //why this changes the dispatcher for the whole thing?
            .stateIn(this, SharingStarted.WhileSubscribed(), "init state")

        flow.test {
            assertEquals("heavy result", expectItem())
            cancelAndIgnoreRemainingEvents()
        }

    }

When I do this it also finishes with kotlinx.coroutines.test.UncompletedCoroutinesError: Test finished with active jobs: ["coroutine#6":StandaloneCoroutine{Active}@409bf450]

[Michal Klimczak]

I might try too much to make it look like rxjava. The threading in Flow seems to be fundamentally different... I posted question on SO, because I honestly don't know what else I can do to understand what happens here.
https://stackoverflow.com/questions/64823459/oneach-changes-the-dispatcher-in-stateflow-kotlin-coroutines