如何使用 delay() 调用测试 Kotlin 流程构建器？

Question

我在一个 Android Kotlin 项目中有一段代码类似于下面，我在其中使用 flow 构建器方法生成一个无限循环的周期性发射：

fun doSomething(): Flow<Int> = flow {
 var i = 0
  while (true) {
    emit(i++)
    delay(5000L)
  }
}

然后我尝试使用（非常有用！）Turbine 库对该流程进行单元测试，如下所示：

@Test
fun myTest() = runTest {
  doSomething().test {
    assertEquals(expected = 0, actual = awaitItem())
    assertEquals(expected = 1, actual = awaitItem())
    assertEquals(expected = 2, actual = awaitItem())
    cancelAndIgnoreRemainingEvents()
  }
}

我对 coroutines-test 库中的 runTest 方法的理解是，它会自动跳过该范围内对 delay 的任何调用，但上述测试因超时异常而失败。我试过在整个过程中乱扔日志语句，并且只触发了第一个 assertEquals 调用。后两者从未达到，因为从流程中调用 delay 显然阻止了测试。

这是使用 Flow 构建器时的预期行为吗？如果是这样，在我的场景中有什么方法可以控制时间的流逝吗？

Answer 1

看起来这是 Turbine 0.8.0 和 0.9.0-SNAPSHOT 中的错误。它不继承测试配置，可以跳过延迟。在此处查看问题：https://github.com/cashapp/turbine/issues/84

如果你想现在测试，那么你可以手动进行。

import kotlin.test.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.runTest

@OptIn(ExperimentalCoroutinesApi::class)
class DelayTest {

    fun doSomething(): Flow<Int> = flow {
        var i = 0
        while (true) {
            println("${System.currentTimeMillis()} emitting $i")
            emit(i++)
            delay(5000L)
        }
    }

    @Test
    fun testFlow() = runTest {
        val result = doSomething().take(3).toList()

        assertEquals(listOf(0, 1, 2), result)
    }
}

或者，作为解决方法，您可以使用 flowOn() 让流程使用测试调度程序。这会改变上下文 upstream 而不会“泄漏下游”。

    @Test
    fun testFlowOnWorkaround() = runTest {
        val testFlow = doSomething()
            .flowOn(UnconfinedTestDispatcher(testScheduler))

        testFlow.test {
            assertEquals(expected = 0, actual = awaitItem())
            assertEquals(expected = 1, actual = awaitItem())
            assertEquals(expected = 2, actual = awaitItem())
            cancelAndIgnoreRemainingEvents()
        }
    }

作为另一种解决方法，您可以先将流正常收集到列表中。感谢 runTest{} 延迟被跳过。然后您可以将其转换回流，并测试该流。

    @Test
    fun testFlowToListWorkaround() = runTest {
        val myFlow = doSomething().take(3).toList().asFlow()

        myFlow.test {
            assertEquals(expected = 0, actual = awaitItem())
            assertEquals(expected = 1, actual = awaitItem())
            assertEquals(expected = 2, actual = awaitItem())
            cancelAndIgnoreRemainingEvents()
        }
    }

版本：

科特林 1.6.21
协程 1.6.1
Kotlin 测试 1.6.21

如何使用 delay() 调用测试 Kotlin 流程构建器？

How can I test a Kotlin flow builder with delay() calls?

turbine

kotlin

kotlin-flow