如何对 MutableSharedFlow<T>(replay=0) 发出值进行单元测试？

Question

我无法弄清楚如何SharedFlow使用replay=0发出的值来测试a 。

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.runBlocking
import org.junit.Test

class ShowcaseTest {

    @Test
    fun testIntSharedFlowFlow() {
        val intSharedFlow = MutableSharedFlow<Int>()

        runBlocking {
            intSharedFlow.emit(1)
        }
        
        // Does not work as there is no buffer because MutableSharedFlow(replay=0)
        assert(intSharedFlow.replayCache.first() == 1)
    }
}

Answer 1

如果你想用 replay=1 进行测试，你可以尝试在“观察/收集”之前发出，也就是在作业开始之前发出。

@Test
fun testIntSharedFlowFlow() = runBlockingTest{
    val _intSharedFlow = MutableSharedFlow<Int>()
    val intSharedFlow: SharedFlow = _intSharedFlow
    val testResults = mutableListOf<Int>()

    val job = launch {
        intSharedFlow.toList(testResults)
    }
    _intSharedFlow.emit(5) 
    
    assertEquals(1, testResults.size)
    assertEquals(5, testResults.first())
    job.cancel()
}

不要忘记取消作业，否则sharedFlow将继续收集，测试会给你一个错误，甚至永远循环。

Answer 2

一种解决方案是使用Deferred返回的对象async

@Test
fun `test shared flow with deferred`() = runBlockingTest {
    val sharedFlow = MutableSharedFlow<Int>(replay = 0)

    val deferred = async {
        sharedFlow.first()
    }

    sharedFlow.emit(1)
   
    assertEquals(1, deferred.await())
}

注意：您必须使用runBlockingTest这样的async主体立即执行，这样就没有机会emit(1)在之前执行first()。

Answer 3

您可以使用tryEmit()代替并验证返回的结果

更新：

考虑使用Turbine For ex：

sharedFlow.test {
    sharedFlow.emit(1)
    assertEquals(expected = 1, expectItem())
    cancelAndIgnoreRemainingEvents()

}

Answer 4

runBlockingTest 自 1.6.0 起已弃用，请改用 runTest。迁移链接

@Test
fun `test shared flow with deferred`() = runTest(UnconfinedTestDispatcher()) {
    val sharedFlow = MutableSharedFlow<Int>(replay = 0)

    val deferred = async {
        sharedFlow.first()
    }

    sharedFlow.emit(1)
   
    assertEquals(1, deferred.await())
}