如何单元测试MutableSharedFlow<T>(replay=0)是否发出了值？

Question

我不知道如何测试带有replay=0的SharedFlow是否发出了值。

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

您可以改用tryEmit()并验证返回的结果

更新：

考虑对ex使用Turbine：

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

}

Answer 3

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

@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 body才会立即执行，所以emit(1)不可能在first()之前执行。