How to real test the akka stream?

Question

I am using WebsocketClient and would like to test against the received message. I've chosen the Scalatest framework and I know, that the test has be carry out asynchronously.

The websocket client looks as the following:

import akka.{Done}
import akka.http.scaladsl.Http
import akka.stream.scaladsl._
import akka.http.scaladsl.model.ws._
import io.circe.syntax._

import scala.concurrent.Future


object WsClient {

  import Trigger._

  private val convertJson: PreMsg => String = msg =>
    msg.asJson.noSpaces

  val send: PreMsg => (String => Unit) => RunnableGraph[Future[Done]] = msg => fn =>
    Source.single(convertJson(msg))
      .map(TextMessage(_))
      .via(Http().webSocketClientFlow(WebSocketRequest(s"ws://{$Config.host}:{$Config.port}/saprs")))
      .map(_.asTextMessage.getStrictText)
      .toMat(Sink.foreach(fn))(Keep.right)


}  

and the test:

  feature("Process incoming messages") {
    info("As a user, I want that incoming messages is going to process appropriately.")
    info("A message should contain the following properties: `sap_id`, `sap_event`, `payload`")

    scenario("Message is not intended for the server") {
      Given("A message with `sap_id:unknown`")
      val msg = PreMsg("unknown", "unvalid", "{}")
      When("the message gets validated")
      val ws = WsClient.send(msg)
      Then("it should has the `status: REJECT` in the response content")
      ws { msg =>
        //Would like test against the msg here
      }.run()
        .map(_ => assert(1 == 1))

    }  

I would to test against the content of msg, but I do not know, how to do it.

Answer 1

I followed the play-scala-websocket-example

They use a WebSocketClient as a helper, see WebSocketClient.java

Then a test looks like:

Helpers.running(TestServer(port, app)) {
    val myPublicAddress = s"localhost:$port"
    val serverURL = s"ws://$myPublicAddress/ws"

    val asyncHttpClient: AsyncHttpClient = client.underlying[AsyncHttpClient]
    val webSocketClient = new WebSocketClient(asyncHttpClient)
    val queue = new ArrayBlockingQueue[String](10)
    val origin = serverURL
    val consumer: Consumer[String] = new Consumer[String] {
      override def accept(message: String): Unit = queue.put(message)
    }
    val listener = new WebSocketClient.LoggingListener(consumer)
    val completionStage = webSocketClient.call(serverURL, origin, listener)
    val f = FutureConverters.toScala(completionStage)

    // Test we can get good output from the websocket
    whenReady(f, timeout = Timeout(1.second)) { webSocket =>
      val condition: Callable[java.lang.Boolean] = new Callable[java.lang.Boolean] {
        override def call(): java.lang.Boolean = webSocket.isOpen && queue.peek() != null
      }
      await().until(condition)
      val input: String = queue.take()
      val json:JsValue = Json.parse(input)
      val symbol = (json \ "symbol").as[String]
      List(symbol) must contain oneOf("AAPL", "GOOG", "ORCL")
    }
  }
}

See here: FunctionalSpec.scala