Add tests for fs2 module and refactor streaming tests

9 files changed, 211 insertions, 106 deletions

diff --git a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
index aeee5ed..1594ef3 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
@@ -6,18 +6,15 @@ import java.nio.file.Paths
 import java.time.{ZoneId, ZonedDateTime}
 
 import akka.http.scaladsl.coding.{Deflate, Gzip, NoCoding}
-import akka.http.scaladsl.model.{DateTime, FormData}
-import akka.http.scaladsl.model.headers._
 import akka.http.scaladsl.model.headers.CacheDirectives._
+import akka.http.scaladsl.model.headers._
+import akka.http.scaladsl.model.{DateTime, FormData}
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
 import akka.http.scaladsl.server.directives.Credentials
 import akka.util.ByteString
-import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
-import com.typesafe.scalalogging.StrictLogging
-import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
-import org.scalatest.{path => _, _}
 import better.files._
+import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
 import com.softwaremill.sttp.asynchttpclient.cats.CatsAsyncHttpClientHandler
 import com.softwaremill.sttp.asynchttpclient.future.FutureAsyncHttpClientHandler
 import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
@@ -27,8 +24,11 @@ import com.softwaremill.sttp.okhttp.{
   OkHttpFutureClientHandler,
   OkHttpSyncClientHandler
 }
-import scala.concurrent.ExecutionContext.Implicits.global
+import com.typesafe.scalalogging.StrictLogging
+import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
+import org.scalatest.{path => _, _}
 
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.language.higherKinds
 
 class BasicTests
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()
   }
