如何在 Android 上对 SharedFlow 进行单元测试

Question

我有一个具有以下属性的 android viewmodel 类

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

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

我正在使用 SharedFlow，因为它解决了 SingleLiveEvent 问题。

当我尝试对代码进行单元测试时，问题就出现了。我不知道如何使用涡轮机（或提供的基元）来使其工作。

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

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

我得到

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

我知道 SharedFlow 永远不会完成，这可能是部分原因，但我一直无法找到如何执行此操作的任何示例。

我正在使用 Junit 5 并使用 TestCoroutineDispatcher 类扩展。

Answer 1

在涡轮机的文档中有一个热流部分，我认为我们可以这样做：

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

几乎将扳机移动到涡轮.test {机块内