Scala integration tests for Caliban GraphQL subscriptions

Question

We have a Caliban GraphQL application, using it with Play framework. It is well covered with integration tests for queries and mutations, now we're about to add some integration tests for subscriptions and wondering how to do it correctly.

For queries/mutations testing we're using usual FakeRequest, sending it to our router that extends Caliban's PlayRouter, it works very good. Is there any similar way to test websockets/subscriptions?

There is very short amount of information in the Internet about websocket testing in Play and no information at all about GraphQL subscription testing.

Will be grateful for any ideas!

Answer 1

Ok, I managed it. There are couple of rules to follow:

1. Use websocket header "WebSocket-Protocol" -> "graphql-ws"
2. After connection is established, send GraphQLWSRequest of type "connection_init"
3. After receiving response "connection_ack", send GraphQLWSRequest of type "start" with subscription query as payload

After those steps server is listening and you can send your mutation queries.

Some draft example:

import caliban.client.GraphQLRequest
import caliban.client.ws.GraphQLWSRequest
import io.circe.syntax.EncoderOps
import play.api.libs.json.{JsValue, Json}
import play.api.test.Helpers.{POST, contentAsJson, contentAsString, contentType, route, status, _}
import org.awaitility.Awaitility

 def getWS(subscriptionQuery: String, postQuery: String): JsValue = {
    lazy val port    = Helpers.testServerPort
    val initRequest  = prepareWSRequest("connection_init")
    val startRequest = prepareWSRequest("start", Some(GraphQLRequest(subscriptionQuery, Map())))

    Helpers.running(TestServer(port, app)) {

      val headers = new java.util.HashMap[String, String]()
      headers.put("WebSocket-Protocol", "graphql-ws")

      val queue = new ArrayBlockingQueue[String](1)

      lazy val ws = new WebSocketClient(new URI(s"ws://localhost:$port/ws/graphql"), headers) {
        override def onOpen(handshakedata: ServerHandshake): Unit =
          logger.info("Websocket connection established")

        override def onClose(code: Port, reason: String, remote: Boolean): Unit =
          logger.info(s"Websocket connection closed, reason: $reason")

        override def onError(ex: Exception): Unit =
          logger.error("Error handling websocket connection", ex)

        override def onMessage(message: String): Unit = {
          val ttp = (Json.parse(message) \ "type").as[JsString].value
          if (ttp != "connection_ack" && ttp != "ka") queue.put(message)
        }
      }

      ws.connectBlocking()

      Future(ws.send(initRequest))
        .flatMap(_ => Future(ws.send(startRequest)))
        .flatMap(_ => post(query = postQuery)) // post is my local method, it sends usual FakeRequest
      
      Awaitility.await().until(() => queue.peek() != null)
      Json.parse(queue.take())
    }

  def prepareWSRequest(ttp: String, payload: Option[GraphQLRequest] = None) =
    GraphQLWSRequest(ttp, None, payload).asJson.noSpaces
  }