在单元测试中验证 akka-stream 源

Question

在单元测试中验证 akka-stream 源的惯用方法是什么？我这样做：

f.service.fetchData(id).flatMap {
          case Right(source) => {
            // TODO: I need to validate here that source contains "Test value"
          }
          case Left(_) => fail("Wrongly responded with error")
        }

来源基本上是：

Source.single(ByteString("Test value"))

我尝试通过将 Source 连接到 Sink 并检查发出的值来执行，但断言似乎不起作用。

Answer 1

测试源、流和汇的正确方法是使用 test probes。

给定以下来源你想测试的逻辑

val source = Source(Seq("TestValue")).collect {
  case s @ "TestValue" => Right[String, String](s)
  case _               => Left[String, String]("error")
}

（我知道Left永远不会在这里触发，但这只是一个例子）

现在，您可以定义连接到给定源的 TestSink。结果图按以下方式执行

"assert correct value" in {
  implicit val system: ActorSystem = ???
  val probe = source.toMat(TestSink.probe)(Keep.right).run()
  probe.request(1)
  probe.expectNext(Right("TestValue"))
}

TestSink.probe 是一个具体化为 TestSubscriber.Probe[T] 的接收器，它提供对流的控制。由于它是一个接收器，它需要发出对元素的需求。它通过 request(1) 或请求一个元素来完成。然后是断言 sink.expectNext(Right("TestValue")) 检查是否收到了正确的值。

还有一个对应的 TestSource.probe 允许测试 Sink.

并且将两者结合起来，你可以测试一个Flow。

在单元测试中验证 akka-stream 源

Validating akka-stream Source in unit test

scala

unit-testing

akka-stream