<commit_before>package sttp.client3.testing.streaming

import org.scalatest.BeforeAndAfterAll
import org.scalatest.freespec.AsyncFreeSpec
import org.scalatest.matchers.should.Matchers
import sttp.capabilities.{Effect, Streams}
import sttp.client3._
import sttp.client3.internal.Utf8
import sttp.client3.testing.HttpTest.endpoint
import sttp.client3.testing.streaming.StreamingTest._
import sttp.client3.testing.{ConvertToFuture, ToFutureWrapper}
import sttp.model.sse.ServerSentEvent

abstract class StreamingTest[F[_], S]
    extends AsyncFreeSpec
    with Matchers
    with ToFutureWrapper
    with BeforeAndAfterAll
    with StreamingTestExtensions[F, S] {

  val streams: Streams[S]

  def backend: SttpBackend[F, S]

  implicit def convertToFuture: ConvertToFuture[F]

  def bodyProducer(chunks: Iterable[Array[Byte]]): streams.BinaryStream

  private def stringBodyProducer(body: String): streams.BinaryStream =
    bodyProducer(body.getBytes(Utf8).grouped(10).toIterable)

  def bodyConsumer(stream: streams.BinaryStream): F[String]

  def sseConsumer(stream: streams.BinaryStream): F[List[ServerSentEvent]]

  protected def supportsStreamingMultipartParts = true

  "stream request body" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .streamBody(streams)(stringBodyProducer(Body))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Right(Body)
      }
  }

  "stream large request body" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .streamBody(streams)(stringBodyProducer(Body))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Right(Body)
      }
  }

  "handle server sent events SSE" in {
    val sseData = "ala ma kota\nzbyszek ma psa"
    val expectedEvent = ServerSentEvent(data = Some(sseData), eventType = Some("test-event"), retry = Some(42000))
    val expectedEvents =
      Seq(expectedEvent.copy(id = Some("1")), expectedEvent.copy(id = Some("2")), expectedEvent.copy(id = Some("3")))

    basicRequest
      .post(uri"$endpoint/sse/echo3")
      .body(sseData)
      .response(asStreamAlways(streams)(sseConsumer(_)))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe expectedEvents
      }
  }

  "receive a stream" in {
    // TODO: for some reason these explicit types are needed in Dotty
    val r0: RequestT[Identity, String, Effect[F] with S] = basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      .response(asStreamAlways(streams)(bodyConsumer(_)))
    r0.send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Body
      }
  }

  "receive a stream and ignore it (without consuming)" in {
    // TODO: for some reason these explicit types are needed in Dotty
    val r0: RequestT[Identity, String, Effect[F] with S] = basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      // if the backend has any, mechanisms to consume an incorrectly handled (ignored) stream should kick in
      .response(asStreamAlways(streams)(_ => bodyConsumer(stringBodyProducer("ignore"))))
    r0.send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe "ignore"
      }
  }

  "receive a stream (unsafe)" in {
    // TODO: for some reason these explicit types are needed in Dotty
    val r0: RequestT[Identity, streams.BinaryStream, S] = basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      .response(asStreamAlwaysUnsafe(streams))
    r0.send(backend)
      .toFuture()
      .flatMap { response =>
        bodyConsumer(response.body).toFuture()
      }<commit_msg>Simplify test definition<commit_after>package sttp.client3.testing.streaming

import org.scalatest.BeforeAndAfterAll
import org.scalatest.freespec.AsyncFreeSpec
import org.scalatest.matchers.should.Matchers
import sttp.capabilities.Streams
import sttp.client3._
import sttp.client3.internal.Utf8
import sttp.client3.testing.HttpTest.endpoint
import sttp.client3.testing.streaming.StreamingTest._
import sttp.client3.testing.{ConvertToFuture, ToFutureWrapper}
import sttp.model.sse.ServerSentEvent

abstract class StreamingTest[F[_], S]
    extends AsyncFreeSpec
    with Matchers
    with ToFutureWrapper
    with BeforeAndAfterAll
    with StreamingTestExtensions[F, S] {

  val streams: Streams[S]

  def backend: SttpBackend[F, S]

  implicit def convertToFuture: ConvertToFuture[F]

  def bodyProducer(chunks: Iterable[Array[Byte]]): streams.BinaryStream

  private def stringBodyProducer(body: String): streams.BinaryStream =
    bodyProducer(body.getBytes(Utf8).grouped(10).toIterable)

  def bodyConsumer(stream: streams.BinaryStream): F[String]

  def sseConsumer(stream: streams.BinaryStream): F[List[ServerSentEvent]]

  protected def supportsStreamingMultipartParts = true

  "stream request body" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .streamBody(streams)(stringBodyProducer(Body))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Right(Body)
      }
  }

  "stream large request body" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .streamBody(streams)(stringBodyProducer(Body))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Right(Body)
      }
  }

  "handle server sent events SSE" in {
    val sseData = "ala ma kota\nzbyszek ma psa"
    val expectedEvent = ServerSentEvent(data = Some(sseData), eventType = Some("test-event"), retry = Some(42000))
    val expectedEvents =
      Seq(expectedEvent.copy(id = Some("1")), expectedEvent.copy(id = Some("2")), expectedEvent.copy(id = Some("3")))

    basicRequest
      .post(uri"$endpoint/sse/echo3")
      .body(sseData)
      .response(asStreamAlways(streams)(sseConsumer(_)))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe expectedEvents
      }
  }

  "receive a stream" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      .response(asStreamAlways(streams)(bodyConsumer(_)))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe Body
      }
  }

  "receive a stream and ignore it (without consuming)" in {
    basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      // if the backend has any, mechanisms to consume an incorrectly handled (ignored) stream should kick in
      .response(asStreamAlways(streams)(_ => bodyConsumer(stringBodyProducer("ignore"))))
      .send(backend)
      .toFuture()
      .map { response =>
        response.body shouldBe "ignore"
      }
  }

  "receive a stream (unsafe)" in {
    // TODO: for some reason these explicit types are needed in Dotty
    val r0: RequestT[Identity, streams.BinaryStream, S] = basicRequest
      .post(uri"$endpoint/streaming/echo")
      .body(Body)
      .response(asStreamAlwaysUnsafe(streams))
    r0.send(backend)
      .toFuture()
      .flatMap { response =>
        bodyConsumer(response.body).toFuture()
      }
{'fraction_non_alphanumeric': 0.1116701607267645, 'fraction_numerical': 0.006429070580013976, 'mean_word_length': 3.4009840098400983, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '17543', 'n_tokens_mistral': 2241, 'n_tokens_neox': 2179, 'n_words': 483}