Add tests for fs2 module and refactor streaming tests

1 files changed, 24 insertions, 61 deletions

diff --git a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
index 3238c3c..d3c7b89 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
@@ -1,18 +1,9 @@
 package com.softwaremill.sttp
 
-import java.nio.ByteBuffer
-
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
-import akka.stream.scaladsl.Source
-import akka.util.ByteString
-import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
-import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
-import com.softwaremill.sttp.okhttp.monix.OkHttpMonixClientHandler
+import com.softwaremill.sttp.streaming._
 import com.typesafe.scalalogging.StrictLogging
-import monix.execution.Scheduler.Implicits.global
-import monix.reactive.Observable
-import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
 import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
 
 import scala.language.higherKinds
@@ -21,16 +12,14 @@ class StreamingTests
     extends FlatSpec
     with Matchers
     with BeforeAndAfterAll
-    with ScalaFutures
     with StrictLogging
-    with IntegrationPatience
-    with ForceWrapped
-    with TestHttpServer {
+    with TestHttpServer
+    with ForceWrapped {
 
   override val serverRoutes: Route =
     path("echo") {
       post {
-        parameterMap { params =>
+        parameterMap { _ =>
           entity(as[String]) { body: String =>
             complete(body)
           }
@@ -38,50 +27,24 @@ class StreamingTests
       }
     }
 
-  type BodyProducer[S] = String => S
-  type BodyConsumer[S] = S => String
-
   override def port = 51824
+
   val body = "streaming test"
 
-  val akkaHandler = AkkaHttpSttpHandler.usingActorSystem(actorSystem)
-  val monixAsyncHttpClient = MonixAsyncHttpClientHandler()
-  val monixOkHttpClient = OkHttpMonixClientHandler()
-
-  val akkaHttpBodyProducer: BodyProducer[Source[ByteString, Any]] = s =>
-    Source.single(ByteString(s))
-  val akkaHttpBodyConsumer: BodyConsumer[Source[ByteString, Any]] =
-    _.runReduce(_ ++ _).futureValue.utf8String
-
-  val monixBodyProducer: BodyProducer[Observable[ByteBuffer]] =
-    s =>
-      Observable.fromIterable(
-        s.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
-
-  val monixBodyConsumer: BodyConsumer[Observable[ByteBuffer]] = stream =>
-    new String(stream
-                 .flatMap(bb => Observable.fromIterable(bb.array()))
-                 .toListL
-                 .runAsync
-                 .futureValue
-                 .toArray,
-               "utf-8")
-
-  runTests("Akka HTTP", akkaHttpBodyProducer, akkaHttpBodyConsumer)(
-    akkaHandler,
-    ForceWrappedValue.future)
-  runTests("Monix Async Http Client", monixBodyProducer, monixBodyConsumer)(
-    monixAsyncHttpClient,
-    ForceWrappedValue.monixTask)
-  runTests("Monix OkHttp Client", monixBodyProducer, monixBodyConsumer)(
-    monixOkHttpClient,
-    ForceWrappedValue.monixTask)
-
-  def runTests[R[_], S](name: String,
-                        bodyProducer: BodyProducer[S],
-                        bodyConsumer: BodyConsumer[S])(
-      implicit handler: SttpHandler[R, S],
-      forceResponse: ForceWrappedValue[R]): Unit = {
+  var closeHandlers: List[() => Unit] = Nil
+
+  runTests("Akka Http", new AkkaStreamingTests(actorSystem))
+  runTests("Monix Async Http Client", new MonixAHCStreamingTests)
+  runTests("Monix OkHttp", new MonixOKHStreamingTests)
+  runTests("fs2 Async Http Client", new Fs2StreamingTests)
+
+  def runTests[R[_], S](
+      name: String,
+      testStreamingHandler: TestStreamingHandler[R, S]): Unit = {
+    import testStreamingHandler._
+
+    closeHandlers = handler.close _ :: closeHandlers
+
     name should "stream request body" in {
       val response = sttp
         .post(uri"$endpoint/echo")
@@ -89,7 +52,7 @@ class StreamingTests
         .send()
         .force()
 
-      response.body should be(body)
+      response.body shouldBe body
     }
 
     it should "receive a stream" in {
@@ -100,7 +63,7 @@ class StreamingTests
         .send()
         .force()
 
-      bodyConsumer(response.body) should be(body)
+      bodyConsumer(response.body).force() shouldBe body
     }
 
     it should "receive a stream from an https site" in {
@@ -113,13 +76,13 @@ class StreamingTests
         .send()
         .force()
 
-      bodyConsumer(response.body) should include("</div>")
+      bodyConsumer(response.body).force() should include("</div>")
     }
   }
 
   override protected def afterAll(): Unit = {
-    akkaHandler.close()
-    monixAsyncHttpClient.close()
+    closeHandlers.foreach(_())
     super.afterAll()
   }
+
 }