使用akka http的Web套接字的单元测试用例

Posted 2021-04-03

我使用Akka HTTP实现了Web套接字。我正在使用Kafka的数据并使用Web套接字发送通知。功能正常，但我陷入了测试用例。

 ` private def loginNotificationRoute(): Route = {
    cors() {
      pathPrefix("notifications" / Segment) { userName =>
        get {
          handleWebSocketMessages(notificationClientFlow(userName))
        }
      }
    }
  }`

通知Actor从数据库读取通知并将数据发送回客户端。

`  def notificationClientFlow(userName: String): Flow[Message, Message, NotUsed] = {
    info(s"Connection Request accepted for user $userName")
    val notificationActor = actorSystem.actorOf(NotificationActor.props(userName, jsonHelper))

    val incomingMessages: Sink[Message, NotUsed] =
      Flow[Message].map {
        case TextMessage.Strict(text) => NotificationActor.IncomingMessage(text)
      }.to(Sink.actorRef(notificationActor, PoisonPill))

    val outgoingMessages: Source[Message, NotUsed] =
      Source
        .actorRef[NotificationActor.ResponseType](BUFFER_SIZE, OverflowStrategy.fail)
        .mapMaterializedValue { outgoingActor =>
          notificationActor ! NotificationActor.Connected(outgoingActor)
          NotUsed
        }
        .map {
          notificationResponse: NotificationActor.ResponseType =>
            info(s"sending notification ${notificationResponse.action}")
            TextMessage.Strict(jsonHelper.write(notificationResponse))
        }

    Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
  }`

答案

重新设计依赖关系

在我看来，你当前设计的最大缺点是对特定的actorSystem存在固有的依赖性;令人讨厌的代码行

//actorSystem comes from the outside world!!!
val notificationActor = actorSystem.actorOf(NotificationActor.props(userName, jsonHelper))

应重新参数化notificationClientFlow方法，以便在参数中明确列出所有依赖项。并且不应该依赖整个ActorSystem，而应该将其简化为ActorRef：

def notificationClientFlow(userName: String, 
                           notificationActor : ActorRef): Flow[Message, Message, NotUsed] = {
  //notificationActor is now passed in
  //val notificationActor = actorSystem.actorOf(NotificationActor.props(userName, jsonHelper))
}

这需要在Route创建方法中使用类似的显式声明：

private def loginNotificationRoute(notificationActor : ActorRef)() : Route = {
  ...
  handleWebSocketMessages(notificationClientFlow(userName, notificationActor))
}

测试

现在可以使用TestKit实现测试：

class MySpec() extends TestKit(ActorSystem("MySpec")) {

  val userName = "testUser"

  val notificationActor = 
    system.actorOf(NotificationActor.props(userName, jsonHelper))

  val testFlow = notificationClientFlow(userName, notificationActor)

}

另外，因为我们已经明确地传递了ActorRef而不仅仅是ActorSystem，我们可以使用其他高级测试功能，如probes：

val probe = TestProbe()

val testProbeFlow = notificationClientFlow(userName, probe.ref)

上述技术同样适用于使用standard techniques测试路线：

val testRoute = loginNotificationRoute(testProbe)

Get() ~> testRoute() ~> check {
  //testing assertions here
}