Cannot test SharedFlow

[rakesh-sh2020]

I have an android viewmdeol class with the following property

private val _trainingNavigationEvents = MutableSharedFlow<NavigationEventTraining>(replay = 0)
    val trainingNavigationEvents = _trainingNavigationEvents.asSharedFlow()

fun navigate(navigationEvent: NavigationEventTraining) {
        viewModelScope.launch {
            _trainingNavigationEvents.emit(navigationEvent)
        }
    }

I am using a SharedFlow as it solves the SingleLiveEvent problem.

The issue arises when I try and unit test the code. I can't see how to use turbine (or supplied primitives) to get it to work.

    @ExperimentalTime
    @Test
    fun `navigate`() = runBlockingTest {
        viewModel.handleIntent(TrainingViewModel.TrainingIntent.ShowQuestions)

        viewModel.navigationEvents.test {
            assertEquals(
                TrainingViewModel.TrainingNavigationEvent.NavigateToQuestions::class,
                expectItem()::class
            )
            cancelAndConsumeRemainingEvents()
        }
    }

and I get

kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms

I know that a SharedFlow never completes and that may be part of the reason but I have been unable to find any examples of how to do this instead.

I am using Junit 5 and am using a TestCoroutineDispatcher class extension.

[JakeWharton]

Can you turn this into an executable sample that I can run and debug locally? I don't use MutableSharedFlow, view models, runBlockingTest, or JUnit 5... so yeah it's not entirely clear what's to blame.

[kevincianfarini]

Can you share a full stacktrace? @rakesh-sh2020

I have created a simple reproduction of this bug. My above hypothesis I think is wrong. Here's my test.

  @Test fun foo() = suspendTest {
    val flow1 = MutableSharedFlow<Int>(replay = 0)
    val flow2 = flow1.asSharedFlow()
    flow1.emit(1)

    flow2.test {
      assertEquals(1, expectItem())
      cancelAndConsumeRemainingEvents()
    }
  }

with the corresponding stacktrace:

kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms
	(Coroutine boundary)
	at app.cash.turbine.ChannelBasedFlowTurbine$expectEvent$2.invokeSuspend(FlowTurbine.kt:238)
	at app.cash.turbine.ChannelBasedFlowTurbine$withTimeout$2.invokeSuspend(FlowTurbine.kt:206)
	at app.cash.turbine.ChannelBasedFlowTurbine.expectItem(FlowTurbine.kt:243)
	at app.cash.turbine.TestFoo$foo$1$1.invokeSuspend(TestFoo.kt:16)

Notice how it's failing at expectItem(FlowTurbine.kt:243)

My suspicion is that your MutableStateFlow replay size of 0 is the culprit. You're emitting an event to a flow with no replay, before you start testing the events of the flow. Your event is getting dropped because there's no subscribers. In my above program, adding a replay = 1 solved the problem. If that's not desirable for your application code, you might consider doing something like the following.

  @Test fun foo() = suspendTest {
    val flow1 = MutableSharedFlow<Int>(replay = 0)
    val flow2 = flow1.asSharedFlow()

    launch {
      flow2.test {
        assertEquals(1, expectItem())
        cancelAndConsumeRemainingEvents()
      }
    }

    flow1.emit(1)
  }

[piotrmadry]

@kevincianfarini You can try with:

mutableSharedFlow.apply {
    test {
        emit(1)
        assertEquals(1, expectItem())
        emit(2)
        assertEquals(2, expectItem())
    }
}

[dinorahto]

I know the problem here
For ViewModel you wanna do this:

The ViewModel function will be executed and the flow will be on the right track to be tested with Turbine 😉
Hope this help you

    @ExperimentalTime
    @Test
    fun `navigate`() = suspendTest {
        viewModel.navigationEvents.test {
            viewModel.handleIntent(TrainingViewModel.TrainingIntent.ShowQuestions)
            assertEquals(
                TrainingViewModel.TrainingNavigationEvent.NavigateToQuestions::class,
                expectItem()::class
            )
            cancelAndConsumeRemainingEvents()
        }
    }

[sdoward]

Your problem could be with runBlockingTest as it automatically advances time (as far as I remember)

[Kshitij09-ag]

@kevincianfarini I can confirm that launching sharedFlow tests in a separate coroutine (Using launch { } inside runBlockingTest { } ) makes it work. Do you know why it works?

[kevincianfarini]

@Kshitij09-ag it sets up a flow collector before a value is emitted to it. Once that collector is active and suspended, it will resume when a new emission from the flow is available. From there, the test assertion can be run.

This entire issue happens because StateFlow is HOT. Emitting elements to it when it doesn't have any downstream consumers drops those elements.

The solution @dinorahto provides does the exact same thing my solution does. It starts collecting the flow (with test) before an emission is made (with viewModel.handleIntent).

This issue is not an issue with Turbine, but instead a misunderstanding of the sequence of events that should happen with hot flows. It's possible the documentation should be updated to include this use case.

[kevincianfarini]

@JakeWharton It's likely you can close this issue. If you're up for it, I can write some documentation for how turbine should be used with hot flows.

[JakeWharton]

Yeah I left #19 open to track adding documentation and I think this falls under that category.