How to unit test SharedFlow on Android

Question

I have an android viewmodel 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.

Answer 1

In the Turbine's documentation there is a Hot Flows section, I think we could do something like this:

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

pretty much moving the trigger inside of the Turbine's .test { block