单元测试回调流程

Question

我有一个基于回调的 API，如下所示：

  class CallbackApi {
    fun addListener(callback: Callback) {
      // todo
    }

    fun removeListener(callback: Callback) {
      // todo
    }

    interface Callback {
      fun onResult(result: Int)
    }
  }

以及将 API 转换为冷热流的扩展函数：

  fun CallbackApi.toFlow() = callbackFlow<Int> {
    val callback = object : CallbackApi.Callback {
      override fun onResult(result: Int) {
        trySendBlocking(result)
      }
    }
    addListener(callback)
    awaitClose { removeListener(callback) }
  }

您介意建议如何编写一个单元测试来确保 API 正确转换为热流吗？

这是我的尝试。经过反复试验，我想出了这个解决方案。

  @Test
  fun callbackFlowTest() = runBlocking {
    val callbackApi = mockk<CallbackApi>()
    val callbackSlot = slot<CallbackApi.Callback>()
    every { callbackApi.addListener(capture(callbackSlot)) } just Runs
    every { callbackApi.removeListener(any()) } just Runs
    val list = mutableListOf<Int>()
    val flow: Flow<Int> = callbackApi.toFlow().onEach { list.add(it) }
    val coroutineScope = CoroutineScope(this.coroutineContext + SupervisorJob())
    flow.launchIn(coroutineScope)
    yield()
    launch {
      callbackSlot.captured.onResult(10)
      callbackApi.removeListener(mockk()) // this was a misunderstanding
    }.join()
    assert(list.single() == 10)
  }

但我不明白这个解决方案的两部分。

1- 如果没有这个SupervisorJob()，测试似乎永远不会结束。也许由于某种原因，收集流量永远不会结束，我不明白。我在一个单独的协程中提供捕获的回调。

2-如果我移除其中的launch主体callbackSlot.captured.onResult(10)，测试将因此错误而失败UninitializedPropertyAccessException: lateinit property captured has not been initialized。我认为这yield应该开始流程。

Answer 1

以及将 API 转换为热流的扩展函数

这个扩展看起来是正确的，但是流程并不热（也不应该是）。它仅在实际收集开始时注册回调，并在取消收集器时取消注册（这包括当收集器使用限制项目数量的终端运算符时，例如.first()或.take(n)）。

对于您的其他问题，这是一个非常重要的注意事项。

如果没有 SupervisorJob()，测试似乎永远不会结束。也许由于某种原因收集流量永远不会结束，我不明白

如上所述，由于流的构造方式（以及工作方式CallbackApi），流收集不能由生产者（回调 API）决定结束。它只能通过取消收集器来停止，这也将取消注册相应的回调（这很好）。

您的自定义作业允许测试结束的原因可能是因为您通过不以当前作业作为父项的自定义作业覆盖上下文中的作业来逃避结构化并发。但是，您可能仍然会从永不取消的作用域中泄漏永无休止的协程。

我在一个单独的协程中提供捕获的回调。

这是正确的，尽管我不明白为什么你removeListener从这个单独的协程中调用。您在这里取消注册什么回调？请注意，这也不会对流程产生任何影响，因为即使您可以注销在构建callbackFlow器中创建的回调，它也不会神奇地关闭 的通道callbackFlow，因此流程无论如何都不会结束（我'我假设这就是您在这里尝试做的事情）。

此外，从外部取消注册回调会阻止您检查它实际上是否已被生产代码取消注册。

2-如果我删除其中包含callbackSlot.captured.onResult(10)的启动主体，测试将失败并出现此错误UninitializedPropertyAccessException：捕获的lateinit属性尚未初始化。我认为产量应该开始流动。

yield()很脆。如果你使用它，你必须非常清楚当前每个并发协程的代码是如何编写的。它脆弱的原因是它只会将线程让给其他协程，直到下一个挂起点。你无法预测在让出时将执行哪一个协程，也无法预测到达挂起点后线程将恢复哪一个。如果有几次停赛，那么所有的赌注都会被取消。如果还有其他正在运行的协程，那么所有的赌注也会被取消。

更好的方法是使用kotlinx-coroutines-testwhich 提供的实用程序，例如advanceUntilIdle确保其他协程全部完成或等待挂起点。

现在如何修复这个测试？我现在无法测试任何东西，但我可能会这样处理：

• 使用runTestfromkotlinx-coroutines-test而不是runBlocking可以更好地控制其他协程运行时（并等待流集合执行某些操作）
• 在协程中启动流集合（仅launch/launchIn(this)没有自定义范围）并保留已启动的句柄Job（返回值launch/ launchIn）
• 用一个值调用捕获的回调，advanceUntilIdle()以确保流收集器的协程可以处理它，然后断言列表已获取元素（注意：由于一切都是单线程并且回调不会挂起，如果没有缓冲区，这将死锁，但callbackFlow使用默认缓冲区，所以应该没问题）
• 可选：使用不同的值重复上述几次并确认它们被流收集
• 取消收集作业，advanceUntilIdle()然后测试回调是否未注册（我不是Mockk专家，但应该有一些东西可以检查被removeListener调用）

注意：也许我是老派，但如果你CallbackApi是一个接口（在你的例子中它是一个类，但我不确定它在多大程度上反映了现实），我宁愿使用通道来模拟手动实现模拟事件并断言期望。我发现推理和调试更容易。这是我的意思的一个例子