使用SttpBackendStub对流请求进行单元测试。

Question

我有一个请求，返回一个我想测试的Source流，就像这样。

/* class */
import sttp.client._

class HttpClient()
                (implicit sttpBackend: SttpBackend[Future, Source[ByteString, Any], Nothing]) {
  /** Start a download stream for large files */
  def downloadStream(url: String): Future[Response[Either[String, Source[ByteString, _]]]] = {
    request
      .get(uri"url")
      .response(asStream[Source[ByteString, Any]])
      .send()
  }
}

/* test */
import org.scalatest.FlatSpec
import org.scalatest.Matchers

class HttpClientSpec extends FlatSpec with Matchers {
  implicit val as = ActorSystem()
  implicit val mat = ActorMaterializer()
  implicit val ex = as

  "downloadStream" should "get a file stream from url" in {
    val url = s"http://example.com/api/v1/file/Test.txt"
    val body = "abcdefghijklmnopqrstuzwxyz"
    val expectedStream = Source.single[ByteString](body.getBytes)

    implicit val stubBackend = SttpBackendStub.asynchronousFuture
      .whenRequestMatches(req => req.method.method == "GET" && req.uri == uri"$url")
      .thenRespond(
        Response(Right(expectedStream), StatusCode.Ok)
      )

    val httpTest = new HttpClient()

    val response = Await.result(httpTest.downloadStream(url), 5.seconds)
    response.body.right.get should be(expectedStream)
  }
}

但是它失败了，因为 SttpBackendStub.asynchronousFuture 不处理流响应.

type mismatch;
[error]  found   : sttp.client.testing.SttpBackendStub[scala.concurrent.Future,Nothing]
[error]  required: sttp.client.SttpBackend[scala.concurrent.Future,akka.stream.scaladsl.Source[akka.util.ByteString,Any],Nothing]

我如何使用SttpBackendStub，而不需要求助于类似于 scalamock?

Answer 1

只需要创建你自己的Stub类，就可以了 extends 预期的响应类型。

class StreamHttpStub
    extends SttpBackendStub[Future, Source[ByteString, Any]](new FutureMonad(), PartialFunction.empty, None) {
}

implicit val stubBackend = new StreamHttpStub()
  .whenRequestMatches(req => req.method.method == "GET" && req.uri == uri"$url")
  .thenRespond(
    Response(Right(expectedStream), StatusCode.Ok)
  )