+
 }
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala
new file mode 100644
index 0000000..53fe63e
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala
@@ -0,0 +1,29 @@
+package com.softwaremill.sttp.streaming
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.stream.Materializer
+import akka.stream.scaladsl.Source
+import akka.util.ByteString
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
+
+import scala.concurrent.Future
+
+class AkkaStreamingTests(actorSystem: ActorSystem)(
+    implicit materializer: Materializer)
+    extends TestStreamingHandler[Future, Source[ByteString, Any]] {
+
+  override implicit val handler: SttpHandler[Future, Source[ByteString, Any]] =
+    AkkaHttpSttpHandler.usingActorSystem(actorSystem)
+
+  override implicit val forceResponse: ForceWrappedValue[Future] =
+    ForceWrappedValue.future
+
+  override def bodyProducer(body: String): Source[ByteString, NotUsed] =
+    Source.single(ByteString(body))
+
+  override def bodyConsumer(stream: Source[ByteString, Any]): Future[String] =
+    stream.map(_.utf8String).runReduce(_ + _)
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala
new file mode 100644
index 0000000..b9d249b
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala
@@ -0,0 +1,29 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import cats.effect._
+import cats.implicits._
+import com.softwaremill.sttp.asynchttpclient.fs2.Fs2AsyncHttpClientHandler
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+import fs2._
+
+class Fs2StreamingTests
+    extends TestStreamingHandler[IO, Stream[IO, ByteBuffer]] {
+
+  override implicit val handler: SttpHandler[IO, Stream[IO, ByteBuffer]] =
+    Fs2AsyncHttpClientHandler[IO]()
+
+  override implicit val forceResponse: ForceWrappedValue[IO] =
+    ForceWrappedValue.catsIo
+
+  override def bodyProducer(body: String): Stream[IO, ByteBuffer] =
+    Stream.emits(body.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
+
+  override def bodyConsumer(stream: Stream[IO, ByteBuffer]): IO[String] =
+    stream
+      .map(bb => Chunk.array(bb.array))
+      .through(text.utf8DecodeC)
+      .runFoldMonoid
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala
new file mode 100644
index 0000000..4a4ff96
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala
@@ -0,0 +1,17 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.SttpHandler
+import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
+import monix.eval.Task
+import monix.reactive.Observable
+
+class MonixAHCStreamingTests extends MonixBaseHandler {
+
+  import monix.execution.Scheduler.Implicits.global
+
+  override implicit val handler: SttpHandler[Task, Observable[ByteBuffer]] =
+    MonixAsyncHttpClientHandler()
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala
new file mode 100644
index 0000000..b34a4ae
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala
@@ -0,0 +1,25 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.ForceWrappedValue
+import monix.eval.Task
+import monix.reactive.Observable
+
+trait MonixBaseHandler
+    extends TestStreamingHandler[Task, Observable[ByteBuffer]] {
+
+  override implicit def forceResponse: ForceWrappedValue[Task] =
+    ForceWrappedValue.monixTask
+
+  override def bodyProducer(body: String): Observable[ByteBuffer] =
+    Observable.fromIterable(
+      body.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
+
+  override def bodyConsumer(stream: Observable[ByteBuffer]): Task[String] =
+    stream
+      .flatMap(bb => Observable.fromIterable(bb.array()))
+      .toListL
+      .map(bs => new String(bs.toArray, "utf8"))
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala
new file mode 100644
index 0000000..04666be
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala
@@ -0,0 +1,17 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.SttpHandler
+import com.softwaremill.sttp.okhttp.monix.OkHttpMonixClientHandler
+import monix.eval.Task
+import monix.reactive.Observable
+
+class MonixOKHStreamingTests extends MonixBaseHandler {
+
+  import monix.execution.Scheduler.Implicits.global
+
+  override implicit val handler: SttpHandler[Task, Observable[ByteBuffer]] =
+    OkHttpMonixClientHandler()
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala
new file mode 100644
index 0000000..b786e49
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala
@@ -0,0 +1,15 @@
+package com.softwaremill.sttp.streaming
+
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+
+import scala.language.higherKinds
+
+trait TestStreamingHandler[R[_], S] {
+  implicit def handler: SttpHandler[R, S]
+
+  implicit def forceResponse: ForceWrappedValue[R]
+
+  def bodyProducer(body: String): S
+
+  def bodyConsumer(stream: S): R[String]
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
index 59e1612..407aaca 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
@@ -7,12 +7,13 @@ import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
-import org.scalatest.{BeforeAndAfterAll, Suite}
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.exceptions.TestFailedException
 import org.scalatest.matchers.{MatchResult, Matcher}
+import org.scalatest.{BeforeAndAfterAll, Suite}
 
 import scala.concurrent.Future
+import scala.concurrent.duration._
 import scala.language.higherKinds
 import scalaz._
 
@@ -36,45 +37,54 @@ trait TestHttpServer extends BeforeAndAfterAll with ScalaFutures {
   def port: Int
 }
 
-trait ForceWrapped extends ScalaFutures { this: Suite =>
-  trait ForceWrappedValue[R[_]] {
-    def force[T](wrapped: R[T]): T
+trait ForceWrappedValue[R[_]] {
+  def force[T](wrapped: R[T]): T
+}
+
+object ForceWrappedValue extends ScalaFutures {
+  override implicit val patienceConfig: PatienceConfig =
+    PatienceConfig(timeout = 5.seconds, interval = 150.milliseconds)
+
+  val id = new ForceWrappedValue[Id] {
+    override def force[T](wrapped: Id[T]): T =
+      wrapped
   }
-  object ForceWrappedValue {
-    val id = new ForceWrappedValue[Id] {
-      override def force[T](wrapped: Id[T]): T =
-        wrapped
-    }
-    val future = new ForceWrappedValue[Future] {
-      override def force[T](wrapped: Future[T]): T =
-        try {
-          wrapped.futureValue
-        } catch {
-          case e: TestFailedException if e.getCause != null => throw e.getCause
-        }
-    }
-    val scalazTask = new ForceWrappedValue[scalaz.concurrent.Task] {
-      override def force[T](wrapped: scalaz.concurrent.Task[T]): T =
-        wrapped.unsafePerformSyncAttempt match {
-          case -\/(error) => throw error
-          case \/-(value) => value
-        }
-    }
-    val monixTask = new ForceWrappedValue[monix.eval.Task] {
-      import monix.execution.Scheduler.Implicits.global
-
-      override def force[T](wrapped: monix.eval.Task[T]): T =
-        try {
-          wrapped.runAsync.futureValue
-        } catch {
-          case e: TestFailedException => throw e.getCause
-        }
-    }
-    val catsIo = new ForceWrappedValue[cats.effect.IO] {
-      override def force[T](wrapped: cats.effect.IO[T]): T =
-        wrapped.unsafeRunSync
-    }
+  val future = new ForceWrappedValue[Future] {
+
+    override def force[T](wrapped: Future[T]): T =
+      try {
+        wrapped.futureValue
+      } catch {
+        case e: TestFailedException if e.getCause != null => throw e.getCause
+      }
+  }
+  val scalazTask = new ForceWrappedValue[scalaz.concurrent.Task] {
+    override def force[T](wrapped: scalaz.concurrent.Task[T]): T =
+      wrapped.unsafePerformSyncAttempt match {
+        case -\/(error) => throw error
+        case \/-(value) => value
+      }
+  }
+  val monixTask = new ForceWrappedValue[monix.eval.Task] {
+    import monix.execution.Scheduler.Implicits.global
+
+    override def force[T](wrapped: monix.eval.Task[T]): T =
+      try {
+        wrapped.runAsync.futureValue
+      } catch {
+        case e: TestFailedException => throw e.getCause
+      }
   }
+  val catsIo = new ForceWrappedValue[cats.effect.IO] {
+    override def force[T](wrapped: cats.effect.IO[T]): T =
+      wrapped.unsafeRunSync
+  }
+}
+
+trait ForceWrapped extends ScalaFutures { this: Suite =>
+  type ForceWrappedValue[R[_]] = com.softwaremill.sttp.ForceWrappedValue[R]
+  val ForceWrappedValue: com.softwaremill.sttp.ForceWrappedValue.type =
+    com.softwaremill.sttp.ForceWrappedValue
 
   implicit class ForceDecorator[R[_], T](wrapped: R[T]) {
     def force()(implicit fwv: ForceWrappedValue[R]): T = fwv.force(wrapped)